Data Transmission Service:Consume subscribed data using an SDK

Step 1: Download and update the SDK

1. Download the aliyun-dts-subscribe-sdk-java repository and decompress the package.

2. Open pom.xml in the decompressed directory and update dts-new-subscribe-sdk to the latest version. Find the latest version on the dts-new-subscribe-sdk Maven page. The version entry in pom.xml looks like this:

   <groupId>com.aliyun.dts</groupId>
   <artifactId>dts-new-subscribe-sdk</artifactId>
   <version>2.1.4</version>

Step 2: Configure the SDK client

1. Open the decompressed project in your integrated development environment (IDE).

2. Navigate to aliyun-dts-subscribe-sdk-java-master/src/test/java/com/aliyun/dts/subscribe/clients/ and open the Java file for your chosen mode: DTSConsumerAssignDemo.java or DTSConsumerSubscribeDemo.java.

3. Set the connection parameters in the main method:

   Parameter	Description	Where to find it
   brokerUrl	Network address and port of the change tracking channel. Use the virtual private cloud (VPC) address when the SDK client and change tracking instance are in the same VPC to reduce latency. Avoid public endpoints due to potential network instability.	DTS console → subscription instance ID → Basic Information page → Network section
   topic	Topic of the change tracking channel	DTS console → subscription instance ID → Basic Information page → Basic Information section → Topic
   sid	Consumer group ID	DTS console → subscription instance ID → Consume Data page → Consumer Group ID/Name
   userName	Consumer group username. If you are not using the provided SDK client, format the username as <consumer group username>-<consumer group ID> (for example, dtstest-dtsae******bpv).	DTS console → subscription instance ID → Consume Data page → Account
   password	Consumer group password	The password set when you created the consumer group
   initCheckpoint	UNIX timestamp of the first record to consume. Must be within the Data Range shown in the tracking task list. Use this to resume from a specific point after an interruption or to adjust the starting offset.	Convert the target time to a UNIX timestamp
   subscribeMode	SDK client mode. Set to ASSIGN or SUBSCRIBE to match the Java file you opened. Do not change this after initial configuration.	N/A

   public static void main(String[] args) {
       // Kafka broker address and port of the change tracking channel
       String brokerUrl = "dts-cn-***.com:18001";
       // Topic of the change tracking channel (partition 0)
       String topic = "cn_***_version2";
       // Consumer group credentials
       String sid = "dts***";
       String userName = "dts***";
       String password = "DTS***";
       // Starting offset: UNIX timestamp of the first record to consume
       // Example: 1566180200 starts consumption from 10:03:21 CST on Monday, August 19, 2019
       String initCheckpoint = "1740472***";
       // Consumption mode — matches the file you opened
       ConsumerContext.ConsumerSubscribeMode subscribeMode = ConsumerContext.ConsumerSubscribeMode.SUBSCRIBE;
   
       DTSConsumerSubscribeDemo consumerDemo = new DTSConsumerSubscribeDemo(brokerUrl, topic, sid, userName, password, initCheckpoint, subscribeMode);
       consumerDemo.start();
   }

4. In your IDE, open the project structure settings and confirm that the project uses OpenJDK 1.8.

Step 3: Run the client

Run the project. The IDE downloads required plugins and dependencies on the first run, which may take a moment.

[2025-02-25 18:47:22.991] [INFO ] [...KafkaRecordFetcher] - [Consumer clientId=consumer-dtsl5vy2ao5250****-1, groupId=dtsl5vy2ao5250****] Seeking to offset 8200 for partition cn_hangzhou_vpc_rm_bp15uddebh4a1****_dts****_version2-0
[2025-02-25 18:47:23.226] [INFO ] [...EtlRecordProcessor] [...DefaultRecordPrintListener:49] -
RecordID [8200]
RecordTimestamp [174048****]
Source [{"sourceType": "MySQL", "version": "8.0.36"}]
RecordType [HEARTBEAT]

[2025-02-25 18:48:24.905] [INFO ] [...EtlRecordProcessor] [...DefaultRecordPrintListener:49] -
RecordID [8413]
RecordTimestamp [174048****]
Source [{"sourceType": "MySQL", "version": "8.0.36"}]
RecordType [UPDATE]
Schema info [{, recordFields= [{fieldName='id', ...}, {fieldName='name', ...}], databaseName='dtsdb', tableName='person', ...}]
Before image {[Field [id] [3] Field [name] [test1]]}
After image {[Field [id] [3] Field [name] [test2]]}

