Data Transmission Service:Consume subscription data with Kafka

Notes

• When using the demo in this topic, auto commit can cause data loss because offsets may be committed before data processing is complete. To prevent this, we recommend manual commit.

  Note

  If a commit fails, the client restarts consumption from the last recorded consumption checkpoint. This may cause data duplication, which you must handle in your application logic.

• Data is stored using Avro serialization. For format details, see the Record.avsc file.

  Warning

  If you use a Kafka client other than the one provided in this topic, data may be parsed incorrectly during deserialization (DTS Avro deserialization example). You must verify the data.

• In DTS, the offsetForTimes API uses seconds as its time unit, while the native Kafka API uses milliseconds.

• The Data Subscription server may experience transient network disconnections due to events such as disaster recovery. If you do not use the Kafka client provided in this topic, ensure your client supports network retry.

• If you use a native Kafka client in subscribe mode, the consumption checkpoint stored on the server may be cleared when DTS switches its incremental data collection module. In this case, you must manually adjust the consumption checkpoint. If you need to use subscribe mode, we recommend using the DTS-provided SDK to consume Data Subscription data and manage the consumption checkpoint yourself. For more information, see Consume Data Subscription data by using the SDK and Manage the consumption checkpoint.

Kafka client runtime

Download the Kafka client demo. For usage instructions, see the demo's Readme file.

Procedure

Run the Kafka client demo in IntelliJ IDEA to consume data from a data subscription channel.

1. Create a data subscription channel. For more information, see Subscription solutions overview.

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

3. Download the Kafka client demo and unzip it.

   Note

   Click , then select Download ZIP to download the file.

4. Open IntelliJ IDEA and click Open.

5. In the dialog box, navigate to the directory where you unzipped the Kafka client demo. Expand the folders as shown in the following figure to find the Project Object Model (POM) file: pom.xml.

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

7. In the IntelliJ IDEA window, expand the folders, then find and double-click the Kafka client demo file: NotifyDemoDB.java.

8. Set the parameter values in the NotifyDemoDB.java file.

   Parameter	Description	How to obtain
   USER_NAME	The username for the consumer group. Warning If you use a different client, you must set the username in the <consumer group account>-<consumer group ID> format. Example: dtstest-dtsae******bpv.	In the DTS console, click the ID of the target subscription instance. In the left-side navigation pane, click Consume Data. On the page that appears, you can find the Consumer Group ID/Name and Account information for the consumer group. Note The password for the consumer group account is specified when you create the consumer group.
   PASSWORD_NAME	The password for the consumer group account.
   SID_NAME	The ID of the consumer group.
   GROUP_NAME	This value must match the consumer group ID.
   KAFKA_TOPIC	The subscription topic for the data subscription channel.	In the DTS console, click the ID of the target subscription instance. On the Basic Information page, you can find the Topic and Network information.
   KAFKA_BROKER_URL_NAME	The network address for the data subscription channel. Note If your Kafka client and data subscription channel are in the same classic network or VPC, use the internal IP address to minimize network latency. We do not recommend using a public endpoint because of potential network instability.
   INITIAL_CHECKPOINT_NAME	The time to start data consumption, specified as a UNIX timestamp. Example: 1592269238. Note Saving the timestamp allows you to: Resume consumption from the last saved timestamp after an interruption to prevent data loss. Start consuming data from a specific point in time. If SUBSCRIBE_MODE_NAME is set to subscribe, the value of INITIAL_CHECKPOINT_NAME takes effect only when the subscription client starts for the first time.	The consumption start time must be within the data range of the subscription instance and must be converted to a UNIX timestamp. Note You can view the data range of the subscription instance in the Data Range column of the subscription task list. You can use a search engine to find a UNIX timestamp converter.
   USE_CONFIG_CHECKPOINT_NAME	The default value is true, which forces the client to start from the specified time. This prevents losing received but unprocessed data.	None
   SUBSCRIBE_MODE_NAME	To run two or more Kafka clients in the same consumer group, set this parameter to subscribe on all clients. The default value is assign. In this mode, this feature is disabled, and we recommend deploying only one client.	None

9. From the top menu bar, select Run > Run to run the client.

   Note

   When you run the application for the first time, IntelliJ IDEA may take some time to automatically download and install the required dependency packages.

Results

The client successfully subscribes to change data from the source database.

To view detailed change data, remove the // from //log.info(ret); on line 25 in the NotifyDemoDB.java file and run the client again.

FAQ

• Q: Why do I need to manually track my client's consumption checkpoint?

  A: The consumption checkpoint recorded by DTS indicates when DTS receives the commit from the Kafka client, not when your application finishes processing the data. If your application or Kafka client stops unexpectedly, providing your own recorded checkpoint lets you resume consumption from the exact point of interruption, preventing both duplicate data and data loss.

Manage checkpoint

