How to use the SDK client to consume data tracked from a PolarDB-X 1.0 instance - Data Transmission Service

After you create a change tracking task, you can use the software development kit (SDK) provided by Data Transmission Service (DTS) to subscribe to data changes. This topic describes how to use an SDK to consume data from distributed data sources, such as PolarDB-X 1.0 and DMS logical databases.

Prerequisites

You have created a change tracking instance. The instance is in the Normal state. For more information, see Create a change tracking task for a PolarDB-X 1.0 instance or Create a change tracking task for a DMS logical database.
You have created a consumer group for your subscription instance.
If you use a RAM user to consume subscribed data, the RAM user must have the AliyunDTSFullAccess permission and access permissions to the subscribed objects. For more information, see Grant permissions to a RAM user to manage DTS using a system policy and Manage the permissions of a RAM user.

Usage notes

When you consume subscribed data, you must call the commit method of DefaultUserRecord to commit the offset information. Otherwise, data might be consumed repeatedly.
Different consumption processes are independent of each other.

Procedure

Download and decompress the SDK sample code.
Verify the SDK code version.
1. Go to the directory where you decompressed the sample SDK code.
2. Use a text editor to open the pom.xml file in the directory.
3. Update the change tracking SDK to the latest version.
  
  Note
  You can find the latest Maven dependency on the dts-new-subscribe-sdk page.
  Location of the SDK version parameter (click to expand)
  <name>dts-new-subscribe-sdk</name> <url>https://www.aliyun.com/product/dts</url> <description>The Aliyun new Subscribe SDK for Java used for accessing Data Transmission Service</description> <packaging>jar</packaging> <groupId>com.aliyun.dts</groupId> <artifactId>dts-new-subscribe-sdk</artifactId> <version>2.1.4</version>

Edit the SDK code.

You can open the decompressed file with encoding software.
Based on the usage pattern of the SDK client, open the DistributedDTSConsumerDemo.java file.
Note
The path of the Java file is aliyun-dts-subscribe-sdk-java-master/src/test/java/com/aliyun/dts/subscribe/clients/.

Set the parameters in the Java code.

public static void main(String[] args) throws ClientException {
        // Configuration for subscribing to a distributed data source, such as PolarDB-X 1.0 (formerly DRDS). Configure information such as the AccessKey, instance ID, main task ID, and consumer group.
        String accessKeyId = "LTA***********99reZ";
        String accessKeySecret = "****************";
        String regionId = "cn-hangzhou";
        String dtsInstanceId = "dtse5212sed162****";
        String jobId = "l791216x16d****";
        String sid = "dtsip412t13160****";
        String userName = "xftest";
        String password = "******";
        String proxyUrl = "dts-cn-****.com:18001";
        // initial checkpoint for first seek(a timestamp to set, eg 1566180200 if you want (Mon Aug 19 10:03:21 CST 2019))
        String checkpoint = "1639620090";

        // Convert physical database/table name to logical database/table name
        boolean mapping = true;
        // if force use config checkpoint when start. for checkpoint reset, only assign mode works
        boolean isForceUseInitCheckpoint = false;

        ConsumerContext.ConsumerSubscribeMode subscribeMode = ConsumerContext.ConsumerSubscribeMode.ASSIGN;
        DistributedDTSConsumerDemo demo = new DistributedDTSConsumerDemo(userName, password, regionId,
                jobId, sid, dtsInstanceId, accessKeyId, accessKeySecret, subscribeMode, proxyUrl,
                checkpoint, isForceUseInitCheckpoint, mapping);
        demo.start();
    }

