<commit_before>[id="debezium-connector-for-cassandra"]
= {prodname} Connector for Cassandra

:context: cassandra
:toc:
:toc-placement: macro
:linkattrs:
:icons: font
:source-highlighter: highlight.js

toc::[]

[NOTE]
====
This connector is currently in incubating state, i.e. exact semantics, configuration options etc. may change in future revisions, based on the feedback we receive. Please let us know if you encounter any problems.
====

The Cassanadra connector can monitor a Cassandra cluster and record all row-level changes. The connector must be deployed locally on each node in the Cassandra cluster. The first time the connector connects to a Cassandra node, it performs a snapshot of all CDC-enabled tables in all keyspaces. The connector will also read the changes that are written to Cassandra commit logs and generates corresponding insert, update, and delete events. All events for each table are recorded in a separate kafka topic, where they can be consumed easily by applications and services.


[[cassandra-overview]]
== Overview

Cassandra is an open-sourced NoSQL database. Similar to most databases, the write path of Cassandra starts with the immeidate logging of a change into its commit log. The commit log resides locally on each node, recording every write made to that node.

Since Cassandra 3.0, a http://cassandra.apache.org/doc/3.11.3/operating/cdc.html[change data capture (CDC) feature] is introduced. 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 will be moved to the CDC directory specified in `cassandra.yaml` on discard.

The Cassandra connector resides on each Cassandra node and monitors the `cdc_raw` directory for change. It processes all local commit log segments as they are detected, produces a change event for every row-level insert, update, and delete operations in the commit log, publishes all change events for each table in a separate Kafka topic, and finally deletes the commit log from the `cdc_raw` directory. This last step is important because once CDC is enabled, Cassandra itself cannot purge the commit logs. If the `cdc_free_space_in_mb` fills up, writes to CDC-enabled tables will be rejected.

The connector is tolerant of failures. As the connector reads commit logs and produces events, it records each commit log segment's filename and position along with each event. If the connector stops for any reason (including communication failures, network problems, or crashes), upon restart it simply continues reading the commit log where it last left off. This includes snapshots: if the snapshot was not completed when the connector is stopped, upon restart it will begin a new snapshot. We'll talk later about how the connector behaves {link-prefix}:{link-cassandra-connector}#cassandra-when-things-go-wrong[when things go wrong].

[NOTE]
====
Cassandra is different from the other {prodname} connectors since it is not implemented on top of the Kafka Connect framework. Instead it is a single JVM process that is intended to reside on each Cassandra node and publishes events to Kafka via a Kafka producer.
====

[WARNING]
====
The following features are currently not supported by the Cassandra connector. Changes resulted from any of these features are ignored:

* TTL on collection-type columns
* Range deletes
* Static columns
* Triggers
* Materialized views
* Secondary indices
* Light-weight transactions
====

[[setting-up-cassandra]]
== Setting Up Cassandra

Before the {prodname} Cassandra connector can be used to monitor the changes in a Cassandra cluster, CDC must be enabled on the node level and table level.

[[enabling-cdc-on-node]]
=== Enabling CDC on Node

To enable CDC, update the following CDC config in `cassandra.yaml`:

[source,yaml]
----
cdc_enabled: true
----

Additional CDC configs are have the following default values:

[source,yaml]
----
cdc_raw_directory: $CASSANDRA_HOME/data/cdc_raw
cdc_free_space_in_mb: 4096
cdc_free_space_check_interval_ms: 250
----

* `cdc_enabled` enables or disables CDC operations node-wide
* `cdc_raw_directory` determines the destination for commit log segments to be moved after all corresponding memtables are flushed
* `cdc_free_space_in_mb` is the maximum capacity allocated to store commit log segments, and defaults to the minimum of 4096 MB and 1/8 of volume space.
* `cdc_free_space_check_interval_ms` is frequency with which we re-calculate the space taken up by `cdc_raw_directory` to prevent burning CPU cycles unnecessarily when at capacity.


[[enabling-cdc-on-table]]
=== Enabling CDC on Table

Once CDC is enabled on the Cassandra node, each table must be be explicitly enabled for CDC as well via the CREATE TABLE or ALTER TABLE command. For example:

