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

This topic explains how to 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

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