> ## Documentation Index
> Fetch the complete documentation index at: https://docs.automq.com/llms.txt
> Use this file to discover all available pages before exploring further.

# DataStax Cassandra Sink Connector

> Configure and operate the DataStax Cassandra Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting.

## Overview

The DataStax Cassandra Sink Connector consumes records from Kafka topics and maps fields from message keys, values, or headers to target tables in Apache Cassandra or DataStax Enterprise. It sits between Kafka and Cassandra and is suitable for continuously writing business events, entity states, and log data to a storage model organized by keyspaces and tables.

Each topic can be configured with one or more target tables. The Connector generates CQL write statements from the mapping, and can also use a custom Prepared Statement; when the same Kafka record is written to multiple tables, writes to each table are independent. Plan the target keyspace, table, columns, and primary keys in advance; the mapping defines the relationship between Kafka fields and Cassandra columns.

## Prerequisites

* Create the target Keyspace and Table in Cassandra in advance, and ensure that the connection account can read table metadata and perform the required write or delete operations; columns referenced by the mapping must exist, and automatically generated CQL must cover all primary-key columns.

## License

Uses the DataStax Apache Kafka Connector License Terms. The release package contains inconsistent license labels. Before using or redistributing it, check the bundled license files and complete your customer-side review.

## Quick Start

Prepare the Connect Cluster, Kafka, Cassandra Keyspace, and target table in advance, and confirm network connectivity and access permissions. For preparation and management operations, see [Manage Connectors](../manage-connectors).

The following example subscribes to the `orders` topic, writes the message key to `order_id`, and writes `customer_id` and `status` from the message value to the `app.orders` table. The target table should use `order_id` as its primary key and include these columns.

```properties theme={null}
connector.class=com.datastax.oss.kafka.sink.CassandraSinkConnector
tasks.max=1
topics=orders
contactPoints=<cassandra-host>
loadBalancing.localDc=<local-dc>
topic.orders.app.orders.mapping=order_id=key, customer_id=value.customer_id, status=value.status
topic.orders.app.orders.deletesEnabled=false
```

Replace `<cassandra-host>` with a Cassandra contact point and `<local-dc>` with the name of the data center that contains that node. If the topic, keyspace, table, or field names differ, update `topics`, the dynamic configuration keys, and `mapping` accordingly; after startup, send records with a stable key and `customer_id` and `status` fields, then check the corresponding row in `app.orders`.

## Configuration

### Kafka Connect and Subscription

#### `connector.class`

Specifies the Connector implementation class to load.

* **Type**: `string`
* **Default**: None
* **Importance**: High
* **Valid values / Notes**: Use `com.datastax.oss.kafka.sink.CassandraSinkConnector`. The old class name `com.datastax.kafkaconnector.DseSinkConnector` is retained only as a deprecated compatibility alias.
* **Required**: Yes

#### `tasks.max`

Sets the maximum number of Tasks allowed to start.

* **Type**: `int`
* **Default**: `1`
* **Importance**: High
* **Valid values / Notes**: At least `1`. The actual number of Tasks does not exceed the number of Topic Partitions that can be assigned; increasing the number of Tasks does not guarantee linear throughput growth.

#### `topics`

Explicitly lists the Kafka Topics to consume.

* **Type**: `list`
* **Default**: Empty list
* **Importance**: High
* **Valid values / Notes**: Separate values with commas. Configure exactly one of `topics` and `topics.regex`; every explicitly listed Topic must have at least one corresponding `topic.<topic>.<keyspace>.<table>.*` table configuration.

#### `topics.regex`

Subscribes to Kafka Topics using a Java regular expression.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: Configure exactly one of `topics` and `topics.regex`. A Topic newly matched by the regular expression still requires an exactly corresponding `topic.<topic>.*` configuration; otherwise mapping fails when records arrive.

### Cassandra Connection

#### `contactPoints`

Sets the initial Cassandra contact points.

