Skip to main content

​
Overview

The ScyllaDB Sink Connector writes records from Kafka topics to ScyllaDB. The Connector uses the Kafka record key and value to form the target table’s primary key and columns, and derives the table name from the topic name: periods and hyphens in the topic are replaced with underscores. By default, the Connector can create or extend the table schema from the Connect Schema, and can also configure column mappings, consistency levels, TTL, and delete handling for each topic. The record key and value should be object-shaped (Connect Struct or Map). A non-null value is written; when delete handling is enabled, a null value can delete the target table row by record key. Kafka Connect manages topic subscription, task assignment, and offset commits, while the Connector manages the ScyllaDB session, table schema handling, and write requests.

​
Prerequisites

  • Prepare a ScyllaDB cluster, target keyspace, and, when automatic keyspace creation is disabled, a pre-created keyspace that the Connect Worker can access.
  • Use structured Kafka keys and values; Schema-less Map data requires a compatible ScyllaDB table to be created in advance because the Connector does not infer or modify the table schema from Schema-less data.
  • When enabling TLS, prepare readable truststore or keystore files and their passwords; when enabling authentication, also provide a username and password.

​
License

Licensed under Apache License 2.0.

​
Quick Start

Prepare Connect Cluster, Kafka, and ScyllaDB in advance, and confirm network connectivity and access permissions. For preparation and management instructions, see Manage Connectors.
Replace <topic-name>, <scylladb-host>, and <keyspace-name> with the actual Kafka topic, ScyllaDB address, and keyspace. Do not set topics.regex at the same time. The Worker Converter should convert record keys and values to Struct or Map. By default, the Connector creates the keyspace and tables, uses LOCAL QUORUM, enables delete handling, and enables the ScyllaDB Offset table.

​
Configuration

​
ScyllaDB Connection

​
scylladb.contact.points

ScyllaDB contact points. You can provide a comma-separated address list or JSON containing address mappings.
  • Type: string
  • Default: localhost
  • Importance: High
  • Valid values / notes: Addresses must be accessible to the Connect Worker; Connector validation attempts to establish a ScyllaDB session.

​
scylladb.port

ScyllaDB port used for public contact points.
  • Type: int
  • Default: 9042
  • Importance: Medium
  • Valid values / notes: 1 to 65535; JSON private-address mappings can specify a separate port for each address.

​
scylladb.loadbalancing.localdc

Local data center name used by the ScyllaDB driver.
  • Type: string
  • Default: Empty string
  • Importance: High
  • Valid values / notes: Case-sensitive; when empty, no local data center is explicitly specified.

​
scylladb.security.enabled

Whether to enable ScyllaDB username and password authentication.
  • Type: boolean
  • Default: false
  • Importance: High
  • Valid values / notes: When set to true, scylladb.username and scylladb.password must both be set.

​
scylladb.username

ScyllaDB authentication username.
  • Type: string
  • Default: null
  • Importance: High
  • Valid values / notes: Must be set together with scylladb.password; required when authentication is enabled.

​
scylladb.password

ScyllaDB authentication password.
  • Type: password
  • Default: null
  • Importance: High
  • Valid values / notes: Must be set together with scylladb.username; do not write it to logs or documentation examples.

​
scylladb.compression

Protocol compression method used by the ScyllaDB driver.
  • Type: string
  • Default: none
  • Importance: Low
  • Valid values / notes: none, lz4, or snappy, in lowercase.

​
TLS

​
scylladb.ssl.enabled

Whether to enable the ScyllaDB TLS connection.
  • Type: boolean
  • Default: false
  • Importance: High
  • Valid values / notes: After setting to true, set truststore, keystore, passwords, cipher suites, and hostname verification options as needed.

​
scylladb.ssl.truststore.path

Path to the TLS truststore file.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid values / notes: Used only when scylladb.ssl.enabled=true; when set, the file must be readable.

​
scylladb.ssl.truststore.password

TLS truststore password.
  • Type: password
  • Default: null
  • Importance: Medium
  • Valid values / notes: Used only when scylladb.ssl.enabled=true; this is a sensitive value.

​
scylladb.ssl.keystore.path

Path to the TLS keystore file.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid values / notes: Used only when scylladb.ssl.enabled=true; when set, the file must be readable.

