Skip to main content

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