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

# Scylla CDC Source Connector

> Configure and operate the Scylla CDC Source Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting.

## Overview

The Scylla CDC Source Connector reads row-level changes from CDC-enabled Scylla tables and writes INSERT, UPDATE, and supported DELETE events to Kafka. By default, each source table maps to a topic named `<topic-prefix>.<keyspace>.<table>`. Record values use the Debezium change-event structure, and record keys can retain the complete partition and clustering keys.

The Connector continuously reads only the Scylla CDC log and does not take an initial snapshot of the base table. It is suitable for feeding incremental changes from tables such as orders, accounts, and device states into event streams, cache updates, search indexes, or audit pipelines. New deployments should explicitly use the `advanced` output format to obtain direct field values, complex types, and configurable before and after images. The default `legacy` format is deprecated.

## Prerequisites

* Before capture begins, the target Scylla tables must exist and have CDC enabled. The Quick Start and other basic examples in this document use `cdc.include.after=only-updated`, so the target tables must also have CDC postimage enabled. If a `before` image is required, CDC preimage must also be enabled. The CDC log retention period should cover the expected outage and recovery window.
* The Scylla account used by the Connector must have permission to read the target keyspaces, tables, CDC logs, and related metadata. When TLS is enabled, the certificates, truststore, keystore, or private key files must be readable by the Connect Worker.

## License

Uses Apache License 2.0.

## Quick Start

Prepare a Connect Cluster, Kafka, and a CDC-enabled Scylla table. Confirm that the Connect Worker can access Scylla and Kafka and has the required permissions. For preparation and management procedures, see [Manage Connectors](../manage-connectors).

```properties theme={null}
connector.class=com.scylladb.cdc.debezium.connector.ScyllaConnector
scylla.cluster.ip.addresses=<scylla-host>:9042
scylla.table.names=<keyspace>.<table>
topic.prefix=scylla
scylla.ssl.enabled=false
cdc.output.format=advanced
cdc.include.primary-key.placement=kafka-key,payload-after,payload-before
cdc.include.after=only-updated
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
```

Replace `<scylla-host>`, `<keyspace>`, and `<table>` with the actual resources, and enable CDC postimage on the target table before starting the Connector. If authentication is enabled in Scylla, also add `scylla.user` and `scylla.password`. TLS should normally also be enabled in production. With this configuration, the `after` field for an INSERT contains the complete new row, while the `after` field for an UPDATE contains the primary key and the columns that actually changed. A DELETE is identified by `op=d` and the Kafka key. The example reads only changes that occur after startup or can be recovered from the CDC log; it does not emit existing rows from the base table.

## Configuration

### Connector Identity and Source Tables

#### `connector.class`

Specifies the Connector implementation class.

* **Type**: `string`
* **Default**: None
* **Importance**: High
* **Valid values / Notes**: Must be set to `com.scylladb.cdc.debezium.connector.ScyllaConnector`.
* **Required**: Yes

#### `scylla.table.names`

Specifies the Scylla tables to read.

* **Type**: `list`
* **Default**: `null`
* **Importance**: High
* **Valid values / Notes**: Required. Use comma-separated `keyspace.table` names. Unquoted keyspace and table names may contain only 1 to 48 ASCII letters, digits, or underscores. Quoted identifiers are not supported. CDC must be enabled on every table. A missing table does not cause validation to fail; the Connector waits for it to appear.
* **Required**: Yes

#### `topic.prefix`

Sets the namespace prefix for data topics and heartbeat topics.

* **Type**: `string`
* **Default**: `null`
* **Importance**: High
* **Valid values / Notes**: Required, and should be unique across Connector instances. The default naming strategy uses this prefix to generate `<topic-prefix>.<keyspace>.<table>`.
* **Required**: Yes

#### `topic.naming.strategy`

Specifies the Debezium topic naming strategy implementation.

* **Type**: `class`
* **Default**: `io.debezium.schema.SchemaTopicNamingStrategy`
* **Importance**: Medium
* **Valid values / Notes**: A custom class must implement the Debezium `TopicNamingStrategy`. Together with `topic.prefix` and `heartbeat.topics.prefix`, it determines topic names.

### Scylla Connection and Authentication

#### `scylla.cluster.ip.addresses`

Specifies the Scylla CQL contact points.