* **Type**: `list`
* **Default**: Empty list
* **Importance**: High
* **Valid values / Notes**: Enter comma-separated IP addresses or hostnames; all nodes use the same port specified by `port`. When non-empty, the local data center must also be configured, and this setting cannot be used with `cloud.secureConnectBundle`. To specify different ports for different nodes, use the Java Driver native contact-point configuration instead.

#### `port`

Sets the Cassandra native transport port used by `contactPoints`.

* **Type**: `int`
* **Default**: `9042`
* **Importance**: High
* **Valid values / Notes**: At least `1`; ignored when a secure connect bundle is used.

#### `loadBalancing.localDc`

Sets the local data center that contains the contact points.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: Required when `contactPoints` is non-empty; must be left empty when a secure connect bundle is used. If the replacement setting is also configured, this setting takes precedence.
* **Deprecated**: Yes
* **Replacement**: `datastax-java-driver.basic.load-balancing-policy.local-datacenter`

#### `cloud.secureConnectBundle`

Sets the path to the DataStax Cloud secure connect bundle.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: A non-empty value enables cloud connection mode and cannot be used with `contactPoints`, `loadBalancing.localDc`, or any `ssl.*` setting. Cloud mode changes the `ANY`, `ONE`, and `LOCAL_ONE` write consistency levels to `LOCAL_QUORUM`. If the replacement setting is also configured, this setting takes precedence.
* **Deprecated**: Yes
* **Replacement**: `datastax-java-driver.basic.cloud.secure-connect-bundle`

### Authentication

#### `auth.provider`

Selects the Cassandra authentication provider.

* **Type**: `string`
* **Default**: `None`
* **Importance**: High
* **Valid values / Notes**: Case-sensitive; choose `None`, `PLAIN`, or `GSSAPI`. When a username or password is configured, `PLAIN` is used in practice even if this setting is `None`.

#### `auth.username`

Sets the PLAIN authentication username.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: Used for `PLAIN` authentication; when a non-empty password is configured, the username must also be non-empty.

#### `auth.password`

Sets the PLAIN authentication password.

* **Type**: `password`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: Used for `PLAIN` authentication. Inject it through a secure configuration mechanism; do not expose the plaintext password in logs or shared files.

#### `auth.gssapi.keyTab`

Sets the path to the Kerberos keytab file.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: A non-empty path must point to a regular file readable by the Worker. When a keytab is used without a principal, the first principal in the keytab can be inferred; without a keytab, the Driver uses the ticket cache.

#### `auth.gssapi.principal`

Sets the Kerberos principal.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: Used only for `GSSAPI`. Omitting this setting and explicitly configuring an empty string have different behavior; inference from the keytab is available only when the setting is omitted.

#### `auth.gssapi.service`

Sets the GSSAPI SASL service name.

* **Type**: `string`
* **Default**: `dse`
* **Importance**: High
* **Valid values / Notes**: Must be a non-empty string when using `GSSAPI` and must match the Cassandra server configuration.

### TLS

#### `ssl.provider`

Selects the TLS implementation provider.

* **Type**: `string`
* **Default**: `None`
* **Importance**: High
* **Valid values / Notes**: Case-sensitive; choose `None`, `JDK`, or `OpenSSL`. When a secure connect bundle is used, no `ssl.*` setting may be configured.

#### `ssl.cipherSuites`

Sets the TLS cipher suites that may be used.

* **Type**: `list`
* **Default**: Empty list
* **Importance**: High
* **Valid values / Notes**: Separate values with commas; an empty list uses the defaults of the selected TLS provider. Cannot be configured with a secure connect bundle.

#### `ssl.hostnameValidation`

Controls whether Cassandra node hostnames are validated.

* **Type**: `boolean`
* **Default**: `true`
* **Importance**: High
* **Valid values / Notes**: When `true`, JDK SSL resolves contact-point addresses and enables hostname validation. Cannot be configured with a secure connect bundle.

#### `ssl.keystore.password`

Sets the JDK SSL keystore password.

* **Type**: `password`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: Used to access the JDK SSL keystore. Inject it securely, and do not configure it with a secure connect bundle.

#### `ssl.keystore.path`

