ApsaraMQ for Kafka:Migrasi data menggunakan tool

Prasyarat

Sebelum memulai, pastikan Anda telah:

• Menginstal JDK 8 atau 11 di server kluster Apache Kafka mandiri Anda. Untuk tautan unduhan, lihat Java Downloads.

• Mengunduh tool migrasi kafka-migration-assessment.jar ke server.

Migrasi topik

Migrasi topik membaca metadata topik dari ZooKeeper sumber dan membuat topik yang sesuai di instans ApsaraMQ for Kafka tujuan. Proses ini terdiri dari dua fase: pemeriksaan awal (dry run) dan commit (migrasi aktual).

Fase 1: Pemeriksaan Awal

Pratinjau topik yang akan dimigrasikan:

java -jar kafka-migration-assessment.jar TopicMigrationFromZk  \
  --sourceZkConnect 192.168.XX.XX  \
  --destAk <your-access-key-id>  \
  --destSk <your-access-key-secret>  \
  --destRegionId <your-region-id>  \
  --destInstanceId <your-instance-id>

Ganti placeholder berikut dengan nilai aktual Anda:

Contoh output:

13:40:08 INFO - Begin to migrate topics:[test]
13:40:08 INFO - Total topic number:1
13:40:08 INFO - Will create topic:test, isCompactTopic:false, partition number:1

Tinjau output untuk memastikan daftar topik, jumlah partisi, dan pengaturan compaction.

Fase 2: Commit

Tambahkan flag --commit untuk menjalankan migrasi aktual:

java -jar kafka-migration-assessment.jar TopicMigrationFromZk  \
  --sourceZkConnect 192.168.XX.XX  \
  --destAk <your-access-key-id>  \
  --destSk <your-access-key-secret>  \
  --destRegionId <your-region-id>  \
  --destInstanceId <your-instance-id>  \
  --commit

Contoh output:

13:51:12 INFO - Begin to migrate topics:[test]
13:51:12 INFO - Total topic number:1
13:51:13 INFO - cmd=TopicMigrationFromZk, request=null, response={"code":200,"requestId":"7F76C7D7-AAB5-4E29-B49B-CD6F1E0F508B","success":true,"message":"operation success"}
13:51:13 INFO - TopicCreate success, topic=test, partition number=1, isCompactTopic=false

Tanggapan "code":200 dengan "success":true mengonfirmasi bahwa topik berhasil dibuat di instans tujuan.

Migrasi kelompok konsumen

Migrasi kelompok konsumen memerlukan file properti consumer Kafka untuk membaca offset konsumen dari kluster sumber.

Langkah 1: Buat file kafka.properties

Buat file bernama kafka.properties dengan konten berikut:

# Titik akhir kluster Apache Kafka mandiri
bootstrap.servers=localhost:9092

# ID kelompok konsumen. Gunakan kelompok yang tidak memiliki data offset yang ada
# agar konsumsi dimulai dari pesan pertama.
group.id=XXX

## Jika kluster sumber memerlukan otentikasi, hapus komentar
## dan konfigurasikan parameter berikut:

# Mekanisme otentikasi SASL
#sasl.mechanism=PLAIN

# Protokol keamanan
#security.protocol=SASL_SSL

# Jalur truststore SSL
#ssl.truststore.location=/path/to/kafka.client.truststore.jks

# Kata sandi truststore SSL
#ssl.truststore.password=***

# Jalur konfigurasi JAAS SASL
#java.security.auth.login.config=/path/to/kafka_client_jaas.conf

Langkah 2: Pemeriksaan Awal

Pratinjau kelompok konsumen yang akan dimigrasikan:

java -jar kafka-migration-assessment.jar ConsumerGroupMigrationFromTopic  \
  --propertiesPath /usr/local/kafka_2.12-2.4.0/config/kafka.properties  \
  --destAk <your-access-key-id>  \
  --destSk <your-access-key-secret>  \
  --destRegionId <your-region-id>  \
  --destInstanceId <your-instance-id>

Contoh output:

15:29:45 INFO - Will create consumer groups:[XXX, test-consumer-group]

Langkah 3: Commit

Tambahkan flag --commit untuk menjalankan migrasi aktual:

java -jar kafka-migration-assessment.jar ConsumerGroupMigrationFromTopic  \
  --propertiesPath /usr/local/kafka_2.12-2.4.0/config/kafka.properties  \
  --destAk <your-access-key-id>  \
  --destSk <your-access-key-secret>  \
  --destRegionId <your-region-id>  \
  --destInstanceId <your-instance-id>  \
  --commit

Contoh output:

15:35:51 INFO - cmd=ConsumerGroupMigrationFromTopic, request=null, response={"code":200,"requestId":"C9797848-FD4C-411F-966D-0D4AB5D12F55","success":true,"message":"operation success"}
15:35:51 INFO - ConsumerCreate success, consumer group=XXX
15:35:57 INFO - cmd=ConsumerGroupMigrationFromTopic, request=null, response={"code":200,"requestId":"3BCFDBF2-3CD9-4D48-92C3-385C8DBB9709","success":true,"message":"operation success"}
15:35:57 INFO - ConsumerCreate success, consumer group=test-consumer-group

Monitor progres migrasi

1. Login ke Konsol ApsaraMQ for Kafka.

2. Di bagian Resource Distribution pada halaman Overview, pilih wilayah instans tujuan Anda.

3. Di panel navigasi kiri, klik Migration, lalu klik tab Metadata Import.

4. Temukan instans tujuan Anda dan periksa status migrasi untuk topik dan kelompok konsumen.

Verifikasi topik dan kelompok konsumen yang dimigrasikan

1. Login ke Konsol ApsaraMQ for Kafka.

2. Di bagian Resource Distribution pada halaman Overview, pilih wilayah instans tujuan Anda.

3. Di panel navigasi kiri, klik Instances, lalu klik nama instans.

4. Verifikasi bahwa resource yang dimigrasikan tersedia:

   • Klik Topics di panel navigasi kiri untuk memastikan semua topik tersedia.

   • Klik Groups di panel navigasi kiri untuk memastikan semua kelompok konsumen tersedia.

Prasyarat

Sebelum memulai, pastikan Anda telah:

• Mengunduh MirrorMaker (termasuk dalam distribusi Apache Kafka).

• Topik telah dimigrasikan ke instans tujuan (Langkah 3).

Batasan

Tinjau batasan berikut sebelum memulai migrasi data:

Verifikasi replikasi pesan

Gunakan salah satu metode berikut untuk memastikan MirrorMaker berjalan:

• Periksa lag konsumen di kluster sumber. Jalankan perintah berikut untuk memverifikasi bahwa MirrorMaker mengonsumsi pesan: Nilai LAG yang menurun mengonfirmasi bahwa MirrorMaker aktif mengonsumsi dan meneruskan pesan.

    bin/kafka-consumer-groups.sh --new-consumer --describe \
      --bootstrap-server <source-cluster-endpoint> \
      --group test-consumer-group

• Kirim pesan uji dan verifikasi kedatangannya. Produksi pesan ke topik di kluster sumber, lalu periksa status partisi topik tersebut di Konsol ApsaraMQ for Kafka. Verifikasi bahwa jumlah total pesan meningkat dan konten pesannya benar. Untuk detailnya, lihat Query messages.