* **Type**: `list`
* **Default**: `null`
* **Importance**: High
* **Valid values / Notes**: Required. Use comma-separated `host:port` entries. Every entry must explicitly provide a numeric port and be reachable from the Worker. The current parser does not support IPv6 literals that contain colons.
* **Required**: Yes

#### `scylla.local.dc`

Specifies the local data center to prefer.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Low
* **Valid values / Notes**: The name must match the Scylla cluster topology metadata. When omitted, the driver makes the selection.

#### `scylla.user`

Specifies the CQL username.

* **Type**: `string`
* **Default**: `null`
* **Importance**: High
* **Valid values / Notes**: When omitted, an unauthenticated connection is used. When set, `scylla.password` must also be set.

#### `scylla.password`

Specifies the CQL password.

* **Type**: `password`
* **Default**: `null`
* **Importance**: High
* **Valid values / Notes**: When set, `scylla.user` must also be set. Supply it through a controlled secret or Config Provider; do not write it to ordinary configuration files or logs.

#### `scylla.consistency.level`

Sets the consistency level used when reading the CDC log.

* **Type**: `string`
* **Default**: `quorum`
* **Importance**: Medium
* **Valid values / Notes**: Use a consistency level supported by the underlying Scylla CDC driver. This value affects only CDC log queries.

#### `scylla.query.options.fetch.size`

Sets the CQL query page size.

* **Type**: `int`
* **Default**: `0`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative integer. `0` uses the driver default, and a positive value sets the number of rows read per page.

### TLS

#### `scylla.ssl.enabled`

Controls whether TLS connections to Scylla are enabled.

* **Type**: `boolean`
* **Default**: `null`
* **Importance**: High
* **Valid values / Notes**: Should be explicitly set to `true` or `false`. Other TLS settings are read only when this is set to `true`.

#### `scylla.ssl.provider`

Selects the TLS implementation provider.

* **Type**: `string`
* **Default**: `jdk`
* **Importance**: Low
* **Valid values / Notes**: Valid values are `jdk`, `openssl`, or `openssl_refcnt`. This setting takes effect only when `scylla.ssl.enabled=true`.

#### `scylla.ssl.truststore.path`

Specifies the path to the Java truststore file.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Medium
* **Valid values / Notes**: Used only for TLS. The file must exist and be readable by the Worker. The Connector does not validate the path in advance.

#### `scylla.ssl.truststore.password`

Specifies the Java truststore password.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Medium
* **Valid values / Notes**: Used with `scylla.ssl.truststore.path`. Although its ConfigDef type is `string`, it should still be managed as a secret.

#### `scylla.ssl.keystore.path`

Specifies the path to the Java keystore file.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Medium
* **Valid values / Notes**: Used for TLS connections that require a client certificate. The file must be readable by the Worker. The Connector does not validate the path in advance.

#### `scylla.ssl.keystore.password`

Specifies the Java keystore password.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Medium
* **Valid values / Notes**: Used with `scylla.ssl.keystore.path`. Although its ConfigDef type is `string`, it should still be managed as a secret.

#### `scylla.ssl.cipherSuites`

Restricts the TLS cipher suites.

* **Type**: `list`
* **Default**: `null`
* **Importance**: High
* **Valid values / Notes**: Takes effect only with TLS. When omitted, the defaults of the selected TLS Provider are used. Explicit values must be supported by the Provider.

#### `scylla.ssl.openssl.keyCertChain`

Specifies the path to the OpenSSL client certificate chain.

* **Type**: `string`
* **Default**: `null`
* **Importance**: High
* **Valid values / Notes**: Used only with an OpenSSL Provider. The file must be readable by the Worker and should match `scylla.ssl.openssl.privateKey`.

#### `scylla.ssl.openssl.privateKey`

Specifies the path to the OpenSSL client private key.

* **Type**: `string`
* **Default**: `null`
* **Importance**: High
* **Valid values / Notes**: Used only with an OpenSSL Provider. Restrict the file permissions and ensure that the Worker can read the file. The Connector does not validate the path or the certificate and key pairing.

### Startup, Read Windows, and Buffering

#### `scylla.initial.lookback.ms`

Sets the lookback period for the initial read when no stored offset exists.