******
[2025-02-25 18:22:18.160] [INFO ] [subscribe-logMetricsReporter-1-thread-1] [log.metrics:184] - {"outCounts":0.0,"outBytes":0.0,"outRps":0.0,"outBps":0.0,"count":11.0,"inBytes":0.0,"DStoreRecordQueue":0.0,"inCounts":0.0,"inRps":0.0,"inBps":0.0,"__dt":174047893****,"DefaultUserRecordQueue":0.0}
[2025-02-25 18:22:22.002] [WARN ] [com.aliyun.dts.subscribe.clients.recordfetcher.KafkaRecordFetcher] [org.apache.kafka.clients.NetworkClient:780] - [Consumer clientId=consumer-dtsnd7u2n0625m****-1, groupId=dtsnd7u2n0625m****] Connection to node 1 (47.118.XXX.XXX/47.118.XXX.XXX:18001) could not be established. Broker may not be available.
[2025-02-25 18:22:22.509] [INFO ] [com.aliyun.dts.subscribe.clients.recordfetcher.KafkaRecordFetcher] [com.aliyun.dts.subscribe.clients.recordfetcher.ClusterSwitchListener:44] - Cluster not changed on update:5aPLLlDtTHqP8sKq-DZVfg
[2025-02-25 18:22:23.160] [INFO ] [subscribe-logMetricsReporter-1-thread-1] [log.metrics:184] - {"outCounts":0.0,"outBytes":0.0,"outRps":0.0,"outBps":0.0,"count":11.0,"inBytes":0.0,"DStoreRecordQueue":0.0,"inCounts":0.0,"inRps":0.0,"inBps":0.0,"__dt":1740478943160,"DefaultUserRecordQueue":0.0}
[2025-02-25 18:22:27.192] [WARN ] [com.aliyun.dts.subscribe.clients.recordfetcher.KafkaRecordFetcher] [org.apache.kafka.clients.NetworkClient:780] - [Consumer clientId=consumer-dtsnd7u2n0625m****1, groupId=dtsnd7u2n0625m****] Connection to node 1 (47.118.XXX.XXX/47.118.XXX.XXX:18001) could not be established. Broker may not be available.
[2025-02-25 18:22:27.618] [INFO ] [com.aliyun.dts.subscribe.clients.recordfetcher.KafkaRecordFetcher] [com.aliyun.dts.subscribe.clients.recordfetcher.ClusterSwitchListener:44] - Cluster not changed on update:5aPLLlDtTHqP8sKq-DZVfg
******

The client also logs consumption metrics every few seconds:

{"outCounts":0.0,"outBytes":0.0,"outRps":0.0,"outBps":0.0,"count":11.0,"inBytes":0.0,"DStoreRecordQueue":0.0,"inCounts":0.0,"inRps":0.0,"inBps":0.0,"__dt":174048044****,"DefaultUserRecordQueue":0.0}

Step 4: Process subscribed data

Edit the client code to handle the change records according to your application logic. Manage consumer offsets carefully to prevent data loss, minimize duplication, and support on-demand consumption starting points.

Query and reset offsets

Query the current offset from either of these locations:

• The localCheckpointStore file on the server running the SDK client

• The Data Consumption page of the change tracking channel in the DTS console

If you configured an external persistent storage using setUserRegisteredStore(new UserMetaStore()) in consumerContext.java, query the offset from that storage instead.

Offset lookup order on restart or retry

The SDK client searches for a saved offset in a specific order when it starts, retries internally, or restarts. The order depends on the mode:

ASSIGN mode — internal retry:

1. External storage configured with setUserRegisteredStore(new UserMetaStore())

2. localCheckpointStore file on the SDK client server

3. initCheckpoint passed in DTSConsumerSubscribeDemo.java

ASSIGN mode — client restart:

The behavior depends on setForceUseCheckpoint in consumerContext.java:

• true: Always uses the initCheckpoint value on every restart.

• false (default): Searches in this order:

  1. localCheckpointStore file on the SDK client server

  2. Offset saved on the DTS server (updated only after commit() is called)

  3. External storage configured with setUserRegisteredStore(new UserMetaStore())

SUBSCRIBE mode — internal retry or restart:

setForceUseCheckpoint has no effect in SUBSCRIBE mode.

1. External storage configured with setUserRegisteredStore(new UserMetaStore())

2. Offset saved on the DTS server (updated only after commit() is called)

3. initCheckpoint passed in DTSConsumerSubscribeDemo.java

4. Start offset of the DTS server's incremental data ingestion module

Store offsets persistently

Persistent offset storage prevents duplicate consumption after a module switchover. The following example stores offsets in a MySQL database.

1. Create a UserMetaStore class that extends AbstractUserMetaStore:

   public class UserMetaStore extends AbstractUserMetaStore {
   
       @Override
       protected void saveData(String groupID, String toStoreJson) {
           Connection con = getConnection();
           String sql = "insert into dts_checkpoint(group_id, checkpoint) values(?, ?)";
   
           PreparedStatement pres = null;
           ResultSet rs = null;
   
           try {
               pres = con.prepareStatement(sql);
               pres.setString(1, groupID);
               pres.setString(2, toStoreJson);
               pres.execute();
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               close(rs, pres, con);
           }
       }
   
       @Override
       protected String getData(String groupID) {
           Connection con = getConnection();
           String sql = "select checkpoint from dts_checkpoint where group_id = ?";
   
           PreparedStatement pres = null;
           ResultSet rs = null;
   
           try {
               pres = con.prepareStatement(sql);
               pres.setString(1, groupID);
               rs = pres.executeQuery();
   
               if (rs.next()) {
                   String checkpoint = rs.getString("checkpoint");
                   return checkpoint;
               }
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               close(rs, pres, con);
           }
           return null;
       }
   }

2. In consumerContext.java, register the store:

   consumerContext.setUserRegisteredStore(new UserMetaStore());

The persisted offset is stored as JSON. The timestamp value (for example, 1700709977) is the UNIX timestamp you pass directly to the SDK as initCheckpoint:

{"groupID":"dtsglg11d48230***","streamCheckpoint":[{"partition":0,"offset":577989,"topic":"ap_southeast_1_vpc_rm_t4n22s21iysr6****_root_version2","timestamp":170070****,"info":""}]}

Connection failures

Slow consumption

If data consumption is slower than expected, check the DStoreRecordQueue and DefaultUserRecordQueue values in the metrics output:

• `DStoreRecordQueue` stays at 0: The DTS server is pulling data slowly. Check server-side performance.

• `DefaultUserRecordQueue` stays at 512: The SDK client is processing data slowly. Reset the offset by adjusting initCheckpoint in the code as needed.