Sets the JDK SSL keystore path.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: A non-empty path must point to a regular file readable by the Worker. Cannot be configured with a secure connect bundle.

#### `ssl.openssl.keyCertChain`

Sets the path to the OpenSSL client certificate chain.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: A non-empty path must point to a regular file readable by the Worker. When using `OpenSSL`, configure this setting and `ssl.openssl.privateKey` together or omit both; cannot be configured with a secure connect bundle.

#### `ssl.openssl.privateKey`

Sets the path to the OpenSSL client private key.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: A non-empty path must point to a regular file readable by the Worker. When using `OpenSSL`, configure this setting and `ssl.openssl.keyCertChain` together or omit both; protect the private-key file permissions, and do not configure it with a secure connect bundle.

#### `ssl.truststore.password`

Sets the TLS truststore password.

* **Type**: `password`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: Used to load the truststore when initializing JDK or OpenSSL. Inject it securely, and do not configure it with a secure connect bundle.

#### `ssl.truststore.path`

Sets the TLS truststore path.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid values / Notes**: A non-empty path must point to a regular file readable by the Worker; OpenSSL mode loads the file as JKS. Cannot be configured with a secure connect bundle.

### Write Throughput and Requests

#### `maxConcurrentRequests`

Limits the number of in-flight Cassandra requests for this Connector instance within the same Worker JVM.

* **Type**: `int`
* **Default**: `500`
* **Importance**: High
* **Valid values / Notes**: At least `1`. Multiple Tasks of Connectors with the same name in the same Worker JVM share this limit; it is not shared across Workers.

#### `maxNumberOfRecordsInBatch`

Sets the maximum number of records in one Cassandra batch request.

* **Type**: `int`
* **Default**: `32`
* **Importance**: High
* **Valid values / Notes**: At least `1`. The Connector buckets records by Topic, target table, and Cassandra Routing Key; multiple statements use an UNLOGGED batch, while a single statement is executed asynchronously.

#### `compression`

Sets the Cassandra protocol compression algorithm.

* **Type**: `string`
* **Default**: `None`
* **Importance**: High
* **Valid values / Notes**: Case-insensitive; choose `none`, `snappy`, or `lz4`. If the replacement setting is also configured, this setting takes precedence.
* **Deprecated**: Yes
* **Replacement**: `datastax-java-driver.advanced.protocol.compression`

#### `queryExecutionTimeout`

Sets the CQL request timeout in seconds.

* **Type**: `int`
* **Default**: `30`
* **Importance**: High
* **Valid values / Notes**: At least `1`; the value is passed to the Java Driver in seconds. If the replacement setting is also configured, this setting takes precedence.
* **Deprecated**: Yes
* **Replacement**: `datastax-java-driver.basic.request.timeout`

#### `connectionPoolLocalSize`

Sets the connection-pool size for each local Cassandra node.

* **Type**: `int`
* **Default**: `4`
* **Importance**: High
* **Valid values / Notes**: At least `1`. If the replacement setting is also configured, this setting takes precedence.
* **Deprecated**: Yes
* **Replacement**: `datastax-java-driver.advanced.connection.pool.local.size`

### Error Handling and Metrics

#### `ignoreErrors`

Controls whether the Connector ignores record-mapping or Cassandra Driver write errors.

* **Type**: `string`
* **Default**: `None`
* **Importance**: High
* **Valid values / Notes**: Case-insensitive; choose `None`, `Driver`, or `All`. `Driver` ignores only database Driver write failures; `All` also ignores mapping failures. Ignoring errors can allow the Offset to pass failed records, and the Connector does not automatically write those records again. This policy is independent of Kafka Connect's `errors.tolerance`.

#### `jmx`

Controls whether the default Java Driver JMX session metrics are enabled.

* **Type**: `boolean`
* **Default**: `true`
* **Importance**: High
* **Valid values / Notes**: When `true` and no Driver metric list is explicitly configured, enables `cql-requests` and `cql-client-timeouts`; the default sampling interval for `cql-requests` is 30 seconds.