​
scylladb.ssl.keystore.password

TLS keystore password.
  • Type: password
  • Default: null
  • Importance: Medium
  • Valid values / notes: Used only when scylladb.ssl.enabled=true; this is a sensitive value.

​
scylladb.ssl.cipherSuites

List of cipher suites allowed for TLS.
  • Type: list
  • Default: Empty list
  • Importance: High
  • Valid values / notes: Used only when scylladb.ssl.enabled=true; an empty list leaves the driver’s cipher-suite selection unchanged.

​
scylladb.ssl.hostname.verification

Whether to enable TLS hostname verification.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / notes: Used only when scylladb.ssl.enabled=true.

​
Write Behavior

​
scylladb.consistency.level

Consistency level used when writing to ScyllaDB.
  • Type: string
  • Default: LOCAL_QUORUM
  • Importance: High
  • Valid values / notes: ANY, ONE, TWO, THREE, QUORUM, ALL, LOCAL_QUORUM, EACH_QUORUM, SERIAL, LOCAL_SERIAL, or LOCAL_ONE; a matching topic-level configuration overrides this value.

​
scylladb.deletes.enabled

Whether to treat null values as delete candidates and delete the target row by record key.
  • Type: boolean
  • Default: true
  • Importance: High
  • Valid values / notes: Deletion requires the record key to contain all primary-key fields of the target table; a matching topic-level configuration overrides this value.

​
scylladb.execute.timeout.ms

Timeout for waiting for asynchronous ScyllaDB operations to complete.
  • Type: long
  • Default: 30000
  • Importance: Low
  • Valid values / notes: In milliseconds; must be greater than or equal to 0; 0 is also accepted.

​
scylladb.ttl

Default TTL for insert statements.
  • Type: int
  • Default: null
  • Importance: Medium
  • Valid values / notes: When null, no USING TTL is added; a matching topic-level ttlSeconds overrides this value.

​
behavior.on.error

How to handle DataException or NullPointerException during record validation or statement construction.
  • Type: string
  • Default: FAIL
  • Importance: Medium
  • Valid values / notes: FAIL throws a Connect exception, LOG records and continues, and IGNORE records at trace level and continues; this does not replace Kafka Connect’s errors.tolerance.

​
Keyspace and Tables

​
scylladb.keyspace

Target keyspace used by the Connector.
  • Type: string
  • Default: None
  • Importance: High
  • Valid values / notes: Required; used to create the keyspace when automatic creation is enabled, and must already exist when automatic creation is disabled.
  • Required: Yes

​
scylladb.keyspace.create.enabled

Whether to automatically create the keyspace when it does not exist.
  • Type: boolean
  • Default: true
  • Importance: High
  • Valid values / notes: When set to false, the configured keyspace must already exist.

​
scylladb.keyspace.replication.factor

Replication factor used when automatically creating the keyspace.
  • Type: int
  • Default: 3
  • Importance: High
  • Valid values / notes: Must be greater than or equal to 1; effective only when scylladb.keyspace.create.enabled=true.

​
scylladb.table.manage.enabled

Whether the Connector creates or alters the target table schema.
  • Type: boolean
  • Default: true
  • Importance: High
  • Valid values / notes: When disabled, the target table and required columns must be prepared in advance; DDL is not generated from Schema-less data.

​
scylladb.table.create.compression.algorithm

Compression algorithm used when creating or altering a table.
  • Type: string
  • Default: none
  • Importance: Medium
  • Valid values / notes: SnappyCompressor, LZ4Compressor, DeflateCompressor, or none; affects creation or alteration only when table management is enabled.

​
scylladb.offset.storage.table

Name of the table that stores Connector ScyllaDB offsets.
  • Type: string
  • Default: kafka_connect_offsets
  • Importance: Low
  • Valid values / notes: Used only when scylladb.offset.storage.table.enable=true.

​
scylladb.offset.storage.table.enable

Whether to create, read, and write the Offset table in ScyllaDB.
  • Type: boolean
  • Default: true
  • Importance: Medium
  • Valid values / notes: When disabled, the ScyllaDB Offset table is skipped; Kafka Connect still manages Kafka offsets according to its Worker mechanism.