* **Type**: `long`
* **Default**: `0`
* **Importance**: Medium
* **Valid values / Notes**: Must be non-negative. `0` reads from the starting window of the first CDC Generation. A positive value looks back by the specified duration from the current time. It takes effect only for a specific task that has no stored offset and cannot recover data older than the CDC retention period.

#### `streaming.delay.ms`

Sets the delay before streaming begins.

* **Type**: `long`
* **Default**: `0`
* **Importance**: Low
* **Valid values / Notes**: Must be non-negative. The Scylla snapshot phase does not read data, so this should normally remain `0`.

#### `scylla.query.time.window.size`

Sets the time window covered by each CDC query, in milliseconds.

* **Type**: `int`
* **Default**: `30000`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative integer. A smaller window reduces the work of each query but increases the number of queries. It is also used to construct the initial lookback window.

#### `scylla.confidence.window.size`

Sets the safety window during which the latest CDC interval is not read, in milliseconds.

* **Type**: `int`
* **Default**: `30000`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative integer. A larger value helps avoid the latest CDC data that has not yet stabilized, but increases end-to-end latency. `0` removes this delay.

#### `scylla.minimal.wait.for.window.time`

Sets the minimum wait between consecutive CDC query windows, in milliseconds.

* **Type**: `int`
* **Default**: `0`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative integer. `0` applies no active rate limiting. A positive value can reduce query pressure during catch-up.

#### `poll.interval.ms`

Sets the interval for polling records from the internal change queue, in milliseconds.

* **Type**: `long`
* **Default**: `500`
* **Importance**: Medium
* **Valid values / Notes**: Must be non-negative. A smaller value can reduce response latency after an idle period, but increases polling frequency.

#### `max.batch.size`

Sets the maximum number of records returned by each Poll.

* **Type**: `int`
* **Default**: `2048`
* **Importance**: Medium
* **Valid values / Notes**: Must be a positive integer and should be less than `max.queue.size`.

#### `max.queue.size`

Sets the maximum number of records that the internal change queue can hold.

* **Type**: `int`
* **Default**: `8192`
* **Importance**: Medium
* **Valid values / Notes**: Must be a positive integer and should be greater than `max.batch.size` to retain buffer capacity. When the queue is full, backpressure is applied to the reader.

#### `max.queue.size.in.bytes`

Sets the queue limit in bytes.

* **Type**: `long`
* **Default**: `0`
* **Importance**: Medium
* **Valid values / Notes**: In 2.0.6, this public configuration is not connected to the Scylla queue construction logic and cannot be used to limit memory. Use `max.queue.size` to control the number of records.

### CDC Output

#### `cdc.output.format`

Selects the output format for CDC records.

* **Type**: `string`
* **Default**: `legacy`
* **Importance**: High
* **Valid values / Notes**: Valid values are `legacy` or `advanced`. `legacy` is deprecated and scheduled for removal in 3.0.0. New deployments should explicitly use `advanced` to support direct field values, complex types, and before and after images.
* **Deprecation**: The default value `legacy` is deprecated. The replacement value is `advanced`.

#### `experimental.preimages.enabled`

Controls whether the Legacy format reads Preimage records.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Low
* **Valid values / Notes**: Applies only to `legacy`. Setting it to `true` when `cdc.output.format=advanced` is rejected. For the Advanced format, use `cdc.include.before`.

#### `cdc.include.before`

Controls the before image in Advanced records.

* **Type**: `string`
* **Default**: `none`
* **Importance**: Medium
* **Valid values / Notes**: Valid values are `none`, `full`, or `only-updated`, and the setting applies only to `advanced`. `full` and `only-updated` require CDC preimage to be enabled on all existing target tables.

#### `cdc.include.after`

Controls the after image in Advanced records.

* **Type**: `string`
* **Default**: `none`
* **Importance**: Medium
* **Valid values / Notes**: Valid values are `none`, `full`, or `only-updated`, and the setting applies only to `advanced`. `full` and `only-updated` require CDC postimage to be enabled on all existing target tables.

#### `cdc.include.primary-key.placement`

Sets the locations of the primary key in output records.

