After you configure a change tracking task, you can use the SDK demo code that is provided by Data Transmission Service (DTS) to track and consume data. This topic describes how to use the SDK demo code to consume tracked data.
Before you begin
Create an AccessKey pair, which consists of an AccessKey ID and AccessKey secret. For more information, see Create an AccessKey pair.
IntelliJ IDEA (Community Edition 2020.1 Windows) is used in this example.
- Create a change tracking task. For more information, see Track data changes from an ApsaraDB RDS for MySQL instance (new), Track data changes from a PolarDB for MySQL cluster, or Track data changes from a self-managed Oracle database.
- Create one or more consumer groups. For more information, see Create consumer groups.
- Download the SDK demo code package and decompress the package.
- 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.
- Open IntelliJ IDEA. In the window that appears, click Open or Import.
- In the dialog box that appears, go to the directory where the package is decompressed,
and expand the folders to find the pom.xml file.
- In the dialog box that appears, select Open as Project.
- 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.
- Set the required parameters in the code of the Java file.
Table 1. Required parameters Parameter Description Method to obtain
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 View Task Settings page, you can obtain the tracked topic, network address, and port number.
The topic of the change tracking instance.
The ID of the consumer group. In the DTS console, click the instance ID, and then click Data Consume. You can obtain the consumer group ID and the corresponding username.Note When you create a consumer group, the password of the consumer group is automatically specified.
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.
The password of the consumer group.
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 is useful in the following scenarios:
- If the consumption process is interrupted, you can specify the consumer offset on the change tracking client to resume data consumption. This allows you to prevent against 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.
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.
- In the top menu bar of IntelliJ IDEA, choose 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 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
The total number of data records that are consumed by the SDK client.
The total amount of data that is consumed by the SDK client. Unit: bytes.
The number of requests per second when the SDK client consumes data.
The number of bits transmitted per second when the SDK client consumes data.
The total amount of data that is sent by the DTS server. Unit: bytes.
The size of the current data cache queue when the DTS server sends data.
The total number of data records that are sent by the DTS server.
The number of requests per second when the DTS server sends data.
The current timestamp when the SDK client receives data. Unit: milliseconds.
The size of the current data cache queue after serialization.
Save and query the consumer offsetWhen the SDK client is started for the first time or restarted, or an internal retry occurs, you need to query and specify the consumer offset to start or resume data consumption. The following table describes how to manage and query the consumer offset in different situations. To implement on-demand data consumption, you must ensure that no data is lost or duplicate.
|Scenario||SDK usage mode||Query method|
|Query the consumer offset||ASSIGN and SUBSCRIBE||
|When you start the SDK client for the first time, you must specify the consumer offset to consume data.||ASSIGN and SUBSCRIBE||Select the DTSConsumerAssignDemo.java or DTSConsumerSubscribeDemo.java file based on the mode in which you use the SDK client. Then, specify the
|If an internal retry occurs, you must specify the consumer offset of the last data record to resume data consumption.||ASSIGN||Perform the following steps to find the consumer offset of the last data record:
|SUBSCRIBE||Perform the following steps to find the consumer offset of the last data record:
|After the SDK client is restarted, you must specify the consumer offset of the last data record to resume data consumption.||ASSIGN||Check the setting of the
|SUBSCRIBE||In this mode, the setting of the
||Enter the valid
||The broker address cannot be redirected to the real IP address.|
||The specified username and password are invalid.|
||Enter the consumer offset within the data range of the change tracking instance. For more information, see Table 1.|
|The response time of data consumption is increased.||None||You can analyze the cause by querying the DStoreRecordQueue and DefaultUserRecordQueue parameters. For more information, see Table 2.