1. Configure the subscription client to listen for cluster switches in the DTS data collection module.

   To do this, register a ClusterSwitchListener as a consumer interceptor by setting the following property for the consumer:

   properties.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, ClusterSwitchListener.class.getName());

   The following code shows a sample implementation of ClusterSwitchListener:

   public class ClusterSwitchListener implements ClusterResourceListener, ConsumerInterceptor {
       private final static Logger LOG = LoggerFactory.getLogger(ClusterSwitchListener.class);
       private ClusterResource originClusterResource = null;
       private ClusterResource currentClusterResource = null;
   
       public ConsumerRecords onConsume(ConsumerRecords records) {
           return records;
       }
   
   
       public void close() {
       }
   
       public void onCommit(Map offsets) {
       }
   
   
       public void onUpdate(ClusterResource clusterResource) {
           synchronized (this) {
               originClusterResource = currentClusterResource;
               currentClusterResource = clusterResource;
               if (null == originClusterResource) {
                   LOG.info("Cluster updated to " + currentClusterResource.clusterId());
               } else {
                   if (originClusterResource.clusterId().equals(currentClusterResource.clusterId())) {
                       LOG.info("Cluster not changed on update:" + clusterResource.clusterId());
                   } else {
                       LOG.error("Cluster changed");
                       throw new ClusterSwitchException("Cluster changed from " + originClusterResource.clusterId() + " to " + currentClusterResource.clusterId()
                               + ", consumer require restart");
                   }
               }
           }
       }
   
       public boolean isClusterResourceChanged() {
           if (null == originClusterResource) {
               return false;
           }
           if (originClusterResource.clusterId().equals(currentClusterResource.clusterId())) {
               return false;
           }
           return true;
       }
   
       public void configure(Map<String, ?> configs) {
       }
   
       public static class ClusterSwitchException extends KafkaException {
           public ClusterSwitchException(String message, Throwable cause) {
               super(message, cause);
           }
   
           public ClusterSwitchException(String message) {
               super(message);
           }
   
           public ClusterSwitchException(Throwable cause) {
               super(cause);
           }
   
           public ClusterSwitchException() {
               super();
           }
   
       }

2. Handle the cluster switch event from the DTS data collection module.

   When a ClusterSwitchException is caught, reset the initial checkpoint for the next subscription to the timestamp of the last successfully consumed record. The following code snippet provides an example:

   try{
      //do some action
   } catch (ClusterSwitchListener.ClusterSwitchException e) {
      reset();
   }
   
   // Reset the checkpoint.
   public reset() {
     long offset = kafkaConsumer.offsetsForTimes(timestamp);
     kafkaConsumer.seek(tp,offset);
   }

   Note

   For a sample implementation, see KafkaRecordFetcher.

Data type and dataTypeNumber mapping

Consume subscription data by using Python

If your consumption application is written in Python, you can use the kafka-python and avro libraries to consume and deserialize Avro-formatted data from the data subscription channel. The following example has been verified with kafka-python 3.0.7 and avro 1.12.1.

1. Install the required Python libraries.

   pip install kafka-python avro

2. Download the Avro schema file Record.avsc from the avro directory of the Kafka client demo repository. This is the same schema file referenced in the Notes section of this topic.

3. Use the following sample code to create a KafkaConsumer, load the Avro schema, and deserialize the change data:

   from kafka import KafkaConsumer
   import avro.schema
   from avro.io import DatumReader, BinaryDecoder
   
   schema = avro.schema.parse(open("Record.avsc", "rb").read())
   record_schema = schema.schemas[13]
   
   consumer = KafkaConsumer(
       "<topic>",
       bootstrap_servers="<DTS access point:port>",
       group_id="<consumer group ID>",
       enable_auto_commit=False,
       security_protocol="SASL_SSL",
       sasl_mechanism="PLAIN",
       sasl_plain_username="<consumer group account>",
       sasl_plain_password="<password>",
       ssl_ca_location="ca.pem",
   )
   
   for msg in consumer:
       reader = DatumReader(record_schema)
       record = reader.read(BinaryDecoder(msg.value))
       if record.get("operation") == "HEARTBEAT":
           continue
       print(record)

   The sample code does the following:

   • Configures the KafkaConsumer to connect by using SASL_SSL transport with PLAIN authentication.

   • Loads Record.avsc and selects the Record schema definition (schemas[13]) to deserialize each message with DatumReader and BinaryDecoder.

   • Filters out HEARTBEAT messages, which are keepalive messages and do not contain change data.

4. Replace the following placeholders with your actual values. You can obtain these values by using the same methods described in the Procedure section's parameter table (KAFKA_TOPIC, KAFKA_BROKER_URL_NAME, SID_NAME, USER_NAME, and PASSWORD_NAME):

   Placeholder	Description
   <topic>	The subscription topic for the data subscription channel. Corresponds to KAFKA_TOPIC.
   <DTS access point:port>	The network address of the data subscription channel. Corresponds to KAFKA_BROKER_URL_NAME.
   <consumer group ID>	The ID of the consumer group. Corresponds to SID_NAME.
   <consumer group account>	The username for the consumer group. Corresponds to USER_NAME.
   <password>	The password for the consumer group account. Corresponds to PASSWORD_NAME.