* **Type**: `list`
* **Default**: `kafka-key,payload-after,payload-before`
* **Importance**: Medium
* **Valid values / Notes**: The Legacy format accepts only the default combination. The Advanced format can use `kafka-key`, `payload-after`, `payload-before`, `payload-key`, and `kafka-headers`. Although `payload-diff` passes validation, 2.0.6 does not implement its output. The ConfigDef default remains `kafka-key,payload-after,payload-before`, but when using the `advanced` format in 2.0.6, this combination must be set explicitly to generate a non-`null` structured Kafka key. Omitting the configuration may result in a required key schema with a `null` key value. Retain `kafka-key` when relying on key-based partitioning, ordering, or log compaction.

#### `cdc.include.primary-key.payload-key-name`

Sets the field name of the `payload-key` primary-key object.

* **Type**: `string`
* **Default**: `key`
* **Importance**: Low
* **Valid values / Notes**: Used only when the primary-key placement includes `payload-key`. Choose a non-empty name that does not conflict with other top-level fields.

#### `cdc.incomplete.task.timeout.ms`

Sets the timeout for waiting for an Advanced image combination, in milliseconds.

* **Type**: `long`
* **Default**: `15000`
* **Importance**: Low
* **Valid values / Notes**: Must be positive. A change that is still missing a required Preimage or Postimage after this period is logged as an error and discarded. Cleanup depends on subsequent events advancing and is not an exact timer.

#### `tombstones.on.delete`

Controls whether a Tombstone is emitted after a DELETE event.

* **Type**: `boolean`
* **Default**: `true`
* **Importance**: Medium
* **Valid values / Notes**: `true` emits a record with the same key and a null value after the delete event, allowing a log-compacted topic to remove the old value. Removing `kafka-key` prevents normal key-based compaction semantics.

#### `skipped.operations`

Specifies the Debezium operation types that are not emitted.

* **Type**: `list`
* **Default**: `t`
* **Importance**: Low
* **Valid values / Notes**: Valid values are `c`, `u`, `d`, `t`, or `none`. Truncate is skipped by default. This configuration cannot make the Connector support multi-row partition deletes or range deletes that cannot be converted into row events.

#### `event.processing.failure.handling.mode`

Sets the behavior when Debezium event processing fails.

* **Type**: `string`
* **Default**: `fail`
* **Importance**: Medium
* **Valid values / Notes**: Valid values are `fail`, `warn`, or `ignore`. `fail` stops the task. `warn` and `ignore` may skip problematic events and should be used only when data gaps are acceptable to the business.

#### `transaction.metadata.factory`

Specifies the Debezium transaction metadata factory.

* **Type**: `class`
* **Default**: `io.debezium.pipeline.txmetadata.DefaultTransactionMetadataFactory`
* **Importance**: Low
* **Valid values / Notes**: This is an advanced extension point. Version 2.0.6 does not expose a configuration that enables transaction metadata, so replacing this factory does not cause the Connector to produce transaction metadata.

### Retries and Error Recovery

#### `retriable.restart.connector.wait.ms`

Sets how long Debezium waits before restarting after a retriable exception, in milliseconds.

* **Type**: `long`
* **Default**: `10000`
* **Importance**: Low
* **Valid values / Notes**: Must be positive. This setting and the Scylla Worker retry parameters operate at different layers.

#### `errors.max.retries`

Sets the maximum number of retries for Debezium connection errors.

* **Type**: `int`
* **Default**: `-1`
* **Importance**: Low
* **Valid values / Notes**: `-1` means unlimited retries, `0` means no retry, and a positive value means a finite number of retries. Do not confuse it with `worker.max.retries`.

#### `worker.retry.backoff.base`

Sets the base duration for Scylla Worker exponential backoff, in milliseconds.

* **Type**: `int`
* **Default**: `50`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative integer and should not be greater than `worker.maximum.backoff`. The Connector does not validate the relationship between the two values.

#### `worker.maximum.backoff`

Sets the maximum Scylla Worker exponential backoff, in milliseconds.

* **Type**: `int`
* **Default**: `30000`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative integer and should not be less than `worker.retry.backoff.base`.

#### `worker.jitter.percentage`

Sets the random jitter percentage for Worker retry backoff.

* **Type**: `int`
* **Default**: `20`
* **Importance**: Low
* **Valid values / Notes**: Use `1` through `100`. Version 2.0.6 validates only that the value is positive and does not enforce an upper limit, but values greater than `100` should not be configured.

#### `worker.max.retries`

Sets the total number of Scylla Worker execution attempts.

