DataStax Cassandra Sink Connector - AutoMQ Documentation

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. 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.

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.

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.

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, 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.

​Overview

​Prerequisites

​License

​Quick Start

​Configuration

​Kafka Connect and Subscription

​connector.class

​tasks.max

​topics

​topics.regex

​Cassandra Connection

​contactPoints

​port

​loadBalancing.localDc

​cloud.secureConnectBundle

​Authentication

​auth.provider

​auth.username

​auth.password

​auth.gssapi.keyTab

​auth.gssapi.principal

​auth.gssapi.service

​TLS

​ssl.provider

​ssl.cipherSuites

​ssl.hostnameValidation

​ssl.keystore.password

​ssl.keystore.path

​ssl.openssl.keyCertChain

​ssl.openssl.privateKey

​ssl.truststore.password

​ssl.truststore.path

​Write Throughput and Requests

​maxConcurrentRequests

​maxNumberOfRecordsInBatch

​compression

​queryExecutionTimeout

​connectionPoolLocalSize

​Error Handling and Metrics

​ignoreErrors

​jmx

​metricsHighestLatency

​Topic Data Conversion

​topic.<topic>.codec.locale

​topic.<topic>.codec.timeZone

​topic.<topic>.codec.timestamp

​topic.<topic>.codec.date

​topic.<topic>.codec.time

​topic.<topic>.codec.unit

​Table Mapping and Write Semantics

​topic.<topic>.<keyspace>.<table>.mapping

​topic.<topic>.<keyspace>.<table>.deletesEnabled

​topic.<topic>.<keyspace>.<table>.consistencyLevel

​topic.<topic>.<keyspace>.<table>.ttl

​topic.<topic>.<keyspace>.<table>.nullToUnset

​topic.<topic>.<keyspace>.<table>.ttlTimeUnit

​topic.<topic>.<keyspace>.<table>.timestampTimeUnit

​topic.<topic>.<keyspace>.<table>.query

​Java Driver Passthrough

​datastax-java-driver.<driver-path>

​Best Practices

​Synchronize Business Deletes to Cassandra

​Scale Write Throughput by Topic Partition

​Monitoring

​What to Monitor

​Import the Grafana Dashboard

​Limitations

​FAQ

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

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

​Writes start failing after the Cassandra table schema is changed