Data Transmission Service:Use an SDK demo to consume change data from PolarDB-X 1.0

Precautions

To consume change data as a RAM user, you must grant that user the AliyunDTSFullAccess permission and permissions to access the source objects. For more information, see Authorize a RAM user to manage DTS and Manage RAM user permissions.

Procedure

This topic demonstrates how to run the SDK demo to consume change data from a PolarDB-X 1.0 instance, using IntelliJ IDEA Community Edition 2020.1 for Windows as an example.

1. Create a change tracking instance. For more information, see Create a change tracking task for PolarDB-X 1.0.

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

3. Download and decompress the SDK demo. For the download link, see SDK Demo Code.

   Important

   When you consume change data, you must call the commit method of DefaultUserRecord to submit the consumer offset. Otherwise, you risk consuming data repeatedly.

4. Open the project in IntelliJ IDEA.

   1. Open IntelliJ IDEA and click Open or Import.

   2. In the dialog box, navigate to the directory where you decompressed the SDK demo, expand the folders, and then double-click the pom.xml file.

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

5. In IntelliJ IDEA, expand the folders. Based on the usage mode of the SDK client, select and double-click to open the corresponding Java file: DistributedDTSConsumerDemo.

   aliyun-dts-subscribe-sdk-java-master [dts-new-subscr...]
     .idea
     src
       main
       test
         java
           com.aliyun.dts.subscribe.clients
             DBMapperTest
             DistributedDTSConsumerDemo
             DTSConsumerAssignDemo
             DTSConsumerSubscribeDemo
             UserMetaStore
     .gitignore
     dts-new-subscribe-sdk.iml
     LICENSE
     pom.xml
     README.md
   External Libraries
   Scratches and Consoles

6. Set the required parameters in the Java file.

   public static void main(String[] args) throws ClientException {
           // Configure the change tracking settings for a distributed data source, such as a PolarDB-X 1.0 instance. Set parameters such as the AccessKey pair, instance ID, 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";
           // The initial consumer offset, a Unix timestamp in seconds. e.g., 1566180200 for Mon Aug 19 10:03:21 CST 2019.
           String checkpoint = "1639620090";
           // Convert physical database/table name to logical database/table name
           boolean mapping = true;
           // Set to true to force the client to start from the specified checkpoint. A checkpoint reset works only in ASSIGN mode.
           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.
   regionId	The region ID of the change tracking instance.	In the DTS console, click the ID of the target change tracking instance. On the Task Management page, find the region information. For example, if the region is China (Hangzhou), set this parameter to cn-hangzhou. For a list of regions, see List of supported regions.
   dtsInstanceId	The ID of the change tracking instance.	In the DTS console, click the ID of the target change tracking instance. On the Task Management page, you can find the instance ID and task ID.
   jobId	The ID of the change tracking task.
   sid	The ID of the consumer group.	In the DTS console, click the ID of the target change tracking instance and then click Consume Data. You can find the SID and the Account of the consumer group. Note The password for 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 instance. Note To minimize network latency, use the internal endpoint if the ECS instance running the SDK client is in the same classic network or VPC as the change tracking instance. 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 Task Management page, you can find the endpoint and port number.
   checkpoint	The consumer offset, specified as a Unix timestamp in seconds. The SDK client starts consuming data from this point in time. Note The consumer offset is used in the following scenarios: If your application is interrupted, pass the last consumer offset to resume consumption without data loss. When the client starts, you can pass a specific consumer offset to consume data on demand.	The consumer offset must be within the data range of the change tracking instance and must be a Unix timestamp. Note You can use a search engine to find a Unix timestamp converter.

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

   Note

   On the first run, IntelliJ IDEA automatically installs the required dependencies, which may take some time.

   • The output shows that the client can consume data changes from the source instance.

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

     Table 1. 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	Data consumption rate of the SDK client, in requests per second (RPS).
     outBps	The data consumption rate of the SDK client, in bits per second (bps).
     count	This parameter is reserved.
     inBytes	The total volume of data sent by the DTS server, in bytes.
     DStoreRecordQueue	The current size of the data cache queue on the DTS server.
     inCounts	The total number of data records sent by the DTS server.
     inRps	The data sending rate of the DTS server, in requests per second (RPS).
     inBps	The data sending rate of the DTS server, in bits per second (bps).
     __dt	The timestamp when the SDK client receives the data, in milliseconds.
     DefaultUserRecordQueue	The current size of the data cache queue after serialization.

8. Optional: To customize how consumed data is processed, you can modify the buildRecordListener() method or use a custom class.

   public static Map<String, RecordListener> buildRecordListener() {
           // You can implement your 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 the record.
                       RecordListener recordPrintListener = new DefaultRecordPrintListener(DbType.MySQL);
                       recordPrintListener.consume(record);
                       // The commit method pushes the checkpoint update.
                       record.commit("");
                   }
               }
           };
           return Collections.singletonMap("mysqlRecordPrinter", mysqlRecordPrintListener);
       }