* **Type**: `int`
* **Default**: `20`
* **Importance**: Low
* **Valid values / Notes**: Accepts a positive integer or `-1`. `1` means a single execution with no retry, and `-1` means unlimited attempts. `0` and values less than `-1` are invalid.

### Tasks and Connection Pooling

#### `tasks.max`

Sets the maximum number of tasks that Kafka Connect can create.

* **Type**: `int`
* **Default**: `1`
* **Importance**: High
* **Valid values / Notes**: Must be at least `1`. The actual number of tasks may be lower, depending on the current CDC vnode or tablet work assignments. Increasing this value does not guarantee linear throughput growth and affects the total number of connections and queues.

#### `tasks.max.enforce`

Controls whether the framework enforces the `tasks.max` limit.

* **Type**: `boolean`
* **Default**: `true`
* **Importance**: Low
* **Valid values / Notes**: Kafka Connect 3.9.1 has deprecated this configuration and plans to remove it. Keep it set to `true` and plan concurrency through `tasks.max`.
* **Deprecation**: Yes; there is no direct replacement configuration.

#### `worker.shared.session.enabled`

Controls whether tasks in the same Worker JVM share a Scylla Session.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Low
* **Valid values / Notes**: When set to `true`, tasks share a Session only if their connection and pool configurations are identical. Sharing is limited to a single JVM. It can reduce the number of connections but increases contention on the same pool.

#### `worker.pooling.core.pool.local`

Sets the target number of core connections for each local Scylla node.

* **Type**: `int`
* **Default**: `1`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative integer and should not be greater than `worker.pooling.max.pool.local`. The Connector does not validate the relationship between the two values.

#### `worker.pooling.max.pool.local`

Sets the maximum number of connections for each local Scylla node.

* **Type**: `int`
* **Default**: `1`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative integer and should not be less than `worker.pooling.core.pool.local`.

#### `worker.pooling.max.requests.per.connection`

Sets the maximum number of concurrent requests allowed on a single connection.

* **Type**: `int`
* **Default**: `256`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative integer. When Session sharing is disabled, each task uses an independent connection pool, so the aggregate concurrency limit grows with the actual task count. When sharing is enabled, identically configured tasks in the same Worker JVM share a Session, and this value applies to each connection in the shared pool.

#### `worker.pooling.max.queue.size`

Sets the request queue limit for each connection pool.

* **Type**: `int`
* **Default**: `256`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative integer. Requests that exceed capacity or time out while waiting are rejected. When Session sharing is disabled, each task uses an independent connection pool. When sharing is enabled, identically configured tasks in the same Worker JVM share and contend for the same pool.

#### `worker.pooling.pool.timeout.ms`

Sets the maximum wait for obtaining a connection from a host connection pool, in milliseconds.

* **Type**: `int`
* **Default**: `5000`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative integer. Adjust it together with the request queue, connection count, and Scylla latency.

### Heartbeats

#### `heartbeat.interval.ms`

Sets the heartbeat record interval, in milliseconds.

* **Type**: `int`
* **Default**: `30000`
* **Importance**: Medium
* **Valid values / Notes**: Must be a positive integer and cannot be set to `0`. Heartbeats advance and persist the CDC read position during low-traffic periods.

#### `heartbeat.topics.prefix`

Sets the name prefix for heartbeat topics.

* **Type**: `string`
* **Default**: `__debezium-heartbeat`
* **Importance**: Low
* **Valid values / Notes**: The naming strategy combines this value with `topic.prefix` to generate the heartbeat topic. It should comply with Kafka topic naming rules.

### Converters and Extension Processing

#### `key.converter`

Sets the Kafka Connect Converter for the SourceRecord key.

* **Type**: `class`
* **Default**: `null`
* **Importance**: Low
* **Valid values / Notes**: When omitted, the Worker configuration is inherited. The Converter must be instantiable and support structured primary keys. If `kafka-key` is removed from the primary-key placements, the record key may be `null`.

#### `value.converter`

Sets the Kafka Connect Converter for the SourceRecord value.

* **Type**: `class`
* **Default**: `null`
* **Importance**: Low
* **Valid values / Notes**: When omitted, the Worker configuration is inherited. The Converter must support the Debezium Envelope `STRUCT` schema and the data types of the selected output format.

