Data Transmission Service:Use a Kafka client to consume tracked data

Usage notes

Disable auto commit. When auto commit is enabled, the Kafka client commits offsets on a periodic timer — independently of whether your application has finished processing those records. If the client restarts before processing completes, it resumes from the committed offset and skips unprocessed data. Disable auto commit and commit offsets manually after your application successfully processes each batch.

If a commit fails and the client restarts, consumption resumes from the last successfully committed checkpoint. Records between the failed commit and the restart may be delivered again — filter out duplicates in your application logic.

Additional notes:

• Tracked data is serialized in Avro format. See Record.avsc for the schema definition.

• If you are not using the demo client described in this topic, parse tracked data using the Avro schema and validate the parsed output.

• The offsetForTimes operation uses seconds as the time unit in DTS, compared to milliseconds in a standard Kafka client.

• The Kafka client may experience transient disconnections from the change tracking server — for example, during disaster recovery. If you are not using the demo client, implement automatic reconnection in your client.

• In subscribe mode, when DTS switches the incremental data collection module, the consumption checkpoint saved on the DTS server is removed. Specify a consumption checkpoint explicitly based on your requirements. For more information, see Use the SDK demo to consume tracked data and Manage the consumption checkpoint.

How it works

The Kafka client demo follows a three-stage pipeline:

Pre-image completeness depends on the source database type:

• Self-managed Oracle: Enable supplemental logging for all columns before consuming. Without it, the pre-image and post-image may be incomplete.

• Other sources: DTS does not guarantee pre-image completeness. Validate the pre-image in your application before use.

Prerequisites

Before you begin, ensure that you have:

• A change tracking instance. See Overview of change tracking scenarios.

• One or more consumer groups. See Create consumer groups.

• IntelliJ IDEA (this example uses Community Edition 2018.1.4 for Windows).

Run the Kafka client demo

1. Download the Kafka client demo package. If you are using Kafka client V2.0, open subscribe_example-master/javaimpl/pom.xml and change the version number to 2.0.0:

   Click and select Download ZIP to download the package.

   

2. Open IntelliJ IDEA and click Open.

   

3. Navigate to the directory where you extracted the demo, select the pom.xml file, and click Open as Project.

   

4. In the Project tool window, navigate to and double-click NotifyDemoDB.java.

5. Set the parameters in NotifyDemoDB.java. Authentication parameters (required) Connection parameters (required)

   Warning

   If you are not using this demo client, format USER_NAME as <Username>-<Consumer group ID> — for example, dtstest-dtsae******bpv. An incorrect format causes the connection to fail.

   Use an internal network endpoint when possible — it provides lower latency than a public endpoint. An internal network endpoint applies when the Elastic Compute Service (ECS) instance running your Kafka client is on the classic network or in the same virtual private cloud (VPC) as the change tracking instance.

   Parameter	Description	How to get the value
   USER_NAME	Username of the consumer group account.	In the DTS console, click the change tracking instance ID. In the left navigation pane, click Consume Data to view consumer group details.
   PASSWORD_NAME	Password of the consumer group account.	Set when you create the consumer group.
   SID_NAME	Consumer group ID.	In the DTS console, click the change tracking instance ID. In the left navigation pane, click Consume Data.
   GROUP_NAME	Consumer group name. Set this to the consumer group ID.	Same page as SID_NAME.

   Parameter	Description	How to get the value
   KAFKA_TOPIC	Name of the tracked topic.	In the DTS console, click the change tracking instance ID. On the Basic Information page, find the topic and network information.
   KAFKA_BROKER_URL_NAME	Endpoint of the change tracking instance.	Same page as KAFKA_TOPIC.

   Consumption behavior parameters

   Parameter	Description	Default
   INITIAL_CHECKPOINT_NAME	Starting consumption checkpoint as a UNIX timestamp (for example, 1592269238). The value must fall within the data range of the change tracking instance. View the data range in the Data Range column on the Change Tracking Tasks page. In subscribe mode, this value only takes effect on the first start.	—
   USE_CONFIG_CHECKPOINT_NAME	Forces the client to start from the specified INITIAL_CHECKPOINT_NAME. Set to true to prevent unprocessed records from being skipped after a restart.	true
   SUBSCRIBE_MODE_NAME	Run mode for the consumer group. Set to subscribe to allow multiple Kafka clients in the same consumer group. Keep the default assign for single-client deployments.	assign

   

6. From the top menu bar, choose Run > Run to start the client.

   The first run may take some time to download and install dependencies.

Verify the result

The following figure shows the Kafka client successfully consuming tracked data from the source database.

Manage the consumption checkpoint

When DTS switches the incremental data collection module — for example, during a failover — the consumption checkpoint stored on the DTS server is removed. Configure the client to detect this switch and reset the checkpoint to avoid data loss.

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

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();
        }
    }
}

try {
    // do some action
} catch (ClusterSwitchListener.ClusterSwitchException e) {
    reset();
}

// Reset the consumption checkpoint.
public reset() {
    long offset = kafkaConsumer.offsetsForTimes(timestamp);
    kafkaConsumer.seek(tp, offset);
}

FAQ

Why do I need to manage consumption checkpoints manually?

The checkpoint recorded by DTS reflects when DTS received the Kafka commit — not when your application finished processing the data. If your application or the client exits unexpectedly, specifying an accurate checkpoint lets you resume from exactly where processing stopped, preventing both data loss and unnecessary duplicate consumption.

Data type mappings

Use these tables to convert dataTypeNumber values in deserialized records to the corresponding database data types.

What's next

• Use the SDK demo to consume tracked data

• Create consumer groups

• Overview of change tracking scenarios