Data Transmission Service:Consume subscribed data using a Kafka client

Important notes

• If you use the demo provided in this topic and enable auto commit, some data might be lost because the commit operation could run before all data is consumed. Use manual commit to avoid this issue.

  Note

  If a failure prevents successful commits, the client resumes data consumption from the last recorded offset after restart. This may result in duplicate data. Filter duplicates manually.

• Data is stored in Avro serialized format. For details about the format, see the Record.avsc document.

  Warning

  If you do not use the Kafka client provided in this topic, deserialization (see DTS Avro deserialization example) might produce incorrect results. Verify data correctness yourself.

• For the offsetForTimes API, DTS uses seconds as the search unit, while native Kafka uses milliseconds.

• Network transient disconnections may occur on the change tracking server due to disaster recovery or other reasons. If you do not use the Kafka client provided in this topic, ensure your Kafka client supports network retry.

• If you use a native Kafka client to consume subscribed data, DTS might switch its incremental data ingestion module. This clears consumer offsets stored on the server in subscribe mode. Manually adjust the consumer offset to consume data as needed.

Kafka client workflow

Download the Kafka client demo code. For more details about using the code, see the Readme file in the demo.

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.0.0</version>
</dependency>

Procedure

This topic uses IntelliJ IDEA (Community Edition 2018.1.4 for Windows) as an example to describe how to run the client to consume data from a change tracking channel.

1. Create a new change tracking channel. For details, see Create an RDS MySQL change tracking channel, Create a PolarDB MySQL change tracking channel, or Create an Oracle change tracking channel.

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

3. Download the Kafka client demo code, then extract the file.

   Note

   Click , then select Download ZIP to download the file.

4. Open IntelliJ IDEA, then click Open.

5. In the dialog box that appears, go to the directory where you downloaded the Kafka client demo code. Navigate through the folders to find the project object model file: pom.xml.

   Navigate to kafkademo > subscribe_example-master > javaimpl, select pom.xml, then click OK.

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

7. In the IntelliJ IDEA interface, navigate through the folders to find and double-click the Kafka client demo file: NotifyDemoDB.java.

8. Set the values for the parameters in the NotifyDemoDB.java file.

   public static Properties getConfigs() {
       Properties properties = new Properties();
       // user password and sid for auth
       properties.setProperty(USER_NAME, "dtstest");
       properties.setProperty(PASSWORD_NAME, "xxx");
       properties.setProperty(SID_NAME, "dtsxxx");
       // kafka consumer group general same with sid
       properties.setProperty(GROUP_NAME, "dtsxxx");
       // topic to consume, partition is 0
       properties.setProperty(KAFKA_TOPIC, "cn_hangzhou_xxx");
       // kafka broker url
       properties.setProperty(KAFKA_BROKER_URL_NAME, "dts-cn-xxx.com:18001");
       // initial checkpoint for first seek(a timestamp to set, eg 1566180200 if you want (Mon Aug 19 10:03:21 CST 2019))
       properties.setProperty(INITIAL_CHECKPOINT_NAME, "1583307907");
       // if force use config checkpoint when start. for checkpoint reset
       properties.setProperty(USE_CONFIG_CHECKPOINT_NAME, "true");
       // use consumer assign or subscribe interface
       // when use subscribe mode, group config is required. kafka consumer group is enabled
       properties.setProperty(SUBSCRIBE_MODE_NAME, "assign");
       return properties;
   }

   Parameter	Description	How to obtain
   USER_NAME	Account for the consumer group. Warning If you do not use the client provided in this topic, set the username in the format <consumer group account>-<consumer group ID> (for example, dtstest-dtsae******bpv). Otherwise, the connection fails.	In the DTS console, click the target subscription instance ID, then click Consume Data. You can obtain the Consumer Group ID and Account information. Note You specified the password for the consumer group account when you created the consumer group.
   PASSWORD_NAME	Password for the account.
   SID_NAME	Consumer group ID.
   GROUP_NAME	Consumer group name. Set this parameter to the same value as the consumer group ID.
   KAFKA_TOPIC	Subscription topic of the change tracking channel.	In the DTS console, click the target subscription instance ID. On the Task Management page, you can obtain the Topic and network address information. Get the Subscription Topic value from the Basic Information section and the VPC endpoint (example format: xxx.aliyuncs.com:18003) from the Network section.
   KAFKA_BROKER_URL_NAME	Network address information for the data subscription channel. Note If the ECS instance where you deploy the Kafka client and the change tracking channel are in the classic network or the same VPC, use the internal endpoint for minimal network latency. We do not recommend using a public endpoint because of potential network instability.
   INITIAL_CHECKPOINT_NAME	Data timestamp to start consumption, in UNIX timestamp format (for example, 1592269238). Note Save the timestamp yourself to: Resume data consumption from the last consumed timestamp after a program interruption to prevent data loss. Specify the desired consumer offset when starting the subscription client to consume data as needed. If SUBSCRIBE_MODE_NAME is set to subscribe, the INITIAL_CHECKPOINT_NAME value takes effect only during the first startup of the subscription client.	The data timestamp must fall within the timestamp range of the subscription instance and must be converted to a UNIX timestamp. In the DTS subscription task list, check the Timestamp Range field for the subscription task to determine the valid range for INITIAL_CHECKPOINT_NAME. Note Use a search engine to find a UNIX timestamp converter.
   USE_CONFIG_CHECKPOINT_NAME	Default value is true. This forces consumption from the specified data timestamp to avoid losing received but unprocessed data.	None
   SUBSCRIBE_MODE_NAME	To run two or more Kafka clients under one consumer group, set this parameter to subscribe for all clients. The default value is assign, which disables this feature. Deploy only one client in this mode.	None

9. In the top menu of the IntelliJ IDEA interface, choose Run > Run to run the client.

   Note

   During the first run, the software takes time to automatically load and install dependencies.

Execution results

When you run the client, it successfully subscribes to data changes from the source database.

[2020-03-09 10:41:52,408] INFO [Consumer clientId=consumer-1, groupId=dts_xxx] Discovered coordinator xxx (id: xxx rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2020-03-09 10:41:57,203] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721711, offset: 1732521, info: 1583721711] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:41:57,571] INFO EtlRecordProcessor: haven't receive records from generator for  5s (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:02,203] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721721, offset: 1732539, info: 1583721721] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:07,204] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721726, offset: 1732544, info: 1583721726] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:12,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721731, offset: 1732548, info: 1583721731] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:17,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721736, offset: 1732554, info: 1583721736] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:22,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721741, offset: 1732559, info: 1583721741] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:27,206] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721746, offset: 1732569, info: 1583721746] (recordprocessor.EtlRecordProcessor)

You can also uncomment the log printing line in the NotifyDemoDB.java file (remove the // from line 25: //log.info(ret);), then run the client again to view detailed data change information.

FAQ

• Q: Why do I need to record the consumer offset myself?

  A: DTS records the consumer offset based on the time it receives the commit operation from the Kafka consumer client, which might differ from the actual consumption time. After an abnormal interruption of your application or Kafka consumer client, you can resume consumption from your recorded offset to avoid duplicate or missing data.

Mapping between MySQL field types and dataTypeNumber values

Mapping between Oracle field types and dataTypeNumber values

Mapping between PostgreSQL field types and dataTypeNumber values