#### `converters`

Sets Debezium custom Converter aliases.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Low
* **Valid values / Notes**: Use comma-separated aliases and configure the implementation class and options with each alias prefix. The implementation class must be present on the plugin Classpath.

#### `post.processors`

Sets Debezium Post Processor aliases.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Low
* **Valid values / Notes**: Use comma-separated aliases and enable only installed implementations that are compatible with the current Debezium runtime.

### Signals and Notifications

#### `signal.enabled.channels`

Sets the enabled Debezium signal channels.

* **Type**: `list`
* **Default**: `source`
* **Importance**: Medium
* **Valid values / Notes**: Configure only channels supported by the current plugin Classpath and Debezium runtime. The Source channel is enabled by default.

#### `signal.poll.interval.ms`

Sets the polling interval for enabled signal channels, in milliseconds.

* **Type**: `long`
* **Default**: `5000`
* **Importance**: Medium
* **Valid values / Notes**: Must be positive. The Scylla database signal table is not integrated, so this value cannot enable database-table signals.

#### `signal.data.collection`

Specifies the database signal collection.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Medium
* **Valid values / Notes**: The Scylla task in 2.0.6 does not integrate database signal table mapping, so this configuration cannot enable signals through a Scylla table. Use another supported signal channel.

#### `notification.enabled.channels`

Sets the enabled Debezium notification channels.

* **Type**: `list`
* **Default**: `null`
* **Importance**: Medium
* **Valid values / Notes**: Configure only channels available on the current plugin Classpath. When a Sink notification channel is enabled, `notification.sink.topic.name` must also be set.

#### `notification.sink.topic.name`

Specifies the Kafka topic used by the Sink notification channel.

* **Type**: `string`
* **Default**: `null`
* **Importance**: High
* **Valid values / Notes**: Required only when the notification channels include Sink. The topic must be valid, and the Connector must have write permission.

### Observability

#### `custom.metric.tags`

Adds custom tags to Debezium JMX metrics.

* **Type**: `list`
* **Default**: `null`
* **Importance**: Low
* **Valid values / Notes**: Use a list of `key=value` entries. Tag values should be stable, must not contain passwords, tokens, or other sensitive information, and must comply with JMX ObjectName constraints.

### Registered but Unsupported Snapshot Configurations

#### `snapshot.mode.custom.name`

Specifies a custom Snapshotter name.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Medium
* **Valid values / Notes**: Although this field is publicly registered, version 2.0.6 always uses a snapshot process that reads neither schemas nor table data and does not expose `snapshot.mode`. Setting this field has no effect.

#### `snapshot.mode.configuration.based.snapshot.data`

Controls whether Configuration-based mode snapshots data.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid values / Notes**: Version 2.0.6 does not expose a usable Configuration-based Snapshot mode. This configuration cannot request a base-table snapshot.

#### `snapshot.mode.configuration.based.snapshot.schema`

Controls whether Configuration-based mode snapshots the schema.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid values / Notes**: The Scylla snapshot implementation in 2.0.6 does not capture schemas. This configuration has no effect.

#### `snapshot.mode.configuration.based.start.stream`

Controls whether Configuration-based mode starts streaming.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid values / Notes**: Version 2.0.6 does not expose a usable Configuration-based Snapshot mode. This configuration has no effect.

#### `snapshot.mode.configuration.based.snapshot.on.schema.error`

Controls whether a Configuration-based snapshot is rerun after a schema error.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid values / Notes**: The Scylla snapshot implementation reads neither schemas nor table data. This configuration has no effect.

#### `snapshot.mode.configuration.based.snapshot.on.data.error`

Controls whether a Configuration-based snapshot is rerun after a data error.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid values / Notes**: The Scylla snapshot implementation reads neither schemas nor table data. This configuration has no effect.

#### `incremental.snapshot.watermarking.strategy`

Sets the incremental snapshot watermarking strategy.

* **Type**: `string`
* **Default**: `INSERT_INSERT`
* **Importance**: Low
* **Valid values / Notes**: This field is publicly registered by Debezium, but version 2.0.6 does not provide a usable Scylla incremental snapshot capability. It cannot be used to read existing base-table data.

## Best Practices

### Preserve Before and After Images for Update Events

