This topic describes how to use the demo code of a Kafka client to consume tracked data. The change tracking feature of the new version allows you to consume tracked data by using a Kafka client (V0.11 to V2.0).
- If you enable auto commit when you use the change tracking feature, some data may be committed before it is consumed. This results in data loss. We recommend that you manually commit data.
Note If data fails to be committed due to a fault, you can restart the client to continue consuming data from the last recorded consumer offset. However, duplicate data may be generated during this period. You must manually filter out the duplicate data.
- Data is serialized and stored in the Avro format. For more information, see Record.avsc.
Warning If you are not using the Kafka client that is described in this topic, you must parse the tracked data based on the Avro schema.
- The search unit is second when DTS calls the
offsetForTimesoperation. The search unit is millisecond when a native Kafka client calls this operation.
Run the Kafka client
|Step||File or directory|
|1. Use the native Kafka consumer to obtain incremental data from the change tracking instance.||subscribe_example-master/javaimpl/src/main/java/recordgenerator/|
|2. Deserialize the image of the incremental data, and obtain the pre-image, post-image , and other attributes.
|3. Convert the dataTypeNumber values in the deserialized data into MySQL data types.
Note For more information, see Mappings between MySQL data types and dataTypeNumber values.
The following steps show how to run the Kafka client to consume tracked data. IntelliJ IDEA (Community Edition 2018.1.4 Windows) is used in this example.
- Create a change tracking task. For more information, see Track data changes from an ApsaraDB RDS for MySQL instance (new), Track data changes from a PolarDB for MySQL cluster, or Track data changes from a self-managed Oracle database.
- Create one or more consumer groups. For more information, see Create consumer groups.
- Download the demo code of the Kafka client and decompress the package.
- Open IntelliJ IDEA. In the window that appears, click Open.
- In the dialog box that appears, go to the directory where the downloaded demo code resides. Find the pom.xml file.
- In the dialog box that appears, select Open as Project.
- On the IntelliJ IDEA page, expand folders to find the demo file of the Kafka client, and double-click the file. The file name is NotifyDemo.java.
- Set the parameters in the NotifyDemo.java file.
Parameter Description Method to obtain USER_NAME The username of the consumer group.Warning If you are not using the Kafka client that is described in this topic, you must specify this parameter in the following format:
<Username>-<Consumer group ID>, for example,
dtstest-dtsae******bpv. Otherwise, the connection fails.
In the DTS console, click the instance ID, and then click Data Consume. You can obtain the consumer group ID and the corresponding username.Note When you create a consumer group, the password of the consumer group is automatically specified. PASSWORD_NAME The password of the consumer group. SID_NAME The ID of the consumer group. GROUP_NAME The name of the consumer group. Set this parameter to the consumer group ID. KAFKA_TOPIC The topic of the change tracking instance. In the DTS console, click the instance ID. On the View Task Settings page, you can obtain the tracked topic, network address, and port number. KAFKA_BROKER_URL_NAME The network address and port number of the change tracking instance.Note If you track data changes over internal networks, the network latency is minimal. This is applicable if the Elastic Compute Service (ECS) instance where you deploy the Kafka client belongs to the classic network or the same VPC as the change tracking instance. INITIAL_CHECKPOINT_NAME The consumer offset of consumed data. The value is a UNIX timestamp, for example, 1592269238.Note You must save the consumer offset.
- If the consumption process is interrupted, you can specify the consumer offset on the change tracking client to resume data consumption. This allows you to prevent against data loss.
- When you start the change tracking client, you can specify the consumer offset to consume data on demand.
The consumer offset of consumed data must be within the data range of the change tracking instance, as shown in the following figure. The consumer offset must be converted to a UNIX timestamp.Note
- For more information about how to view the data range, see View tracked data changes.
- You can use a search engine to obtain a UNIX timestamp converter.
USE_CONFIG_CHECKPOINT_NAME Default value: true. The default value indicates that the client is forced to consume data from the specified consumer offset. This allows you to retain the data that is received but not processed. N/A SUBSCRIBE_MODE_NAME You can run two Kafka clients for a consumer group to implement disaster recovery. To use this feature, deploy two Kafka clients and set the value of the SUBSCRIBE_MODE_NAME parameter to subscribe.
Default value: assign. The default value indicates that only one Kafka client is deployed.
- In the top menu bar of IntelliJ IDEA, choose
to run the client. Note When you run IntelliJ IDEA for the first time, it requires some time to load and install the relevant dependency.
Running result of the Kafka client
The following figure shows that the Kafka client can track data changes from the source database.
You can delete the // characters from the
//log.info(ret); string in line 25 of the
NotifyDemo.java file. Then, run the client again to view the data change information.
- Q: Why do I need to record the consumer offset of the Kafka client?
A: The consumer offset recorded by DTS is the time when DTS receives a commit operation from the Kafka client. The recorded consumer offset may be different from the actual consumption time. If a business application or the Kafka client is unexpectedly interrupted, you can specify an accurate consumer offset to continue data consumption. This prevents against data loss or duplicate data consumption.
Mappings between MySQL data types and dataTypeNumber values
|MySQL data type||Value of dataTypeNumber|