[source,sql]
----
CREATE TABLE foo (a int, b text, PRIMARY KEY(a)) WITH cdc=true;

ALTER TABLE foo WITH cdc=true;
----

[[how-the-cassandra-connector-works]]
== How the Cassandra Connector Works

This section goes into detail about how the Cassandra connector performs snapshots, transforms commit log events into {prodname} change events, handles commit log life cycle, records events into Kafka, manages schema evolution, and behaves when things go wrong.


[[cassandra-snapshots]]
=== Snapshots

When the Cassandra connector first starts on a Cassandra node, it will by default perform an initial snapshot of the cluster. This is the default mode, since most of the time CDC is enabled on non-empty tables and commit logs do not contain the complete history.

The snapshot reader issues a SELECT statement to query all the columns in a table. Cassandra allows consistency level to be set either globally or on the statement level. For snapshotting, the consistency level is set on the statement level to `ALL` by default to provide the highest consistency. This implies if one node goes down during the snapshot, the snapshot would not be able to continue and a subsequent re-snapshot is required once the node has been brought back online. You can adjust the consistency level of the snapshot to a lower consistency level in order to increase availability, provided that you understand the tradeoff with consistency.

[NOTE]
====
In Cassandra 3.X, it is not possible to read strictly from the local Cassandra node. Starting in Cassandra 4.0, a `NODE_LOCAL` consistency level will be added. This will allow the Cassandra connector to read from the node it resides in only (which would be consistent with the way commit logs are processed).
====

Unlike relational databases, there is no read lock applied during a snapshot, so writes to Cassandra are not blocked during that snapshot. If the queried data has been modified by another client during the snapshot, those changes may be reflected in the snapshot result set.

If the connector fails or stops before the snapshot is completed, the connector will begin a new snapshot upon restarts. In the default snapshot mode (`initial`), once the connector completes its initial snapshot, it will no longer perform any additional snapshots. The only exception would be during a connector restart: if cdc is enabled on a table, and then the connector is restarted, that table would be snapshotted.

The second snapshot mode (`always`) allows the connector to perform snapshot whenever necessary. it will check periodically for newly cdc-enabled tables, and snapshot them as soon as they are detected.

The third snapshot mode ('never') ensures the connector never performs snapshots. When a new connector is configured this way, it will only read the commit log in the CDC directory. This is not the default behavior because starting a new connector in this mode (without a snapshot) requires the commit logs to contain the entire history of all cdc-enabled tables, which is often not the case. Another use case for this mode is if there is one connector already doing the snapshotting, you can disable snapshot on others to avoid duplicated work.

[[reading-the-commitlog]]
=== Reading the Commit Log

The Cassandra connector will typically spend the vast majority of its time reading local commit logs on the Cassandra node.

Commit logs' binary data are deserialized with Cassandra's CommitLogReader and CommitLogReadHandler. Each deserialized object is called a `mutation` in Cassandra. A `mutation` contains one or more change events.

As the Cassandra connector reads the commit log, it transform the log events into {prodname} _create_, _update_, or _delete_ events that include the position in the commit log where the event was found. The Cassandra connector encode these change events with Kafka Connect converters and publish them to the appropriate Kafka topics.

[[limitations-of-commit-logs]]
=== Limitations of Commit Logs

Cassandra's commit logs come with a set of limitations, which are critical for interpreting CDC events correctly:

* Commit logs only arrive in `cdc_raw` directory when it is full, in which case it would be flushed/discarded. This implies there is a delay between when the event is logged and when the event is captured.
* Commit logs on an individual Cassandra node do not reflect all writes to the cluster, they only reflect writes stored on that node. This is why it is necesssary to monitor changes on all nodes in a Cassandra cluster. However, due to replication factor, this also implies it is necessary for downstream consumers of these events to handle deduplication.
* Writes to an individual Cassandra node are logged as they arrive. However, these events may arrive out-of-order from which they are issued. Downstream consumers of these events must understand and implement logic similar to Cassandra's read path to get the correct output.
* Schema changes of tables are not recorded in commit logs, only data changes are recorded. Therefore changes in schema are detected on a best-effort basis. To avoid data loss, it is recommended to pause writes to the table during schema change.
* Cassandra does not perform read-before-write, as a result commit logs do not record the value of every column in the changed row, it only records the values of columns that have been modified (except for partition key columns, which are always recorded as they are required in Cassandra DML commands).
* Due to the nature of CQL, _insert_ DMLs can result in a row insertion or update; _update_ DMLs can result in a row insertion, update, or deletion; _delete_ DMLs can result in a row update or deletion. Since queries are not recorded in commit logs, CDC event type is classified based on the effect on the row in a relational database sense.

