Data Transmission Service:Consume subscription data using SDK sample code

Procedure

The following procedure shows how to run the SDK sample code to consume Data Subscription data in IntelliJ IDEA (Community Edition 2020.1 for Windows).

1. Create a Data Subscription instance. For more information, see Create a Data Subscription channel for an ApsaraDB RDS for MySQL instance, Create a Data Subscription channel for a PolarDB for MySQL cluster, or Create a Data Subscription channel for an Oracle database.

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

   Important

   When you consume Data Subscription data, you must call the commit method of DefaultUserRecord to commit checkpoints. Failure to do so may result in duplicate data consumption.

3. Use the SDK sample code based on your business requirements.

   • Use the new Data Subscription SDK package (recommended)

     1. Open IntelliJ IDEA, and then click Create New Project to create a project for your application.

     2. In the project, find the project object model (POM) file: pom.xml.

     3. Add the following dependency to the pom.xml file:

        <dependency>
            <groupId>com.aliyun.dts</groupId>
            <artifactId>dts-new-subscribe-sdk</artifactId>
            <version>{dts_new_sdk_version}</version>
        </dependency>

        Note

        You can find the latest Maven dependency on the dts-new-subscribe-sdk page.

     4. For more information about how to use the new subscription SDK, see Use the sample code.

   • Use a customized version of the new Data Subscription SDK

     1. Download the SDK sample code package, and then decompress it.

        Note

        Click and select Download ZIP to download the package.

     2. Navigate to the decompressed SDK sample code directory. Use a text editor to open the pom.xml file and update the Data Subscription SDK to the latest version.

        Important

        You can obtain the latest version of the Data Subscription SDK from the Maven website. For more information, see the Maven page for the Data Subscription SDK.

     3. Open IntelliJ IDEA and click Open or Import.

     4. In the dialog box that appears, go to the decompressed SDK sample code directory, expand the folders, and find the pom.xml file.

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

     6. In IntelliJ IDEA, expand the folders. Based on the usage mode of the SDK client, select and double-click the corresponding Java file: DTSConsumerAssignDemo.java or DTSConsumerSubscribeDemo.java.

        Note

        DTS supports the following SDK client usage modes:

        • ASSIGN mode: To ensure the global order of messages, DTS assigns only one partition (partition 0) to each subscription topic. When using an SDK client in ASSIGN mode, we recommend starting only one client.

        • SUBSCRIBE mode: To ensure the global order of messages, DTS assigns only one partition (partition 0) to each subscription topic. If you use the SDK client in SUBSCRIBE mode, you can start multiple SDK clients in a consumer group for disaster recovery. If the active client fails, another SDK client is automatically assigned to partition 0 to resume consumption.

4. Set the required parameters in the Java file.

   

   Table 1. Required parameters

   Parameter	Description	Source
   brokerUrl	The endpoint and port number of the Data Subscription instance. Note If the ECS instance running the SDK client and the Data Subscription instance are in the same classic network or Virtual Private Cloud (VPC), we recommend that you use the internal endpoint for the subscription to minimize network latency. We do not recommend using a public endpoint because of potential network instability.	In the DTS console, click the ID of the target Data Subscription instance. On the Basic Information page, you can obtain the endpoint and port number in the Network section.
   topic	The subscription topic for the instance.	In the DTS console, click the ID of the target Data Subscription instance. On the Basic Information page, you can obtain the Topic in the Basic Information section.
   sid	The ID of the consumer group.	In the DTS console, click the ID of the target Data Subscription instance and then click Consume Data. You can obtain the Consumer Group ID and the Account of the consumer group. Note The password for the consumer group username is specified when you create the consumer group.
   userName	The username for the consumer group. Warning If you do not use the client provided in this topic, you must set the username in the <Username>-<Consumer Group ID> format. Example: dtstest-dtsae******bpv. Otherwise, the connection fails.
   password	The password for the username.
   initCheckpoint	The consumption checkpoint, specified as a UNIX timestamp, from which the SDK client begins consuming data. Example: 1620962769. Note You can use consumption checkpoint information in the following scenarios: To resume consumption and prevent data loss after an application interruption, pass the last known consumption checkpoint. When starting the client, you can pass a specific consumption checkpoint to consume data from a desired position.	The consumption checkpoint must be within the data range of the Data Subscription instance (as shown in the figure) and must be converted to a UNIX timestamp. Note You can use a search engine to find a UNIX timestamp converter.
   ConsumerContext.ConsumerSubscribeMode subscribeMode	The usage mode of the SDK client. Valid values: ConsumerContext.ConsumerSubscribeMode.ASSIGN: ASSIGN mode. Only one SDK client in a consumer group can consume Data Subscription data. ConsumerContext.ConsumerSubscribeMode.SUBSCRIBE: SUBSCRIBE mode. You can start multiple SDK clients in the same consumer group for disaster recovery.	N/A