#### `metricsHighestLatency`

Sets the highest latency range in seconds for CQL request metrics.

* **Type**: `int`
* **Default**: `35`
* **Importance**: High
* **Valid values / Notes**: At least `1`; should normally be greater than the request timeout. This relationship is not enforced by validation. If the replacement setting is also configured, this setting takes precedence.
* **Deprecated**: Yes
* **Replacement**: `datastax-java-driver.advanced.metrics.session.cql-requests.highest-latency`

### Topic Data Conversion

#### `topic.<topic>.codec.locale`

Sets the text-conversion locale for the specified Topic.

* **Type**: `string`
* **Default**: `en_US`
* **Importance**: High
* **Valid values / Notes**: `<topic>` may contain only letters, digits, periods, underscores, and hyphens; the locale value must be parseable by the DSBulk Codec.

#### `topic.<topic>.codec.timeZone`

Sets the time-conversion time zone for the specified Topic.

* **Type**: `string`
* **Default**: `UTC`
* **Importance**: High
* **Valid values / Notes**: Must be a time zone recognized by `java.time.ZoneId`, such as `Asia/Shanghai`.

#### `topic.<topic>.codec.timestamp`

Sets the format used when converting strings to CQL `timestamp` values.

* **Type**: `string`
* **Default**: `CQL_TIMESTAMP`
* **Importance**: High
* **Valid values / Notes**: May use time patterns supported by the DSBulk Codec, `DateTimeFormatter` constants, or `CQL_TIMESTAMP`.

#### `topic.<topic>.codec.date`

Sets the format used when converting strings to CQL `date` values.

* **Type**: `string`
* **Default**: `ISO_LOCAL_DATE`
* **Importance**: High
* **Valid values / Notes**: May use date patterns supported by the DSBulk Codec or `DateTimeFormatter` constants.

#### `topic.<topic>.codec.time`

Sets the format used when converting strings to CQL `time` values.

* **Type**: `string`
* **Default**: `ISO_LOCAL_TIME`
* **Importance**: High
* **Valid values / Notes**: May use time patterns supported by the DSBulk Codec or `DateTimeFormatter` constants.

#### `topic.<topic>.codec.unit`

Sets the time unit for numeric-only time inputs.

* **Type**: `string`
* **Default**: `MILLISECONDS`
* **Importance**: High
* **Valid values / Notes**: Must be exactly `NANOSECONDS`, `MICROSECONDS`, `MILLISECONDS`, `SECONDS`, `MINUTES`, `HOURS`, or `DAYS`.

### Table Mapping and Write Semantics

#### `topic.<topic>.<keyspace>.<table>.mapping`

Defines the mapping from Kafka fields to Cassandra columns or custom-query bind variables.

* **Type**: `string`
* **Default**: None
* **Importance**: High
* **Valid values / Notes**: Every target table must have a non-empty mapping in the comma-separated form `column=field expression`. Field expressions support `key`, `value`, `key.*`, `value.*`, `header.*`, `now()`, and the reserved pseudo-columns `__ttl` and `__timestamp`. With automatically generated CQL, ordinary columns must exist and all primary-key columns must be mapped.
* **Required**: Yes

#### `topic.<topic>.<keyspace>.<table>.deletesEnabled`

Controls whether a full mapping record generates a full-row delete.

* **Type**: `boolean`
* **Default**: `true`
* **Importance**: High
* **Valid values / Notes**: Only when no custom query is used and the mapping covers all columns in the target table, a record whose mapped values other than the primary key are all `null` deletes the full row. Must be set to `false` when a custom `query` is configured.

#### `topic.<topic>.<keyspace>.<table>.consistencyLevel`

Sets the consistency level for writes to the target table.

* **Type**: `string`
* **Default**: `LOCAL_ONE`
* **Importance**: High
* **Valid values / Notes**: Use a consistency-level name supported by the DataStax Driver; the value is case-insensitive. Cloud mode changes `ANY`, `ONE`, and `LOCAL_ONE` to `LOCAL_QUORUM`.