​
Topic-to-Table Mapping

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

Configures column mappings for the table derived from the specified topic.
  • Type: string
  • Default: null
  • Importance: Not declared
  • Valid values / notes: Use comma-separated column=key.<field>, value.<field>, or header.<field> entries; the special targets __ttl and __timestamp can also be mapped. The topic, keyspace, and table segments must follow the dynamic configuration-name format.

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

Overrides the write consistency level for the table derived from the specified topic.
  • Type: string
  • Default: Inherits scylladb.consistency.level
  • Importance: Not declared
  • Valid values / notes: Use a consistency level supported by ScyllaDB; affects only the matching topic configuration.

​
topic.<topic>.<keyspace>.<table>.ttlSeconds

Overrides the default TTL for the table derived from the specified topic.
  • Type: int
  • Default: Inherits scylladb.ttl
  • Importance: Not declared
  • Valid values / notes: Parsed as an integer; an empty value inherits the Connector-level TTL. Set it according to the business retention period.

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

Overrides delete handling for the table derived from the specified topic.
  • Type: boolean
  • Default: Inherits scylladb.deletes.enabled
  • Importance: Not declared
  • Valid values / notes: Accepts only true or false, case-insensitively.

​
Kafka Connect Sink Framework

​
connector.class

Specifies the Connector implementation class to load.
  • Type: string
  • Default: None
  • Importance: High
  • Valid values / notes: Use io.connect.scylladb.ScyllaDbSinkConnector.
  • Required: Yes

​
tasks.max

Maximum number of Tasks that the Connector may create.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid values / notes: Must be greater than or equal to 1; actual parallelism also depends on Kafka partition assignment and Worker scheduling.

​
topics

List of Kafka topics to consume.
  • Type: list
  • Default: Empty string
  • Importance: High
  • Valid values / notes: Mutually exclusive with topics.regex; exactly one must be selected; must not contain the DLQ topic.

​
topics.regex

Kafka topics to consume, matched using Java Pattern syntax.
  • Type: string
  • Default: Empty string
  • Importance: High
  • Valid values / notes: Mutually exclusive with topics; exactly one must be selected; must not match the DLQ topic.

​
transforms

List of Kafka Connect SMTs to execute in order.
  • Type: list
  • Default: Empty list
  • Importance: Low
  • Valid values / notes: Each alias requires a corresponding transforms.<alias>.type; the transformed topic name is used when looking up the topic-to-table mapping.

​
predicates

List of Predicate aliases used by SMT conditions.
  • Type: list
  • Default: Empty list
  • Importance: Low
  • Valid values / notes: Used only when a configured SMT references a Predicate; aliases must be unique.

​
errors.tolerance

Range of errors tolerated by the Kafka Connect framework.
  • Type: string
  • Default: none
  • Importance: Medium
  • Valid values / notes: none or all; takes effect separately from behavior.on.error and primarily affects conversion, SMT, and error-reporting stages.

​
errors.retry.timeout

Total time for the Kafka Connect framework to retry failed operations.
  • Type: long
  • Default: 0
  • Importance: Medium
  • Valid values / notes: In milliseconds; -1 means retry continuously. It does not change scylladb.execute.timeout.ms.

​
errors.deadletterqueue.topic.name

DLQ topic name used by the error-record reporter.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / notes: When non-empty, enables the Sink error-record reporting path; this topic must not be consumed by topics or matched by topics.regex.

​
errors.log.enable

Whether to enable Kafka Connect framework-level error logging.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / notes: Independent of behavior.on.error=LOG; controls only framework-level error logging.

​
Best Practices

​
Onboard a Structured Topic and Automatically Manage Tables

Applicable business scenario: You are writing a Kafka topic containing structured keys and values to a new ScyllaDB keyspace for the first time and want the Connector to create the keyspace and tables. When ordinary columns are added later, you can schedule a Connector task restart so the Connector checks and extends the table schema. Configuration example:
Key notes: Fields in the record key participate in primary-key generation, while the value provides the remaining columns; periods and hyphens in the actual Kafka topic are replaced with underscores before determining the table name. Confirm that key fields are stable and uniquely identify target rows before allowing the Connector to initialize the table schema. Schema changes are suitable for adding ordinary columns; do not use this mechanism to change an existing primary key. For ScyllaDB Sink Connector 1.1.9, the first table creation within the same Task lifetime populates the schema cache, and new ordinary columns in later records do not trigger another check; to make the Connector check and add new ordinary columns, restart the Task processing the topic partitions, then process records containing the new columns. Restarting the Connector is not the necessary boundary here; the relevant condition is rebuilding the cache of the Task that processes the records.

