Data Transmission Service:Use a Kafka client to consume tracked data

Usage notes

• If you enable auto commit when you use the change tracking feature, some data may be committed before it is consumed. This results in data loss. We recommend that you manually commit data.

  Note

  If data fails to be committed, you can restart the client to continue consuming data from the last recorded consumption checkpoint. However, duplicate data may be generated during this period. You must manually filter out the duplicate data.

• Data is serialized and stored in the Avro format. For more information, see Record.avsc.

  Warning

  If you are not using the Kafka client that is described in this topic, you must parse the tracked data based on the Avro schema (DTS Avro deserialization example) and check the parsed data.

• The search unit is second when Data Transmission Service (DTS) calls the offsetForTimes operation. The search unit is millisecond when a native Kafka client calls this operation.

• Transient connections may occur between a Kafka client and the change tracking server due to several reasons, such as disaster recovery. If you are not using the Kafka client that is described in this topic, your Kafka client must have network reconnection capabilities.

• If you use a native Kafka client to consume tracked data, the incremental data collection module may be changed in DTS. In subscribe mode, the consumption checkpoint that the Kafka client saves to the DTS server is removed. You need to specify a consumption checkpoint to consume tracked data based on your business requirements. If you want to consume data in subscribe mode, we recommend that you use the SDK demo that is provided by DTS to track and consume data, or manually manage the consumption checkpoint. For more information, see Consume subscribed data using an SDK and the Manage the consumption checkpoint section of this topic.

Run the Kafka client

Download the Kafka client demo. For more information about how to use the demo, see Readme.

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.0.0</version>
</dependency>

Procedure

The following steps show how to run the Kafka client to consume tracked data. In this example, IntelliJ IDEA Community Edition 2018.1.4 for Windows is used.

1. Create a change tracking instance. For more information, see Change tracking overview.

2. Create one or more consumer groups. For more information, see Add a consumer group.

3. Download the package of Kafka client demo and decompress the package.

   Note

   Click and select Download ZIP to download the package.

4. Open IntelliJ IDEA. In the window that appears, click Open.

5. In the dialog box that appears, go to the directory in which the downloaded demo resides. Find the pom.xml file.

   Navigate to kafkademo > subscribe_example-master > javaimpl, select pom.xml, and then click OK.

6. In the dialog box that appears, select Open as Project.

7. In the Project tool window of IntelliJ IDEA, click folders to find the demo file of the Kafka client, and double-click the file. The file name is NotifyDemoDB.java.

8. Specify the parameters in the NotifyDemoDB.java file.

   public static Properties getConfigs() {
       Properties properties = new Properties();
       // user password and sid for auth
       properties.setProperty(USER_NAME, "dtstest");
       properties.setProperty(PASSWORD_NAME, "xxx");
       properties.setProperty(SID_NAME, "dtsxxx");
       // kafka consumer group general same with sid
       properties.setProperty(GROUP_NAME, "dtsxxx");
       // topic to consume, partition is 0
       properties.setProperty(KAFKA_TOPIC, "cn_hangzhou_xxx");
       // kafka broker url
       properties.setProperty(KAFKA_BROKER_URL_NAME, "dts-cn-xxx.com:18001");
       // initial checkpoint for first seek(a timestamp to set, eg 1566180200 if you want (Mon Aug 19 10:03:21 CST 2019))
       properties.setProperty(INITIAL_CHECKPOINT_NAME, "1583307907");
       // if force use config checkpoint when start. for checkpoint reset
       properties.setProperty(USE_CONFIG_CHECKPOINT_NAME, "true");
       // use consumer assign or subscribe interface
       // when use subscribe mode, group config is required. kafka consumer group is enabled
       properties.setProperty(SUBSCRIBE_MODE_NAME, "assign");
       return properties;
   }

   Parameter	Description	Method to obtain the parameter value
   USER_NAME	The username of the consumer group account. Warning If you are not using the Kafka client that is described in this topic, you must specify this parameter in the following format: <Username>-<Consumer group ID>. Example: dtstest-dtsae******bpv. Otherwise, the connection fails.	In the DTS console, find the change tracking instance that you want to manage and click the instance ID. In the left-side navigation pane, click Consume Data. On the page that appears, you can view the information about a consumer group, such as the ID or name and account of the consumer group. Note The password of the consumer group account is specified when you create the consumer group.
   PASSWORD_NAME	The password of the account.
   SID_NAME	The ID of the consumer group.
   GROUP_NAME	The name of the consumer group. Set this parameter to the consumer group ID.
   KAFKA_TOPIC	The name of the tracked topic of the change tracking instance.	In the DTS console, find the change tracking instance that you want to manage and click the instance ID. On the Basic Information page, you can view the information about topic and network. In the Basic Information section of the change tracking task details page, obtain the Subscription Topic value. In the Network section, obtain the VPC network address (format example: xxx.aliyuncs.com:18003).
   KAFKA_BROKER_URL_NAME	The endpoint of the change tracking instance. Note If you track data changes over internal networks, the network latency is minimal. This is applicable if the Elastic Compute Service (ECS) instance on which you deploy the Kafka client resides on the classic network or in the same virtual private cloud (VPC) as the change tracking instance.
   INITIAL_CHECKPOINT_NAME	The consumption checkpoint of consumed data. The value is a UNIX timestamp. Example: 1592269238. Note You must save the consumption checkpoint for the following reasons: If the consumption process is interrupted, you can specify the consumption checkpoint on the Kafka client to resume data consumption. This prevents data loss. When you start the Kafka client, you can specify the consumption checkpoint to consume data based on your business requirements. If the SUBSCRIBE_MODE_NAME parameter is set to subscribe, the INITIAL_CHECKPOINT_NAME parameter that you specified takes effect only when you start the Kafka client for the first time.	The consumption checkpoint of consumed data must be within the data range of the change tracking instance. The consumption checkpoint must be converted into a UNIX timestamp. In the DTS change tracking task list, you can view the Data Range field of the corresponding task, which shows the start and end time of consumable data. You can use this field to determine the valid value range of INITIAL_CHECKPOINT_NAME. Note You can view the data range of the change tracking instance in the Data Range column on the Change Tracking Tasks page. You can use a search engine to obtain a UNIX timestamp converter.
   USE_CONFIG_CHECKPOINT_NAME	Specifies whether to force the client to consume data from the specified consumption checkpoint. Default value: true. You can set this parameter to true to prevent the data that is received but not processed from being lost.	None
   SUBSCRIBE_MODE_NAME	Specifies whether to run two or more Kafka clients for a consumer group. If you want to use this feature, set this parameter to subscribe for these Kafka clients. The default value is assign, which indicates that the feature is not used. We recommend that you deploy only one Kafka client for a consumer group.	None

