Use the SDK demo code to consume tracked data

Procedure

IntelliJ IDEA (Community Edition 2020.1 Windows) is used in this example.

1. Create a change tracking task. For more information, see Track data changes from an ApsaraDB RDS for MySQL instance, Track data changes from a PolarDB for MySQL cluster, or Track data changes from a self-managed Oracle database.
2. Create one or more consumer groups. For more information, see Create consumer groups.
3. Download the SDK demo code package and decompress the package.
4. Go to the directory where the package is decompressed, open the pom.xml file by using a text editor, and then change the SDK version to the latest.
   

   Notice You can obtain the latest version of the change tracking SDK from the Maven website. For more information, visit the Maven page of the change tracking SDK.
5. Open IntelliJ IDEA. In the window that appears, click Open or Import.
   
6. In the dialog box that appears, go to the directory where the package is decompressed, and open the folders to find the pom.xml file.
   
7. In the dialog box that appears, select Open as Project.
8. On the IntelliJ IDEA page, expand the folders to find the Java files. Then, double-click a Java file based on the mode in which you use the SDK client. The following Java files are available: DTSConsumerAssignDemo.java and DTSConsumerSubscribeDemo.java.
   

   Note DTS supports the following modes for using the SDK client:

   • ASSIGN mode: To ensure the global order of messages, DTS assigns only one partition (Partition 0) to each tracked topic. If you use the SDK client in ASSIGN mode, we recommend that you start only one SDK client.
   • SUBSCRIBE mode: To ensure the global order of messages, DTS assigns only one partition (Partition 0) to each tracked topic. In SUBSCRIBE mode, you can start multiple SDK clients in a consumer group at the same time to implement disaster recovery. If an SDK client in the consumer group fails, other SDK clients will be randomly and automatically allocated to Partition 0 to resume data consumption.
9. Set the required parameters in the code of the Java file.
   

   Table 1. Required parameters

   Parameter	Description	Method to obtain
   brokerUrl	The network address and port number of the change tracking instance. Note If you track data changes over internal networks, the network latency is minimal. This is applicable if the ECS instance where you deploy the SDK client belongs to the classic network or the same VPC as the change tracking instance.	In the DTS console, click the instance ID. On the Task Management page, you can obtain the network address and port number.
   topic	The topic of the change tracking instance.	In the DTS console, click the instance ID. On the Task Management page, you can obtain the tracked Topic.
   sid	The ID of the consumer group.	In the DTS console, click the instance ID, and then click Consume Data. You can obtain the Consumer Group ID and the corresponding Account. Note The password of the consumer group account is automatically specified when you create a consumer group.
   userName	The username of the consumer group. Warning If you are not using the SDK client that is described in this topic, you must specify this parameter in the following format: <Username>-<Consumer group ID>, for example, dtstest-dtsae******bpv. Otherwise, the connection fails.
   password	The password of the account.
   initCheckpoint	The consumer offset. It is the timestamp when the SDK client consumes the first data record. The value is a UNIX timestamp, for example, 1620962769. Note The consumer offset can be used in the following scenarios: If the consumption process is interrupted, you can specify the consumer offset to resume data consumption. This allows you to prevent data loss. When you start the change tracking client, you can specify the consumer offset to consume data on demand.	The consumer offset must be within the data range of the change tracking instance, as shown in the following figure. The consumer offset must be converted to a UNIX timestamp. Note You can use a search engine to obtain a UNIX timestamp converter.
   ConsumerContext.ConsumerSubscribeMode subscribeMode	The mode in which you use the SDK client. Valid values: ConsumerContext.ConsumerSubscribeMode.ASSIGN: In ASSIGN mode, only one SDK client in a consumer group can consume tracked data. For more information, see SDK usage modes. ConsumerContext.ConsumerSubscribeMode.SUBSCRIBE: In SUBSCRIBE mode, you can start multiple SDK clients in a consumer group at the same time to implement disaster recovery. For more information, see SDK usage modes.	None

10. In the top menu bar of IntelliJ IDEA, choose Run > Run to run the client.
    Note When you run IntelliJ IDEA for the first time, it requires some time to load and install the relevant dependency.

    • The following figure shows the result which indicates that the SDK client can track data changes from the source database.
    • The SDK client calculates and displays information about the consumed data at regular intervals. The information includes the total number of data records that are sent and received, the total amount of data, and the number of requests per second (RPS).

      Table 2. Statistics of consumed data

      Parameter	Description
      outCounts	The total number of data records consumed by the SDK client.
      outBytes	The total amount of data consumed by the SDK client. Unit: bytes.
      outRps	The number of requests per second when the SDK client consumes data.
      outBps	The number of bits transmitted per second when the SDK client consumes data.
      inBytes	The total amount of data that is sent by the DTS server. Unit: bytes.
      DStoreRecordQueue	The size of the current data cache queue when the DTS server sends data.
      inCounts	The total number of data records that are sent by the DTS server.
      inRps	The number of requests per second when the DTS server sends data.
      __dt	The current timestamp when the SDK client receives data. Unit: milliseconds.
      DefaultUserRecordQueue	The size of the current data cache queue after serialization.

Save and query the consumer offset

Troubleshooting

ERROR
CheckResult{isOk=false, errMsg='telnet dts-cn-hangzhou.aliyuncs.com:18009
failed, please check the network and if the brokerUrl is correct'}
(com.aliyun.dts.subscribe.clients.DefaultDTSConsumer)

telnet real node *** failed, please check the network

ERROR CheckResult{isOk=false, errMsg='build kafka consumer failed, error: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata, probably the user name or password is wrong'} (com.aliyun.dts.subscribe.clients.DefaultDTSConsumer)

com.aliyun.dts.subscribe.clients.exception.TimestampSeekException: RecordGenerator:seek timestamp for topic [cn_hangzhou_rm_bp11tv2923n87081s_rdsdt_dtsacct-0] with timestamp [1610249501] failed