Change Data Capture with DSE

Capture schema changes in your DSE tables and pass them to Apache Pulsar® with DSE Change Data Capture (CDC). This doc will guide you through installing, configuring, and using CDC with DSE in a VM-based deployment.

This installation requires the following. Latest version artifacts are available here. Use cdc apache cassandra?color=green&display name=tag for the latest version.

  • DSE installation

  • DSE 6.9 CDC Agent - use agent-c4-<version>-all.jar

Prerequisites

  • Running DSE cluster

Installing and configuring

  1. Download the Pulsar tarball and set up a Pulsar cluster. This example uses Pulsar standalone mode, but you can also use our helpful Ansible Scripts.

    cd $PULSAR_HOME
    bin/pulsar standalone

    We recommend using the latest CDC agent version (at least version 1.0.4+) to support C* collection data types.

  2. Configure the Cassandra-env.sh file as follows:

    export CDC_PULSAR_SERVICE_URL="<pulsar_broker_service_url>" (e.g. pulsar://<pulsar_server_ip>:6650)
    
    # needed when Pulsar JWT authentication is enabled
    export CDC_PULSAR_AUTH_PLUGIN_CLASS_NAME="org.apache.pulsar.client.impl.auth.AuthenticationToken"
    export CDC_PULSAR_AUTH_PARAMS="file://</path/to/token/file>"
    
    # needed when Pulsar TLS encryption is enabled
    export CDC_TLS_TRUST_CERTS_FILE_PATH=”</path/to/trusted/cert/file>”
    
    # DSE CDC
    JVM_OPTS="$JVM_OPTS -javaagent:/home/automaton/cdc104/agent-dse4-<version>-all.jar"
  3. Restart DSE to apply the changes.

    For CDC agent versions after 1.0.3, the CDC agent Pulsar connection parameters are provided as system environment variables (see DSE CDC in the example above).

    For CDC agent versions before 1.0.3, the CDC agent Pulsar connection parameters are also provided as extra JVM options, as below:

    export JVM_EXTRA_OPTS="-javaagent:/path/to/agent-c4-<version>-all.jar=pulsarServiceUrl=pulsar://pulsar:6650"
  4. Set the cassandra.yaml configuration:

    cdc_enabled:true
    cdc_raw_directory:/var/lib/cassandra/cdc_raw
    commitlog_sync_period_in_ms:2000
    cdc_total_space_in_mb:4096
  5. Restart DSE and verify your logs are similar to this example system.log file below. CDC agent started indicates your CDC agent has started properly.

    INFO  [main] 2022-04-11 18:47:06,302  Agent.java:60 - Starting CDC agent, cdc_raw_directory=/var/lib/cassandra/cdc_raw
    INFO  [main] 2022-04-11 18:47:06,331  AgentConfig.java:526 - maxInflightMessagesPerTask=16384, sslEnabledProtocols=TLSv1.2,TLSv1.1,TLSv1, cdcWorkingDir=/usr/share/dse/data/cdc, pulsarMaxPendingMessagesAcrossPartitions=50000, pulsarMaxPendingMessages=1000, sslTruststorePath=null, cdcPollIntervalMs=60000, pulsarAuthParams=null, sslHostnameVerificationEnable=false, errorCommitLogReprocessEnabled=false, sslTruststorePassword=null, tlsTrustCertsFilePath=null, sslKeystorePath=null, sslKeystorePassword=null, sslAllowInsecureConnection=false, cdcConcurrentProcessors=-1, pulsarServiceUrl=pulsar://10.101.32.213:6650, pulsarKeyBasedBatcher=false, sslTruststoreType=JKS, pulsarBatchDelayInMs=-1, topicPrefix=events-, sslCipherSuites=null, pulsarAuthPluginClassName=null, sslProvider=null, useKeyStoreTls=false
    INFO  [main] 2022-04-11 18:47:06,433  Agent.java:92 - CDC agent started

Deploy Pulsar Connector

  1. Deploy the Pulsar Cassandra Source Connector (CSC) for each CDC-enabled table. In the below example, we’re configuring the source connector with parameters passed in --source-config.

    The outputFormat source configuration controls the format of messages on the data topic.

    • key-value-avro - Default behavior. Key and value are encoded separately in AVRO format.

    • key-value-json - Key and value are encoded separately in JSON format.

    • json - Key and value are encoded together in a single JSON object. Key field is populated with a JSON string representing key fields.

      • Key-value-avro

      • Key-value-json

      • JSON

      $ pulsar-admin source create \
        --name <csc_connector_name> \
        --archive /path/to/pulsar-cassandra-source-<version>.nar \
        --tenant public \
        --namespace default \
        --destination-topic-name persistent://public/default/data-<keyspace>.<table> \
        --parallelism 1 \
        --source-config {
        	"events.topic": "persistent://public/default/events-<keyspace>.<table>",
        	"keyspace": "<C*_keyspace>",
        	"table": "<C*_table>",
        	"contactPoints": "<C*_contact_point_list>",
        	"port": "9042",
        	"loadBalancing.localDc": "<C*_DC_name>",
          "outputFormat": "key-value-avro"
        }
      $ pulsar-admin source create \
        --name <csc_connector_name> \
        --archive /path/to/pulsar-cassandra-source-<version>.nar \
        --tenant public \
        --namespace default \
        --destination-topic-name persistent://public/default/data-<keyspace>.<table> \
        --parallelism 1 \
        --source-config {
        	"events.topic": "persistent://public/default/events-<keyspace>.<table>",
        	"keyspace": "<C*_keyspace>",
        	"table": "<C*_table>",
        	"contactPoints": "<C*_contact_point_list>",
        	"port": "9042",
        	"loadBalancing.localDc": "<C*_DC_name>",
          "outputFormat": "key-value-json"
        }
      $ pulsar-admin source create \
        --name <csc_connector_name> \
        --archive /path/to/pulsar-cassandra-source-<version>.nar \
        --tenant public \
        --namespace default \
        --destination-topic-name persistent://public/default/data-<keyspace>.<table> \
        --parallelism 1 \
        --source-config {
        	"events.topic": "persistent://public/default/events-<keyspace>.<table>",
        	"keyspace": "<C*_keyspace>",
        	"table": "<C*_table>",
        	"contactPoints": "<C*_contact_point_list>",
        	"port": "9042",
        	"loadBalancing.localDc": "<C*_DC_name>",
          "outputFormat": "json"
        }
  2. Verify the CSC connector is deployed.

    pulsar-admin source list ["<csc_connector_name>"]
  3. Check the CSC connector log file at <$PULSAR_HOME>/logs/functions/public/default/<csc_connector_name> for errors.

Verify end-to-end operation

Now that Pulsar, DSE, CDC, and the CSC connector are installed and verified to be operational, we can monitor the Pulsar data topic for the CDC-enabled C* table.

Any captured CDC events from the C* table will be reflected in the command line output of the following command:

pulsar-client consume -s mysub -st auto_consume -n 0 persistent://public/default/data-<keyspace>.<table>

Was this helpful?

Give Feedback

How can we improve the documentation?

© 2024 DataStax | Privacy policy | Terms of use

Apache, Apache Cassandra, Cassandra, Apache Tomcat, Tomcat, Apache Lucene, Apache Solr, Apache Hadoop, Hadoop, Apache Pulsar, Pulsar, Apache Spark, Spark, Apache TinkerPop, TinkerPop, Apache Kafka and Kafka are either registered trademarks or trademarks of the Apache Software Foundation or its subsidiaries in Canada, the United States and/or other countries. Kubernetes is the registered trademark of the Linux Foundation.

General Inquiries: +1 (650) 389-6000, info@datastax.com