> ## 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. # ScyllaDB Sink Connector > Configure and operate the ScyllaDB Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## 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](../manage-connectors). ```properties theme={null} connector.class=io.connect.scylladb.ScyllaDbSinkConnector topics= scylladb.contact.points= scylladb.keyspace= ``` Replace ``, ``, and `` 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....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.`, `value.`, or `header.` 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...
.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...
.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...
.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..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**: ```properties theme={null} connector.class=io.connect.scylladb.ScyllaDbSinkConnector topics= scylladb.contact.points= scylladb.keyspace= scylladb.keyspace.create.enabled=true scylladb.table.manage.enabled=true ``` **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**: ```properties theme={null} connector.class=io.connect.scylladb.ScyllaDbSinkConnector topics= scylladb.contact.points= scylladb.keyspace= topic...
.mapping=id=key.id,name=value.name,source=header.source ``` **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 `` 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 `` and `
` 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**: ```properties theme={null} connector.class=io.connect.scylladb.ScyllaDbSinkConnector topics= scylladb.contact.points= scylladb.keyspace= scylladb.ttl=86400 scylladb.deletes.enabled=false topic...
.ttlSeconds=3600 topic...
.deletesEnabled=true ``` **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](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 * `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.