Lindorm:Create Data Subscription

Procedure

1. Go to the LTS page. In the left-side navigation pane, choose Change Tracking > Pull mode.

   This takes you to the Subscription channel list page of Lindorm CDC. The Create data subscription channel button is in the upper-right corner. The list displays the subscription channel ID, Lindorm table name, topic name, and actions (Details and Delete) for existing channels.

2. Click Create data subscription channel and configure the following parameters.

   Parameter	Description
   Source cluster	Enter the Lindorm instance ID.
   Lindorm table name	Select the Lindorm table that you want to subscribe to. Each channel can subscribe to only one table.
   Topic name	The name of the data consumption topic.
   Data expiration time (days)	The number of days to retain data. The default value is 7.
   Number of topic partitions	The number of partitions for the topic. Multiple partitions enable concurrent data consumption. The default value is 4.

3. Click Submit.

4. (Optional) To view the details of a channel, find it in the list and click Details in the Operation column. You can view the channel details, consumption details, and storage details.

5. (Optional) Use the following code example with a Kafka client to consume the subscribed data.

   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 {
       // The topic name that you specified when creating the data subscription channel.
       String topic = "test-topic";
       // Properties for connecting to the endpoint.
       Properties props = new Properties();
       // Specify the endpoint address.
       props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "ld-xxx:9092");
       // The key deserializer. Do not change this value.
       props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
       // The value deserializer. Do not change this value.
       props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
       // The name of the consumer group. The group is automatically created during consumption.
       props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id-0");
       // Create a consumer.
       KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props);
       // Subscribe to the topic.
       consumer.subscribe(Arrays.asList(topic));
       // Pull data by using the consumer.
       ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(10000));
       for (ConsumerRecord<byte[], byte[]> record : records) {
         // View the data content.
         System.out.println("key: " + Bytes.toString(record.key()));
         System.out.println("value: " + Bytes.toString(record.value()));
       }
       // Commit the current consumer offset.
       consumer.commitSync();
       // Close the consumer.
       consumer.close();
     }
   }

   Note

   For more information about the data consumption format, see Data consumption format.