**Applicable scenario**: Downstream systems need to compare field changes, generate audit records, or maintain a complete materialized view, and CDC preimage and postimage can be enabled on the target tables before integration. The expected result is that UPDATE events contain complete `before` and `after` images, while INSERT and DELETE events continue to follow their respective image boundaries.

**Configuration example**:

```properties theme={null}
connector.class=com.scylladb.cdc.debezium.connector.ScyllaConnector
scylla.cluster.ip.addresses=<scylla-host>:9042
scylla.table.names=<keyspace>.<table>
topic.prefix=scylla
scylla.ssl.enabled=false
cdc.output.format=advanced
cdc.include.primary-key.placement=kafka-key,payload-after,payload-before
cdc.include.before=full
cdc.include.after=full
cdc.incomplete.task.timeout.ms=30000
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
```

**Key considerations**: Enable the corresponding CDC preimage and postimage on all target tables before starting the Connector. Complete images increase Scylla CDC storage, read traffic, and Kafka message size. If only updated columns are needed, use `only-updated` instead. When an image record remains missing beyond the timeout, the corresponding logical change is discarded, so set the timeout based on error logs and downstream data-integrity requirements.

### Limit the CDC History Lookback During Initial Integration

**Applicable scenario**: The target tables have already been continuously producing CDC logs, but the initial deployment needs only a recent range of changes and should not catch up from the first still-visible CDC Generation. The expected result is that tasks without stored offsets start reading from approximately one hour before the current time instead of producing a base-table snapshot.

**Configuration example**:

```properties theme={null}
connector.class=com.scylladb.cdc.debezium.connector.ScyllaConnector
scylla.cluster.ip.addresses=<scylla-host>:9042
scylla.table.names=<keyspace>.<table>
topic.prefix=scylla
scylla.ssl.enabled=false
scylla.initial.lookback.ms=3600000
cdc.output.format=advanced
cdc.include.primary-key.placement=kafka-key,payload-after,payload-before
cdc.include.after=only-updated
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
```

**Key considerations**: CDC postimage must be enabled on the target tables so that both historical and real-time events carry consumable non-primary-key change values. `scylla.initial.lookback.ms` takes effect only when the corresponding task has no stored offset. It cannot override an existing recovery position or read data that has already aged out of Scylla CDC retention. Determine the lookback duration from the historical range the business permits and the expected catch-up throughput. If existing base-table rows are required, use a separate data initialization process and coordinate deduplication and continuity with subsequent CDC events.

### Increase Concurrency While Controlling Scylla Connection Pressure

**Applicable scenario**: Monitoring shows that the Connector is continuously falling behind, Kafka and the Worker still have processing capacity, but a single task cannot consume the current vnode or tablet assignments in time. The expected result is increased concurrency within the currently assignable work units, while tasks in the same Worker JVM share a Session to avoid multiplying the connection count directly by the number of tasks.

**Configuration example**:

```properties theme={null}
connector.class=com.scylladb.cdc.debezium.connector.ScyllaConnector
tasks.max=4
scylla.cluster.ip.addresses=<scylla-host>:9042
scylla.table.names=<keyspace>.<table>
topic.prefix=scylla
scylla.ssl.enabled=false
cdc.output.format=advanced
cdc.include.primary-key.placement=kafka-key,payload-after,payload-before
cdc.include.after=only-updated
worker.shared.session.enabled=true
worker.pooling.core.pool.local=1
worker.pooling.max.pool.local=2
worker.pooling.max.requests.per.connection=256
worker.pooling.max.queue.size=512
worker.pooling.pool.timeout.ms=5000
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
```

**Key considerations**: CDC postimage must be enabled on the target tables. The actual number of tasks may be lower than `tasks.max`, and topology or CDC Generation changes trigger reassignment, so do not assume that a task is permanently mapped to one vnode or tablet. Session sharing applies only within the same JVM. It reduces connection count while increasing pool contention. Increase concurrency gradually while monitoring Scylla request latency, connection-pool queuing, task retries, Kafka throughput, and end-to-end latency.

## Monitoring

### What to Monitor

Monitor the running status and restart counts of the Connect Cluster, Workers, Connector, and tasks. Also monitor Source Record throughput, end-to-end latency, backlog trends, offset commit latency and failures, errors and retries, and Worker JVM CPU, heap memory, GC, and thread status. If the deployment explicitly enables error tolerance and a DLQ, also monitor DLQ write volume and failures.