​
Derive Tables by Topic and Use Explicit Column Mappings

Applicable business scenario: Multiple Kafka topics write to different derived tables in the same keyspace, and you need to write only selected fields or map key, value, and header fields to explicit ScyllaDB columns. Configuration example:
Key notes: When mapping is enabled, fields not listed are not written; referenced key and value fields must exist in the record Schema, and header mappings require primitive types. The <topic> segment in the dynamic configuration key must use the Connector’s normalized name for the actual Kafka topic: periods and hyphens become underscores, and that result is used to look up topic-level settings. The <keyspace> and <table> segments must still follow the dynamic configuration-name format; they are not independent runtime routing values.

​
Set Retention and Delete Policies per Topic

Applicable business scenario: Different business data requires different retention periods, or only some topics should convert tombstones into primary-key deletes. Configuration example:
Key notes: Topic-level TTL and delete settings override Connector-level values. The topic segment in the dynamic configuration key must use the normalized topic name. TTL applies to the ScyllaDB INSERT statement; when delete handling is enabled and the target table exists, a tombstone results in a DELETE using all primary-key fields from the record key. Neither changes Kafka offsets or event-time semantics. Set them according to business retention policies and verify how tombstones are produced upstream.

​
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 the Grafana data source and that collected labels meet the dashboard’s filtering requirements; download the Kafka Connect Dashboard, import the JSON in Grafana, and select the corresponding data source.

​
Limitations

  • topics and topics.regex are mutually exclusive, and at least one non-empty subscription selector must be configured; the configured DLQ topic must not be subscribed to.
  • Periods and hyphens in the actual Kafka topic name are normalized to underscores; this normalized name is used both for the derived table name and for topic-level dynamic configuration lookup. Topics that collide after normalization point to the same derived table.
  • Schema-less keys or values are not used to generate or alter DDL; a compatible target table must be created in advance before using Schema-less data.
  • Table schema management supports adding ordinary columns only; the primary-key shape of an existing table cannot be changed through mapping.
  • After the schema cache is hit during the same task lifetime, new ordinary columns in records are not checked again; restart the task to check and add new ordinary columns.
  • Kafka offset commits and ScyllaDB business writes are not part of the same transaction; process failures or retries may process records again, so cross-system atomic commits or duplicate-free processing cannot be inferred.

​
FAQ

​
How do I troubleshoot a connection failure to ScyllaDB?

Check scylladb.contact.points, scylladb.port, and network access. When authentication is enabled, confirm that the username and password are both set; when TLS is enabled, confirm that the relevant file paths are readable and the certificate configuration matches. After correcting the connection configuration, redeploy or restart the Connector and observe whether the Task resumes running.

​
Why does the Connector still fail to start after topics is configured?

Check whether topics.regex is also configured or both configurations are empty. A Sink must select exactly one non-empty topics or topics.regex; when errors.deadletterqueue.topic.name is configured, also ensure that the DLQ topic is not in the subscription list or matched by the regular expression.

​
Why do writes fail with an unsupported key or value type?

Confirm that the Worker Converter outputs a Struct or Map rather than a top-level primitive, null key, or unsupported nested Struct. Check that the record key contains the target table’s primary-key fields and that value field types are compatible with ScyllaDB column types. For Schema-less Map scenarios, also confirm that the target table was created in advance.

​
What should I do if a tombstone does not delete the target row?

Confirm that scylladb.deletes.enabled or the matching topic’s deletesEnabled is true, and check that the tombstone key contains all primary-key fields of the target table. If the target table does not exist or the key is incomplete, the Connector cannot construct a valid primary-key delete statement. Also confirm that the topic segment in the dynamic configuration key matches the normalized topic name and that the keyspace and table segments follow the dynamic configuration-name format.