#### `topic.<topic>.<keyspace>.<table>.ttl`

Sets a fixed TTL for writes to the target table.

* **Type**: `int`
* **Default**: `-1`
* **Importance**: High
* **Valid values / Notes**: At least `-1`; `-1` disables a fixed TTL. The value is converted to seconds according to `ttlTimeUnit`; a dynamic `__ttl` value in the mapping takes precedence. Counter tables cannot use TTL, and custom queries determine their own TTL semantics.

#### `topic.<topic>.<keyspace>.<table>.nullToUnset`

Controls whether non-primary-key `null` values are treated as Cassandra `UNSET`.

* **Type**: `boolean`
* **Default**: `true`
* **Importance**: High
* **Valid values / Notes**: When `true`, a non-primary-key `null` does not overwrite the existing column value; when `false`, `null` is explicitly bound. A `null` primary key always fails.

#### `topic.<topic>.<keyspace>.<table>.ttlTimeUnit`

Sets the input unit for the fixed TTL and mapping `__ttl` values.

* **Type**: `string`
* **Default**: `SECONDS`
* **Importance**: High
* **Valid values / Notes**: Must be exactly `NANOSECONDS`, `MICROSECONDS`, `MILLISECONDS`, `SECONDS`, `MINUTES`, `HOURS`, or `DAYS`.

#### `topic.<topic>.<keyspace>.<table>.timestampTimeUnit`

Sets the input unit for the mapping `__timestamp` values.

* **Type**: `string`
* **Default**: `MICROSECONDS`
* **Importance**: High
* **Valid values / Notes**: Must use a `TimeUnit` enum value. When `__timestamp` is not mapped and no custom query is used, the Connector converts the Kafka record timestamp from milliseconds to microseconds for the CQL write timestamp.

#### `topic.<topic>.<keyspace>.<table>.query`

Sets a custom Prepared Statement CQL query for the target table.

* **Type**: `string`
* **Default**: `null`
* **Importance**: High
* **Valid values / Notes**: A non-`null` value replaces the automatically generated INSERT or Counter UPDATE and requires `deletesEnabled=false`. Bind variables must be provided by `mapping`; the Connector no longer automatically validates table columns, primary keys, TTL, timestamps, or delete semantics. An empty string is also treated as configured and fails during database preparation.

### Java Driver Passthrough

#### `datastax-java-driver.<driver-path>`

Passes native DataStax Java Driver configuration to the Driver.

* **Type**: `string`
* **Default**: No ConfigDef default
* **Importance**: Not declared
* **Valid values / Notes**: Use the `datastax-java-driver.<driver-path>` form; the path, value, default, and validation rules are determined by the Java Driver supplied with the Connector. List paths for contact points, keyspace refresh, node metrics, session metrics, and TLS cipher suites are split on commas; when an old Connector alias and the corresponding Driver path are both configured, the old alias takes precedence.

## Best Practices

### Synchronize Business Deletes to Cassandra

Applicable business scenario: Basic writes are already stable, and the upstream system represents entity deletion with a record that retains the primary key while setting all other business fields to `null`. The target table has a stable column structure, and the mapping can cover every column in the table.

Configuration example: Change the delete switch in the Quick Start configuration to `true`. The complete Connector configuration is shown below so that the write before deletion and the subsequent deletion result can be verified independently.

```properties theme={null}
connector.class=com.datastax.oss.kafka.sink.CassandraSinkConnector
tasks.max=1
topics=orders
contactPoints=<cassandra-host>
loadBalancing.localDc=<local-dc>
topic.orders.app.orders.mapping=order_id=key, customer_id=value.customer_id, status=value.status
topic.orders.app.orders.deletesEnabled=true
```

Key points: Automatic deletion does not require the Kafka Value itself to be a tombstone, but the mapping must cover every column in the target table; when all non-primary-key mapped values such as `customer_id` and `status` are `null`, the Connector deletes the full row by `order_id`. If only some columns are mapped, the record still follows the write path. Custom `query` and automatic deletion are mutually exclusive; do not enable both.