**TODO**: is there a way to determine event type which corresponds to the actual Cassandra DML statement? and if so, is that preferred over the semantic of these events?

[NOTE]
====
In Cassandra 4.0 on every segment fsync, an index file will be updated to reflect latest offset. This will eliminate the processing delay in the CDC feature in Cassandra 3.X. This connector feature is to be added in the future with the Cassandra 4.0 release.
====

[[managing-commitlog-lifecycle]]
=== Managing Commit Log Lifecycle

By default, Cassandra connector will delete commit logs which have been processed. It is not recommended to start the connector while deletion of commit logs is disabled, as this could bloat up disk storage and prevent further writes to the Cassandra cluster. To manage the commit logs in a custom manner (i.e. upload it to a cloud provider), the CommitLogTransfer interface can be implemented.


[[cassandra-topic-names]]
=== Topics names

The Cassandra connector writes events for all insert, update, and delete uperations on a single table to a single Kafka topic. The name of the Kafka topics always take the form
_clusterName_._keyspaceName_._tableName_, where _clusterName_ is the logical name of the connector as specified with the `kafka.topic.prefix` configuraiton property, _keyspaceName_ is the name of the keyspace where the operation occurred, and _tableName_ is the name of the table on which the operation occurred.

For example, consider a Cassandra installation with an `inventory` keyspace that contains four tables: `products`, `products_on_hand`, `customers`, and `orders`. If the connector monitoring this database were given a logical server name of `fulfillment`, then the connector would produce events on these four Kafka topics:

* `fulfillment.inventory.products`
* `fulfillment.inventory.products_on_hand`
* `fulfillment.inventory.customers`
* `fulfillment.inventory.orders`

**TODO**: for topic name, is _clusterName_._keyspaceName_._tableName_ okay? or should it be _connectorName_._keyspaceName_._tableName_ or _connectorName_._clusterName_._keyspaceName_._tableName_?

[[cassandra-schema-evolution]]
=== Schema Evolution

DDLs are not recorded in commit logs. When the schema of a table change, this change is issued from one of the Cassandra node and propagated to other nodes via Gossip Protocol.

Schema changes in Cassandra will be detected by an implemented SchemaChangeListener with latency less than 1s, which will then update the schema instance loaded from Cassandra as well as the Kafka key value schemas cached for each table.

Please note that with the current schema evolution approach, Cassandra Connector won't be able provide accurate data change information for the following cases:
1). If CDC is disabled for a table, data changes which have happened before CDC gets disabled will be skipped.
2). If a column is removed from a table, data changes involving this column before it's removed cannot be deserialized correctly and will be skipped.

[[cassandra-events]]
=== Events

