index.adoc

About {product-short}

{product} is open-source software (OSS) that sends {cass-short} mutations for tables having Change Data Capture (CDC) enabled to IBM Elite Support for {pulsar} or your own self-managed {pulsar-reg}) deployment, which in turn can write the data to platforms such as Elasticsearch® or Snowflake®.

Key Features

• Supports {cass} version 3.11 or later, {cass} version 4.0 or later, and {dse} version 6.8.16 or later

• Supports IBM Elite Support for {pulsar} (formerly {company} Luna Streaming) and self-managed {pulsar} version 2.8.1 or later

• De-duplicates updates from multiple replicas

• Propagates {cass-short} schema change to the built-in {pulsar-short} schema registry

• Supports AVRO message format

Architecture

Other than the prerequisite {cass-short} and {pulsar-short} clusters, {product-short} has two components:

• {cdc_agent_first}, which is an event producer deployed as a JVM agent on each {cass-short} data node

• {csc_pulsar_first}, which is a source connector deployed in your {pulsar-short} cluster

Figure 1. General {product-short} architecture

Since version 3.0, {cass-short} has included a change data capture (CDC) feature. The CDC feature can be enabled on the table level by setting the table property cdc=true, after which any commit log containing data for a CDC-enabled table is moved to the CDC directory specified in cassandra.yaml on discard/flush (default: cdc_raw). Since {cass-short} 4.0 (and back ported in {cass-short} 3.11 and later), the {cass-short} CDC feature has been improved such that commit logs can be synced to the cdc_raw directory periodically (by default, 10 seconds). This improvement makes the {cass-short} CDC feature near real-time.

When CDC is enabled:

1. The change agent running on each node reads the {cass-short} commitlog files in the cdc_raw directory and emits a message for each {cass-short} mutation on a CDC-enabled table to a per-table events topic.

2. The {cdc_agent} processes these events and fetches the mutated {cass-short} row.

3. The {cass-short} Source Connector writes the row into the data topic with necessary processing tasks that make sure the most recent state of the {cass-short} table is replicated into the data topic correctly (no duplication, right order, etc.)

4. Once the change agent processes all mutations in the commit log, it deletes the file from the cdc_raw directory.

{cdc_agent} is tolerant of failures when processing the commit logs in the cdc_raw directory. It maintains a processing offset for each commit log. If the {cdc_agent} restarts, it picks up where it left off using the recorded offset value.

Table 1. Event data published to the data topic for each write to a CDC-enabled {cass-short} table.

Type	Event Data
insert	Key set to primary key of the row, value set to all column values
update	Key set to primary key of the row, value set to all column values
delete	Key set to primary key of the row, value set to null

The {csc_pulsar} updates the schema registry to dynamically reflect the {cass-short} table schema. You can then deploy various sink connectors to replicate data into the backends of your choice. For more, see {pulsar-short} built-in sink connectors.

Sink connectors processing messages from the data topic should interpret an event with a null value as a delete. For example, with the {pulsar-short} Elasticsearch connector, you need to set nullValueAction to DELETE.

The change agent runs on all {cass-short} data nodes. This means that the agent processes the original write plus its replicas. To minimize the number of duplicate events that end up in the data topic, the {cdc_agent} maintains an in-memory de-duplication cache. For each update to the table, an MD5 digest is calculated to de-duplicate the updates from the replicas.

Table 2. Change Agent deployment matrix

{cass-short} version	Self-managed {pulsar} or IBM Elite Support for {pulsar} (formerly {company} Luna Streaming) agent
{cass-short} 3.x	{product-repo}/tree/master/agent-c3[agent-c3]
{cass-short} 4.x	{product-repo}/tree/master/agent-c4[agent-c4]
{dse-short} 6.8.16 or later	{product-repo}/tree/master/agent-dse4[agent-dse4]

Compatibility

Supported streaming platforms

• IBM Elite Support for {pulsar} (formerly {company} Luna Streaming) 2.8 and later (current version is {luna_version})

• Self-managed {pulsar} version 2.8.1 and later

Connectors

• {product-repo}/tree/master/connector[Connector for self-managed {pulsar} or IBM Elite Support for {pulsar}]

Supported databases

• Open-source {cass-reg} 3.11.x and 4.x

• {dse} 6.8.16 or later

Supported {cass-short} data structures

The following CQL data types are encoded as AVRO logical types:

• ascii (string)

• bigint (long)

• blob(bytes)

• boolean (boolean)

• Collection types:

  • list (array)

  • set (array)

  • map (map)

• date (date)

• decimal (cql_decimal)

• double (double)

• duration (cql_duration)

• float (float)

• inet (string)

• int (int)

• smallint (int)

• text (string)

• time (time-micros)

• timestamp (timestamp-millis)

• timeuuid (uuid)

• tinyint (int)

• tuple (record)

• User Defined Types (record)

• uuid (uuid)

• varint (cql_varint)

If using the key-value-json output format, the supported {cass-short} types are the same as AVRO. The output is an exact schema with logical types, but with a JSON schema type.

{cass-short} static columns are supported in the following ways:

• On row-level updates, static columns are included in the message value.

• On partition-level updates, the clustering keys are null in the message key, and the message value only has static columns on insert/update operations.

For data types that aren’t supported, columns using those data types are omitted from the events sent to the data topic. If a row update contains both supported and unsupported data types, the event includes only columns with supported data types.

Limitations

{product-short} has the following limitations:

• Doesn’t manage table truncates. Don’t use the TRUNCATE TABLE_NAME command.

• Doesn’t sync data available before starting the {cdc_agent}.

• Doesn’t replay logged batches.

• Doesn’t manage time-to-live (TTL).

• Doesn’t support range deletes.

• CQL column names cannot match {pulsar-short} primitive type names, such as the following:

  Table 3. {pulsar-short} primitive types

  Primitive type	Description
  BOOLEAN	A binary value
  INT8	A 8-bit signed integer
  INT16	A 16-bit signed integer
  INT32	A 32-bit signed integer
  INT64	A 64-bit signed integer
  FLOAT	A single precision (32-bit) IEEE 754 floating-point number
  DOUBLE	A double-precision (64-bit) IEEE 754 floating-point number
  BYTES	A sequence of 8-bit unsigned bytes
  STRING	A Unicode character sequence
  TIMESTAMP (DATE, TIME)	A logic type represents a specific instant in time with millisecond precision. It stores the number of milliseconds since January 1, 1970, 00:00:00 GMT as an INT64 value

Manage schema updates on topics

Schema registry updates on a {pulsar-short} topic are controlled by the is-allow-auto-update-schema option:

• true allows the broker to register a new schema for a topic and connect the producer if the schema isn’t registered.

• false rejects the producer’s connection to the broker if the schema isn’t registered.

To ensure the {csc_pulsar} can automatically update the schema on the {pulsar-short} topic, set the option to true. For more, see Schema Auto-Update.

Deploy on multiple {cass-short} datacenters

In a multi-datacenter {cass-short} configuration, enable CDC and install the change agent in only one datacenter. To ensure the data sent to all datacenters are delivered to the data topic, make sure to configure replication to the datacenter that has CDC enabled on the table.

For example, given a {cass-short} cluster with three datacenters (DC1, DC2, and DC3), you would enable CDC and install the change agent in only DC1. To ensure all updates in DC2 and DC3 are propagated to the data topic, configure the table’s keyspace to replicate data from DC2 and DC3 to DC1. For example, replication = {'class': 'NetworkTopologyStrategy', 'dc1': 3, 'dc2': 3, 'dc3': 3}. The data replicated to DC1 is processed by the change agent and eventually end up in the data topic.