### Import the Grafana Dashboard

Download the [AutoMQ Connect Cluster Grafana Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json). Confirm that Grafana has a data source configured to collect Kafka Connect JMX metrics and that labels such as Connector, Task, and Worker are available for filtering. Then upload the JSON file on the Grafana Dashboard import page and select the corresponding data source.

## Limitations

* The Connector does not take an initial snapshot of the base table and does not support incremental snapshots. It can read only changes that are still retained in the Scylla CDC log.
* Delivery semantics are at least once. If a failure occurs after Kafka has received a record but before the Source Offset is persisted, the record may be emitted again after restart. The Connector does not provide Exactly-once delivery or its own deduplication.
* Global ordering is not guaranteed across tables, vnodes, tablets, CDC Generations, tasks, or Kafka partitions. Removing `kafka-key` also weakens the partition-ordering basis normally used for the same primary key.
* When a multi-row partition delete occurs on a table with clustering columns, the Connector does not emit one delete event per row. Row-range deletes are also unsupported. Only a table without clustering columns can represent a partition delete as a single-row delete.
* The Connector does not support row filtering, column Include/Exclude filtering, or dynamic table discovery. The capture scope can be specified only by explicitly listing tables in `scylla.table.names`.
* The `legacy` output format does not support List, Set, Map, Tuple, UDT, or Postimage and is deprecated. Its orphaned Preimages have no timeout cleanup mechanism.
* The Connector does not emit transaction metadata or establish a transaction boundary across the Scylla read position, Kafka records, and Connect Offsets.

## FAQ

### Why is the Kafka key empty, or why does the key converter report that a required field is `null` when using the `advanced` format?

Scylla CDC Source Connector 2.0.6 cannot rely only on the ConfigDef default of `cdc.include.primary-key.placement` in the `advanced` format. Explicitly set `cdc.include.primary-key.placement=kafka-key,payload-after,payload-before` and retain a `key.converter` that supports structured primary keys. After the Connector restarts, the Kafka key contains the source-table primary-key fields. Also confirm that the target table actually defines a primary key and that the effective Connector configuration does not override or remove `kafka-key`.

### Why are existing rows in the table not emitted after the Connector starts?

The Connector reads only the CDC log and does not scan the base table. Check that CDC is enabled on the target table, that the table is listed in `scylla.table.names`, and that the CDC log still retains the required time range. When recent history is required during initial integration, set `scylla.initial.lookback.ms` only when no stored offset exists. When the complete existing dataset is required, first use a separate initialization process and then continue synchronization from an explicit CDC boundary.

### Why are changes from a table still missing after configuration validation succeeds?

A missing target table does not cause configuration validation to fail; the Connector waits for the table to appear. Confirm the `keyspace.table` spelling, ensure that quoted identifiers are not used, and check that the table exists and has CDC enabled. If `cdc.include.before` or `cdc.include.after` is configured, also confirm that Preimage or Postimage is enabled on the table as appropriate, and check the Connector and task status, Scylla permissions, and error logs.

### Why do duplicate events appear after a restart?

The Connector uses at-least-once delivery. If Kafka acknowledges a record but the corresponding Source Offset has not yet been flushed successfully, a restart recovers from the previous persisted position and emits some events again. Retain the complete primary key as the Kafka key, implement idempotent writes or deduplication downstream using fields such as the primary key, operation type, and source timestamp, and monitor offset commit failures and latency. Do not rely on the Connector to provide Exactly-once delivery.

### Why do some delete operations have no corresponding Kafka event?

First confirm that `skipped.operations` does not include `d`. A normal row delete emits a delete event, but a multi-row partition delete on a table with clustering columns and a row-range delete are not expanded into per-row events. If the business must capture these deletes, issue explicit per-row deletes or supplement the unrepresentable bulk-delete semantics with application-level events.

### Why are changes missing after `before` or `after` is enabled?

The Advanced format must combine the primary change with the corresponding Preimage or Postimage. Confirm that all target tables have the required CDC options enabled and check for incomplete-combination errors that exceed `cdc.incomplete.task.timeout.ms`. Do not mask persistently missing images by only increasing the timeout. First check the Scylla CDC configuration, log retention, read latency, and Connector errors.