### Scale Write Throughput by Topic Partition

Applicable business scenario: A single Task writes correctly, but Kafka consumer lag continues to grow; the `orders` Topic has multiple Partitions, and the Cassandra cluster and Worker still have available capacity. You want to gradually increase parallel consumption and the number of in-flight requests.

Configuration example: The following configuration raises the Task limit to `4` on the safe-write baseline from Quick Start and uses explicit initial values for requests and batches. Validate it in a load-test environment before production, then adjust each value according to actual latency and Cassandra load.

```properties theme={null}
connector.class=com.datastax.oss.kafka.sink.CassandraSinkConnector
tasks.max=4
topics=orders
contactPoints=<cassandra-host>
loadBalancing.localDc=<local-dc>
maxConcurrentRequests=250
maxNumberOfRecordsInBatch=64
topic.orders.app.orders.mapping=order_id=key, customer_id=value.customer_id, status=value.status
topic.orders.app.orders.deletesEnabled=false
```

Key points: `tasks.max=4` is only the Task limit; actual parallelism does not exceed the number of Topic Partitions. `maxConcurrentRequests` is shared by all Tasks of Connectors with the same name within the same Worker JVM; adding Workers also adds independent Sessions and total concurrency. Batches are bucketed by target table and Cassandra Routing Key, and `64` is not a universally optimal value. Compare Kafka Lag, write throughput, request latency, timeouts, and Cassandra load before and after changes, and do not rely on global ordering across Partitions, Tasks, or batches.

## Monitoring

### What to Monitor

Monitor Kafka Connect health, Connector and Task status, throughput, latency, Offset commits, errors, retries, and Worker JVM signals; monitor DLQ activity only when the corresponding error handling is enabled.

### Import the Grafana Dashboard

Confirm that Connect metrics are connected to a Grafana data source and that collection labels meet the dashboard's filtering requirements; download the [Kafka Connect Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), import the JSON in Grafana, and select the corresponding data source.

## Limitations

* The Connector does not create Keyspaces, Tables, or columns automatically, nor does it automatically adapt to incompatible table-schema changes; after changing the target Schema or mapping, restart the Task so that it rereads metadata and prepares CQL again.
* Cassandra writes and Kafka Offset commits are not atomic transactions; failure recovery may replay records that were written successfully but whose Offset had not been committed, so exactly-once or duplicate-free guarantees are not provided.
* When the same record is written to multiple target tables, no cross-table transaction is provided; if a failure occurs after some tables have been written successfully, replay may write those tables again.
* Counter, `now()`, TTL, custom CQL, and deletes each have their own replay side effects; do not assume that the idempotency characteristics of ordinary Upserts apply to these write modes.

## FAQ

### The Task reports an invalid mapping or target column at startup

The target Keyspace, Table, or column may not exist, the mapping may omit a primary-key column, or the Topic, Keyspace, or Table in a dynamic configuration key may not match the actual resource. First verify the target table Schema, then check every target column and field expression in `topic.<topic>.<keyspace>.<table>.mapping`; when CQL is generated automatically, ensure that every primary-key column is mapped, then restart the Connector after making changes.

### The Connector is running, but some records are not written to Cassandra

Writes may fail when a record lacks a required mapped field, a primary-key value is `null`, a type cannot be converted, or a Cassandra request fails; enabling `ignoreErrors=Driver` or `All` may also allow the Offset to pass failed records. Check Task logs, error counts, and consumer Offsets, confirm that the mapped fields match the message structure, and prefer `ignoreErrors=None` while investigating and fixing the root cause; records that were ignored must be compensated for separately according to the business process.

### Writes start failing after the Cassandra table schema is changed

The Connector reads table metadata, validates the mapping, and prepares CQL when the Task starts; compatibility during runtime depends on the cached mapping and Prepared Statement. After a Schema change, first confirm that the mapping still covers valid columns and primary keys, then restart the relevant Task to trigger validation again. For incompatible changes such as dropping columns, changing primary keys, or changing types, plan a migration and message-compatibility strategy first.