5. In the top navigation bar of IntelliJ IDEA, choose Run > Run to run the client.

   Note

   The first time you run the client, it may take some time to automatically load and install the required dependencies.

   • The following figure shows the result. The client successfully consumes change data from the source database.

   • The SDK client periodically aggregates and displays statistics about data consumption, including the total count and volume of records sent and received, and requests per second (RPS).

     Table 2. 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. Unit: bytes.
     outRps	The number of requests per second (RPS) at which the SDK client consumes data.
     outBps	The data consumption rate of the SDK client, in bits per second (bps).
     inBytes	The total volume of data sent by the DTS server. Unit: bytes.
     DStoreRecordQueue	The size of the internal data cache queue for incoming records from the DTS server.
     inCounts	The total number of data records sent by the DTS server.
     inRps	The RPS at which the DTS server sends data.
     __dt	The timestamp when the SDK client receives the data. Unit: milliseconds.
     DefaultUserRecordQueue	The size of the data queue holding records ready for the consumer application to process.

Save and query consumption checkpoints

To start or resume data consumption (for example, on first launch, restart, or internal retry), the SDK client requires a consumption checkpoint. The following table describes how to manage and query checkpoints in different scenarios to prevent data loss, minimize duplicate consumption, and enable on-demand consumption.

Persist consumption checkpoints

During a disaster recovery event in the incremental data collection module (especially in SUBSCRIBE mode), the new module does not retain the client’s latest consumption checkpoint. The client may resume from an older checkpoint, resulting in duplicate consumption of historical data. For example, before the switch, the checkpoint range of the old module is from 08:00:00 on November 11, 2023 to 08:00:00 on November 12, 2023, and the client checkpoint is 08:00:00 on November 12, 2023. After the switch, the checkpoint range of the new module is from 10:00:00 on November 08, 2023 to 08:01:00 on November 12, 2023. The client starts from the new module’s start checkpoint (10:00:00 on November 08, 2023), causing duplicate consumption.

To avoid duplicate consumption in this scenario, configure a persistent checkpoint store on the client. The following example provides one possible implementation that you can adapt to your requirements.

1. Create a UserMetaStore class that extends AbstractUserMetaStore.

   For example, to store checkpoint information in a MySQL database, use the following Java code:

   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);
   						ResultSet rs = pres.executeQuery()
                             
               String checkpoint = rs.getString("checkpoint");
             
               return checkpoint;
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               close(rs, pres, con);
           }
       }
   }
   

2. In the consumerContext.java file, configure the external storage medium by using the setUserRegisteredStore(new UserMetaStore()) method.

FAQ

• How can I resolve connection issues with a Data Subscription instance?

  Troubleshoot the issue based on the error message. For more information, see Troubleshooting.

• In what format are consumption checkpoints persisted?

  Persisted consumption checkpoint data is stored in JSON format. The persisted checkpoint is a UNIX timestamp that can be passed directly to the SDK. In the following example response, the value 1700709977 for the "timestamp" key is the persisted consumption checkpoint.

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

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