All data change events produced by the Cassandra connector have a key and a value, although the structure of the key and value depends on the table from which the change events originated (see {link-prefix}:{link-cassandra-connector}#cassandra-topic-names[Topic Names]).

[[cassandra-change-events-key]]
==== Change Event's Key

For a given table, the change event's key will have a structure that contains a field for each column in the primary key of the table at the time the event was created. Consider an `inventory` database with a `customers` table defined as:

[source,sql,indent=0]
----
CREATE TABLE customers (
  id bigint,
  registration_date timestamp,
  first_name text,
  last_name text,
  email text,
  PRIMARY KEY (id, registration_date)<commit_msg>DBZ-3506 Formatting fix<commit_after>[id="debezium-connector-for-cassandra"]
= {prodname} Connector for Cassandra

:context: cassandra
:toc:
:toc-placement: macro
:linkattrs:
:icons: font
:source-highlighter: highlight.js

toc::[]

[NOTE]
====
This connector is currently in incubating state, i.e. exact semantics, configuration options etc. may change in future revisions, based on the feedback we receive. Please let us know if you encounter any problems.
====

The Cassanadra connector can monitor a Cassandra cluster and record all row-level changes. The connector must be deployed locally on each node in the Cassandra cluster. The first time the connector connects to a Cassandra node, it performs a snapshot of all CDC-enabled tables in all keyspaces. The connector will also read the changes that are written to Cassandra commit logs and generates corresponding insert, update, and delete events. All events for each table are recorded in a separate kafka topic, where they can be consumed easily by applications and services.


[[cassandra-overview]]
== Overview

Cassandra is an open-sourced NoSQL database. Similar to most databases, the write path of Cassandra starts with the immeidate logging of a change into its commit log. The commit log resides locally on each node, recording every write made to that node.

Since Cassandra 3.0, a http://cassandra.apache.org/doc/3.11.3/operating/cdc.html[change data capture (CDC) feature] is introduced. 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 will be moved to the CDC directory specified in `cassandra.yaml` on discard.

The Cassandra connector resides on each Cassandra node and monitors the `cdc_raw` directory for change. It processes all local commit log segments as they are detected, produces a change event for every row-level insert, update, and delete operations in the commit log, publishes all change events for each table in a separate Kafka topic, and finally deletes the commit log from the `cdc_raw` directory. This last step is important because once CDC is enabled, Cassandra itself cannot purge the commit logs. If the `cdc_free_space_in_mb` fills up, writes to CDC-enabled tables will be rejected.

The connector is tolerant of failures. As the connector reads commit logs and produces events, it records each commit log segment's filename and position along with each event. If the connector stops for any reason (including communication failures, network problems, or crashes), upon restart it simply continues reading the commit log where it last left off. This includes snapshots: if the snapshot was not completed when the connector is stopped, upon restart it will begin a new snapshot. We'll talk later about how the connector behaves {link-prefix}:{link-cassandra-connector}#cassandra-when-things-go-wrong[when things go wrong].

[NOTE]
====
Cassandra is different from the other {prodname} connectors since it is not implemented on top of the Kafka Connect framework. Instead it is a single JVM process that is intended to reside on each Cassandra node and publishes events to Kafka via a Kafka producer.
====

[WARNING]
====
The following features are currently not supported by the Cassandra connector. Changes resulted from any of these features are ignored:

* TTL on collection-type columns
* Range deletes
* Static columns
* Triggers
* Materialized views
* Secondary indices
* Light-weight transactions
====

[[setting-up-cassandra]]
== Setting Up Cassandra

Before the {prodname} Cassandra connector can be used to monitor the changes in a Cassandra cluster, CDC must be enabled on the node level and table level.

[[enabling-cdc-on-node]]
=== Enabling CDC on Node

To enable CDC, update the following CDC config in `cassandra.yaml`:

[source,yaml]
----
cdc_enabled: true
----

Additional CDC configs are have the following default values:

[source,yaml]
----
cdc_raw_directory: $CASSANDRA_HOME/data/cdc_raw
cdc_free_space_in_mb: 4096
cdc_free_space_check_interval_ms: 250
----

* `cdc_enabled` enables or disables CDC operations node-wide
* `cdc_raw_directory` determines the destination for commit log segments to be moved after all corresponding memtables are flushed
* `cdc_free_space_in_mb` is the maximum capacity allocated to store commit log segments, and defaults to the minimum of 4096 MB and 1/8 of volume space.
* `cdc_free_space_check_interval_ms` is frequency with which we re-calculate the space taken up by `cdc_raw_directory` to prevent burning CPU cycles unnecessarily when at capacity.


[[enabling-cdc-on-table]]
=== Enabling CDC on Table

Once CDC is enabled on the Cassandra node, each table must be be explicitly enabled for CDC as well via the CREATE TABLE or ALTER TABLE command. For example:

[source,sql]
----
CREATE TABLE foo (a int, b text, PRIMARY KEY(a)) WITH cdc=true;

ALTER TABLE foo WITH cdc=true;
----

[[how-the-cassandra-connector-works]]
== How the Cassandra Connector Works

This section goes into detail about how the Cassandra connector performs snapshots, transforms commit log events into {prodname} change events, handles commit log life cycle, records events into Kafka, manages schema evolution, and behaves when things go wrong.


[[cassandra-snapshots]]
=== Snapshots

When the Cassandra connector first starts on a Cassandra node, it will by default perform an initial snapshot of the cluster. This is the default mode, since most of the time CDC is enabled on non-empty tables and commit logs do not contain the complete history.

The snapshot reader issues a SELECT statement to query all the columns in a table. Cassandra allows consistency level to be set either globally or on the statement level. For snapshotting, the consistency level is set on the statement level to `ALL` by default to provide the highest consistency. This implies if one node goes down during the snapshot, the snapshot would not be able to continue and a subsequent re-snapshot is required once the node has been brought back online. You can adjust the consistency level of the snapshot to a lower consistency level in order to increase availability, provided that you understand the tradeoff with consistency.

[NOTE]
====
In Cassandra 3.X, it is not possible to read strictly from the local Cassandra node. Starting in Cassandra 4.0, a `NODE_LOCAL` consistency level will be added. This will allow the Cassandra connector to read from the node it resides in only (which would be consistent with the way commit logs are processed).
====

Unlike relational databases, there is no read lock applied during a snapshot, so writes to Cassandra are not blocked during that snapshot. If the queried data has been modified by another client during the snapshot, those changes may be reflected in the snapshot result set.

If the connector fails or stops before the snapshot is completed, the connector will begin a new snapshot upon restarts. In the default snapshot mode (`initial`), once the connector completes its initial snapshot, it will no longer perform any additional snapshots. The only exception would be during a connector restart: if cdc is enabled on a table, and then the connector is restarted, that table would be snapshotted.

The second snapshot mode (`always`) allows the connector to perform snapshot whenever necessary. it will check periodically for newly cdc-enabled tables, and snapshot them as soon as they are detected.

The third snapshot mode ('never') ensures the connector never performs snapshots. When a new connector is configured this way, it will only read the commit log in the CDC directory. This is not the default behavior because starting a new connector in this mode (without a snapshot) requires the commit logs to contain the entire history of all cdc-enabled tables, which is often not the case. Another use case for this mode is if there is one connector already doing the snapshotting, you can disable snapshot on others to avoid duplicated work.

[[reading-the-commitlog]]
=== Reading the Commit Log

The Cassandra connector will typically spend the vast majority of its time reading local commit logs on the Cassandra node.

Commit logs' binary data are deserialized with Cassandra's CommitLogReader and CommitLogReadHandler. Each deserialized object is called a `mutation` in Cassandra. A `mutation` contains one or more change events.

As the Cassandra connector reads the commit log, it transform the log events into {prodname} _create_, _update_, or _delete_ events that include the position in the commit log where the event was found. The Cassandra connector encode these change events with Kafka Connect converters and publish them to the appropriate Kafka topics.

[[limitations-of-commit-logs]]
=== Limitations of Commit Logs

Cassandra's commit logs come with a set of limitations, which are critical for interpreting CDC events correctly:

* Commit logs only arrive in `cdc_raw` directory when it is full, in which case it would be flushed/discarded. This implies there is a delay between when the event is logged and when the event is captured.
* Commit logs on an individual Cassandra node do not reflect all writes to the cluster, they only reflect writes stored on that node. This is why it is necesssary to monitor changes on all nodes in a Cassandra cluster. However, due to replication factor, this also implies it is necessary for downstream consumers of these events to handle deduplication.
* Writes to an individual Cassandra node are logged as they arrive. However, these events may arrive out-of-order from which they are issued. Downstream consumers of these events must understand and implement logic similar to Cassandra's read path to get the correct output.
* Schema changes of tables are not recorded in commit logs, only data changes are recorded. Therefore changes in schema are detected on a best-effort basis. To avoid data loss, it is recommended to pause writes to the table during schema change.
* Cassandra does not perform read-before-write, as a result commit logs do not record the value of every column in the changed row, it only records the values of columns that have been modified (except for partition key columns, which are always recorded as they are required in Cassandra DML commands).
* Due to the nature of CQL, _insert_ DMLs can result in a row insertion or update; _update_ DMLs can result in a row insertion, update, or deletion; _delete_ DMLs can result in a row update or deletion. Since queries are not recorded in commit logs, CDC event type is classified based on the effect on the row in a relational database sense.

**TODO**: is there a way to determine event type which corresponds to the actual Cassandra DML statement? and if so, is that preferred over the semantic of these events?

[NOTE]
====
In Cassandra 4.0 on every segment fsync, an index file will be updated to reflect latest offset. This will eliminate the processing delay in the CDC feature in Cassandra 3.X. This connector feature is to be added in the future with the Cassandra 4.0 release.
====

[[managing-commitlog-lifecycle]]
=== Managing Commit Log Lifecycle

By default, Cassandra connector will delete commit logs which have been processed. It is not recommended to start the connector while deletion of commit logs is disabled, as this could bloat up disk storage and prevent further writes to the Cassandra cluster. To manage the commit logs in a custom manner (i.e. upload it to a cloud provider), the CommitLogTransfer interface can be implemented.


[[cassandra-topic-names]]
=== Topics names

The Cassandra connector writes events for all insert, update, and delete uperations on a single table to a single Kafka topic. The name of the Kafka topics always take the form
_clusterName_._keyspaceName_._tableName_, where _clusterName_ is the logical name of the connector as specified with the `kafka.topic.prefix` configuraiton property, _keyspaceName_ is the name of the keyspace where the operation occurred, and _tableName_ is the name of the table on which the operation occurred.

For example, consider a Cassandra installation with an `inventory` keyspace that contains four tables: `products`, `products_on_hand`, `customers`, and `orders`. If the connector monitoring this database were given a logical server name of `fulfillment`, then the connector would produce events on these four Kafka topics:

* `fulfillment.inventory.products`
* `fulfillment.inventory.products_on_hand`
* `fulfillment.inventory.customers`
* `fulfillment.inventory.orders`

**TODO**: for topic name, is _clusterName_._keyspaceName_._tableName_ okay? or should it be _connectorName_._keyspaceName_._tableName_ or _connectorName_._clusterName_._keyspaceName_._tableName_?

[[cassandra-schema-evolution]]
=== Schema Evolution

DDLs are not recorded in commit logs. When the schema of a table change, this change is issued from one of the Cassandra node and propagated to other nodes via Gossip Protocol.

Schema changes in Cassandra will be detected by an implemented SchemaChangeListener with latency less than 1s, which will then update the schema instance loaded from Cassandra as well as the Kafka key value schemas cached for each table.

Please note that with the current schema evolution approach, the Cassandra connector won't be able to provide accurate data change information for a small period of time in the following cases:

. If CDC gets disabled for a table, data changes which have happened before CDC got disabled will be skipped.
. If a column is removed from a table, data changes involving this column before it's removed cannot be deserialized correctly and will be skipped.

[[cassandra-events]]
=== Events

All data change events produced by the Cassandra connector have a key and a value, although the structure of the key and value depends on the table from which the change events originated (see {link-prefix}:{link-cassandra-connector}#cassandra-topic-names[Topic Names]).

[[cassandra-change-events-key]]
==== Change Event's Key

For a given table, the change event's key will have a structure that contains a field for each column in the primary key of the table at the time the event was created. Consider an `inventory` database with a `customers` table defined as:

[source,sql,indent=0]
----
CREATE TABLE customers (
  id bigint,
  registration_date timestamp,
  first_name text,
  last_name text,
  email text,
  PRIMARY KEY (id, registration_date)
{'fraction_non_alphanumeric': 0.048145715964469635, 'fraction_numerical': 0.0022120070507724745, 'mean_word_length': 4.836998184385717, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3953267', 'n_tokens_mistral': 7195, 'n_tokens_neox': 6842, 'n_words': 4243}