Parameter	Description	How to obtain
accessKeyId	The AccessKey ID.	For more information, see Obtain an AccessKey pair.
accessKeySecret	The AccessKey secret.	For more information, see Obtain an AccessKey pair.
regionId	The ID of the region where the change tracking task is located.	In the DTS console, click the ID of the target change tracking instance. On the Basic Information page, you can retrieve the region information. For example, if the region is China (Hangzhou), set this parameter to `cn-hangzhou`. For more information, see List of regions.
dtsInstanceId	The ID of the change tracking instance.	In the DTS console, click the ID of the target change tracking instance. On the Basic Information page, you can retrieve the DTS Instance ID of the change tracking instance.
jobId	The ID of the change tracking task.	You can call the DescribeDtsJobs operation to retrieve the change tracking task ID (DtsJobId).
sid	The ID of the consumer group.	In the DTS console, click the ID of the target change tracking instance. In the navigation pane on the left, click Consume Data. You can retrieve the Consumer Group ID/Name and the Account of the consumer group. Note The password of the consumer group account is specified when you create the consumer group.
userName	The account of the consumer group.
password	The password of the consumer group account.
proxyUrl	The endpoint and port of the change tracking channel. Note If the ECS instance where you deploy the SDK client and the change tracking channel are in the same classic network or virtual private cloud (VPC), subscribe to data over the internal network to achieve the lowest latency. We do not recommend using a public endpoint because of potential network instability.	In the DTS console, click the ID of the target change tracking instance. On the Basic Information page, you can retrieve the Network information.
checkpoint	The consumer offset. This is the timestamp from which the SDK client starts to consume data records. The value is a UNIX timestamp in seconds. Note You can use consumer offset information for the following: If the consumption process is interrupted, you can pass the consumer offset to resume data consumption and prevent data loss. When you start the SDK client, you can pass the required consumer offset to adjust the subscription offset and consume data as needed.	The consumer offset must be within the timestamp range of the change tracking instance and must be converted to a UNIX timestamp. Note You can view the timestamp range of the change tracking instance in the Data Range column of the tracking task list. You can use a search engine to find a UNIX timestamp converter.

Optional: To modify the data type of the subscribed data, you can modify the buildRecordListener() method or use a custom class.

public static Map<String, RecordListener> buildRecordListener() {
        // user can impl their own listener
        RecordListener mysqlRecordPrintListener = new RecordListener() {
            @Override
            public void consume(DefaultUserRecord record) {

                OperationType operationType = record.getOperationType();

                if (operationType.equals(OperationType.INSERT)
                        || operationType.equals(OperationType.UPDATE)
                        || operationType.equals(OperationType.DELETE)
                        || operationType.equals(OperationType.HEARTBEAT)) {

                    // consume record
                    RecordListener recordPrintListener = new DefaultRecordPrintListener(DbType.MySQL);

                    recordPrintListener.consume(record);

                    //commit method push the checkpoint update
                    record.commit("");
                }
            }
        };
        return Collections.singletonMap("mysqlRecordPrinter", mysqlRecordPrintListener);
    }

Open the project structure in your IDE and make sure that the OpenJDK version for the project is 1.8.

Run the client code.

The output shows that the client is subscribing to data changes from the source database.

The SDK client periodically collects and displays statistics about data consumption. The statistics include the total number of data records sent and received, the total data volume, and the records per second (RPS).

Table 1. Data consumption statistics

Parameter	Description
`outCounts`	The total number of data records consumed by the SDK client.
`outBytes`	The total volume of data consumed by the SDK client, in bytes.
`outRps`	The number of requests per second sent by the SDK client to consume data.
`outBps`	The number of bits transmitted per second when the SDK client consumes data.
`count`	None.
`inBytes`	The total volume of data sent by the DTS server, in bytes.
`DStoreRecordQueue`	The size of the data cache queue when the DTS server sends data.
`inCounts`	The total number of data records sent by the DTS server.
`inRps`	The number of requests sent by the DTS server per second.
`inBps`	The number of bits transmitted per second when the DTS server sends data.
`__dt`	The timestamp when the SDK client receives the data, in milliseconds.
`DefaultUserRecordQueue`	The size of the data cache queue after serialization.