Lindorm:Buat Langganan Data

Prosedur

1. Buka halaman LTS. Di panel navigasi sebelah kiri, pilih Change Tracking > Pull mode.

   Anda akan diarahkan ke halaman Subscription channel list dari Lindorm CDC. Tombol Create data subscription channel berada di pojok kanan atas. Daftar tersebut menampilkan ID saluran langganan, nama tabel Lindorm, nama topik, dan aksi (Details dan Delete) untuk saluran yang sudah ada.

2. Klik Create data subscription channel dan konfigurasikan parameter berikut.

   Parameter	Deskripsi
   Source cluster	Masukkan ID instans Lindorm.
   Lindorm table name	Pilih tabel Lindorm yang ingin Anda langganan. Setiap saluran hanya dapat berlangganan pada satu tabel.
   Topic name	Nama topik konsumsi data.
   Data expiration time (days)	Jumlah hari penyimpanan data. Nilai default adalah 7.
   Number of topic partitions	Jumlah partisi untuk topik. Beberapa partisi memungkinkan konsumsi data secara konkuren. Nilai default adalah 4.

3. Klik Submit.

4. (Opsional) Untuk melihat detail saluran, temukan saluran tersebut dalam daftar lalu klik Details di kolom Operation. Anda dapat melihat detail saluran, detail konsumsi, dan detail penyimpanan.

5. (Opsional) Gunakan contoh kode berikut dengan client Kafka untuk mengonsumsi data yang telah dilanggan.

   import org.apache.hadoop.hbase.util.Bytes;
   import org.apache.kafka.clients.admin.AdminClientConfig;
   import org.apache.kafka.clients.consumer.ConsumerConfig;
   import org.apache.kafka.clients.consumer.ConsumerRecord;
   import org.apache.kafka.clients.consumer.ConsumerRecords;
   import org.apache.kafka.clients.consumer.KafkaConsumer;
   import org.apache.kafka.common.serialization.ByteArrayDeserializer;
   import java.time.Duration;
   import java.util.Arrays;
   import java.util.Properties;
   public class TestConsume {
     public static void main(String[] args) throws Exception {
       // Nama topik yang Anda tentukan saat membuat saluran langganan data.
       String topic = "test-topic";
       // Properti untuk menghubungkan ke titik akhir.
       Properties props = new Properties();
       // Tentukan alamat titik akhir.
       props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "ld-xxx:9092");
       // Deserializer kunci. Jangan ubah nilai ini.
       props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
       // Deserializer nilai. Jangan ubah nilai ini.
       props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
       // Nama kelompok konsumen. Kelompok ini dibuat otomatis saat konsumsi.
       props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id-0");
       // Buat konsumen.
       KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props);
       // Berlangganan ke topik.
       consumer.subscribe(Arrays.asList(topic));
       // Tarik data menggunakan konsumen.
       ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(10000));
       for (ConsumerRecord<byte[], byte[]> record : records) {
         // Lihat isi data.
         System.out.println("key: " + Bytes.toString(record.key()));
         System.out.println("value: " + Bytes.toString(record.value()));
       }
       // Commit offset konsumen saat ini.
       consumer.commitSync();
       // Tutup konsumen.
       consumer.close();
     }
   }

   Catatan

   Untuk informasi selengkapnya mengenai format konsumsi data, lihat Format konsumsi data.