9. In the top menu bar of IntelliJ IDEA, choose Run > Run to run the client.

   Note

   If you run IntelliJ IDEA for the first time, a specific time period is required to load and install relevant dependencies.

Results on the Kafka client

The following result shows that the Kafka client can track data changes from the source database.

[2020-03-09 10:41:52,408] INFO [Consumer clientId=consumer-1, groupId=dts_xxx] Discovered coordinator xxx (id: xxx rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2020-03-09 10:41:57,203] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721711, offset: 1732521, info: 1583721711] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:41:57,571] INFO EtlRecordProcessor: haven't receive records from generator for  5s (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:02,203] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721721, offset: 1732539, info: 1583721721] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:07,204] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721726, offset: 1732544, info: 1583721726] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:12,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721731, offset: 1732548, info: 1583721731] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:17,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721736, offset: 1732554, info: 1583721736] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:22,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721741, offset: 1732559, info: 1583721741] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:27,206] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721746, offset: 1732569, info: 1583721746] (recordprocessor.EtlRecordProcessor)

You can delete the double forward slashes (//) from the //log.info(ret) string in line 25 of the NotifyDemoDB.java file. Then, run the client again to view the data change information.

FAQ

• Q: Why do I need to record the consumption checkpoint of the Kafka client?

  A: The consumption checkpoint recorded by DTS is the point in time when DTS receives a commit operation from the Kafka client. The recorded consumption checkpoint may be different from the actual consumption time. If a business application or the Kafka client is unexpectedly interrupted, you can specify an accurate consumption checkpoint to continue data consumption. This prevents data loss or duplicate data consumption.

Manage the consumption checkpoint

1. Configure the Kafka client to listen to the switchover of the data collection module in DTS.

   You can configure the consumer properties of the Kafka client to listen to the switchover of the data collection module in DTS. The following code provides an example on how to configure the consumer properties:

   properties.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, ClusterSwitchListener.class.getName());

   The following code provides an example on how to implement ClusterSwitchListener:

   public class ClusterSwitchListener implements ClusterResourceListener, ConsumerInterceptor {
       private final static Logger LOG = LoggerFactory.getLogger(ClusterSwitchListener.class);
       private ClusterResource originClusterResource = null;
       private ClusterResource currentClusterResource = null;
   
       public ConsumerRecords onConsume(ConsumerRecords records) {
           return records;
       }
   
   
       public void close() {
       }
   
       public void onCommit(Map offsets) {
       }
   
   
       public void onUpdate(ClusterResource clusterResource) {
           synchronized (this) {
               originClusterResource = currentClusterResource;
               currentClusterResource = clusterResource;
               if (null == originClusterResource) {
                   LOG.info("Cluster updated to " + currentClusterResource.clusterId());
               } else {
                   if (originClusterResource.clusterId().equals(currentClusterResource.clusterId())) {
                       LOG.info("Cluster not changed on update:" + clusterResource.clusterId());
                   } else {
                       LOG.error("Cluster changed");
                       throw new ClusterSwitchException("Cluster changed from " + originClusterResource.clusterId() + " to " + currentClusterResource.clusterId()
                               + ", consumer require restart");
                   }
               }
           }
       }
   
       public boolean isClusterResourceChanged() {
           if (null == originClusterResource) {
               return false;
           }
           if (originClusterResource.clusterId().equals(currentClusterResource.clusterId())) {
               return false;
           }
           return true;
       }
   
       public void configure(Map<String, ?> configs) {
       }
   
       public static class ClusterSwitchException extends KafkaException {
           public ClusterSwitchException(String message, Throwable cause) {
               super(message, cause);
           }
   
           public ClusterSwitchException(String message) {
               super(message);
           }
   
           public ClusterSwitchException(Throwable cause) {
               super(cause);
           }
   
           public ClusterSwitchException() {
               super();
           }
   
       }

2. Specify the consumption checkpoint based on the captured switchover of the data collection module in DTS.

   Set the start consumption checkpoint of the next data tracking to the timestamp of the latest tracked data entry that was consumed by the client. The following code provides an example on how to specify the consumption checkpoint:

   try{
      //do some action
   } catch (ClusterSwitchListener.ClusterSwitchException e) {
      reset();
   }
   
   // Reset the consumption checkpoint.
   public reset() {
     long offset = kafkaConsumer.offsetsForTimes(timestamp);
     kafkaConsumer.seek(tp,offset);
   }

   Note

   For more information about the examples, see KafkaRecordFetcher.

Mappings between MySQL data types and dataTypeNumber values

For more information, see SQL Type field.

Mappings between Oracle data types and dataTypeNumber values

Mappings between PostgreSQL data types and dataTypeNumber values