Debezium SQL Server Source Connector

Overview

The Debezium SQL Server Source Connector sends existing data and subsequent row-level changes from SQL Server business tables to Kafka. It is suitable for downstream data synchronization, cache updates, and change event processing. During initial onboarding, it can read a row snapshot of the tables and then continuously read INSERT, UPDATE, and DELETE operations captured by SQL Server CDC instead of periodically scanning entire tables. By default, each table maps to a data topic named topic.prefix.<database>.<schema>.<table>. Message keys are typically derived from the table’s primary key, while message values contain the operation type, before and after data, and source position. Schema change messages and internal schema history are stored separately. Internal schema history is used for recovery and is not intended for business consumption.

Prerequisites

CDC for databases and tables: Confirm that the SQL Server version, edition, and deployment model support CDC. An administrator must first enable CDC for the database and then for each base table to be synchronized. Enabling CDC at the database level requires sysadmin, while enabling it at the table level requires db_owner. The Connector does not create or refresh capture instances on your behalf.
CDC capture jobs and retention window: Standard SQL Server deployments require SQL Server Agent to be running and the CDC capture and cleanup jobs to operate normally. The retention window must cover the longest expected downtime plus the catch-up period after recovery. Capture mechanisms for managed deployments must be prepared according to their own requirements and must not be managed as though they were a local SQL Server Agent deployment.
Runtime account permissions: The account must be able to read the source tables and captured columns used for the snapshot, and access CDC metadata and change data in every selected database. If a CDC gating role is configured, an account that is not sysadmin or db_owner must also belong to that role. CDC administration privileges do not need to be granted to the persistent runtime account. Querying SQL Server Agent service status separately requires service DMV visibility appropriate to the server version; do not treat VIEW SERVER STATE as a universal requirement for every version. A DBA may instead provide a controlled status query endpoint. A successful login does not mean that all of these read permissions are present.
SQL Server TLS: The Quick Start requires the server to present a trusted TLS certificate whose name matches the connection hostname, and the Connect Worker JVM must trust the certificate chain. If a private CA is used, add the CA to the Worker’s trust configuration first. The driver enables connection encryption and certificate validation by default; disabling encryption or bypassing certificate validation is not a prerequisite for onboarding.
Internal schema history: Prepare a dedicated Kafka history topic for this Connector. Keep it single-partitioned and retain the complete history indefinitely so that the retention policy does not delete records required for recovery. Ensure that the history clients have read and write permissions, as well as permission to create the topic when automatic creation is required. Do not assume that these clients fully inherit Worker authentication settings.

License

Licensed under the Apache License 2.0.

Quick Start

Prepare a Connect Cluster, Kafka, and a SQL Server database and business tables with CDC enabled. Confirm that network connectivity, access permissions, and TLS trust configuration meet the requirements. For creation and management operations, see AutoMQ’s Manage Connectors. The following example uses database account authentication, inherits the message converters configured on the Worker, and establishes an encrypted connection through a trusted certificate.

connector.class=io.debezium.connector.sqlserver.SqlServerConnector
database.hostname=<database-host>
database.names=<database-name>
database.user=<database-user>
database.password=<database-password>
topic.prefix=<topic-prefix>
schema.history.internal.kafka.bootstrap.servers=<kafka-bootstrap-servers>
schema.history.internal.kafka.topic=<schema-history-topic>

Replace the environment placeholders. The hostname must match the certificate name, the topic prefix must be unique among other Connectors, and the history topic must not be shared with business topics or other independent Connectors. This example uses the default port 1433, the default single Task, and the default initial snapshot mode: when no Offset exists, it reads the existing rows from in-scope CDC tables and then continuously synchronizes changes. Supply account credentials through a secure configuration mechanism supported by the deployment, and do not write them to public files or logs. The example depends on a prepared JVM trust chain and cannot be used directly in an environment where certificate trust has not been configured.

Configuration

Connection and Database Scope

`database.hostname`

The resolvable hostname or IP address of SQL Server.

Type: string
Default: None (null)
Importance: High
Valid values / Notes: Required. When TLS is used, the value must match the certificate identity.

`database.port`

The SQL Server connection port.

Type: int
Default: 1433
Importance: High
Valid values / Notes: Specify the port on which the server actually listens.

`database.instance`

The SQL Server named instance.

Type: string
Default: None (null)
Importance: Low
Valid values / Notes: Set only when connecting to a named instance, and verify how its port is resolved.

`database.names`

Selects databases on the same SQL Server connection.

Type: list
Default: None (null)
Importance: High
Valid values / Notes: At runtime, a non-empty, comma-separated list of database names is required. Prepare CDC and permissions for each database. With multiple databases, Tasks are assigned by database; a database is not split by table.

`database.user`

The database connection account.

Type: string
Default: None (null)
Importance: High
Valid values / Notes: Set when using username and password authentication. Other authentication methods are determined by the selected JDBC mode and cannot be enabled merely by omitting the account.

`database.password`

The database connection password.

Type: password
Default: None (null)
Importance: High
Valid values / Notes: Supply it as required by the authentication method, and do not expose it in logs or shared configuration.

`database.query.timeout.ms`

The database query execution timeout in milliseconds.

Type: int
Default: 600000
Importance: Low
Valid values / Notes: 0 means that query execution time is unlimited. This is distinct from the connection timeout.

`connection.validation.timeout.ms`

The maximum time to wait for database connection validation to complete, in milliseconds.

Type: long
Default: 60000
Importance: Low
Valid values / Notes: A positive integer. This is not a data query timeout.

`data.query.mode`

Controls how incremental CDC data is queried.

Type: string
Default: function
Importance: Low
Valid values / Notes: function calls CDC change functions. direct queries change tables directly and requires matching permissions.

`database.sqlserver.agent.status.query`

Queries whether SQL Server Agent is running.

Type: string
Default: SELECT CASE WHEN dss.[status]=4 THEN 1 ELSE 0 END AS isRunning FROM #db.sys.dm_server_services dss WHERE dss.[servicename] LIKE N'SQL Server Agent (%';
Importance: Medium
Valid values / Notes: Returns the isRunning status, where #db is a database replacement marker. This setting is read from the database subconfiguration, so Connector configuration must use the database. prefix; the field suffix is sqlserver.agent.status.query. Service DMV permissions depend on the server version. Verify the result semantics of a custom query with the DBA. A failed query does not mean that Agent has stopped.

`database.<JDBC-property-name>`

Passes connection properties through to the Microsoft SQL Server JDBC driver for TLS, authentication, and driver connection behavior.

Type: string
Default: No fixed default
Importance: Not declared
Valid values / Notes: The property is passed to the driver after removing the database. or driver. prefix. If the same property is supplied under both prefixes, the driver. value takes precedence. Connector connection parameters listed separately still follow their own semantics; do not use pass-through properties to bypass database selection. A property accepted by the driver does not mean that every authentication method or optional dependency is available.

TLS properties include encrypt, trustServerCertificate, trustStore, trustStorePassword, trustStoreType, and hostNameInCertificate. Unless overridden, the driver defaults to encrypt=True and trustServerCertificate=false. These are driver defaults, not ConfigDef defaults for this pass-through family. Encryption does not replace identity validation. If the certificate chain is untrusted or the name does not match, correct the certificate and trust configuration instead of disabling encryption or bypassing validation. Set other JDBC properties according to the driver’s property names, values, and dependency requirements; do not treat the driver’s complete property catalog as a static Connector configuration list.

`driver.*`

Passes properties to the JDBC driver under a separate prefix.

Type: string
Default: No fixed default
Importance: Not declared
Valid values / Notes: Replace * with the actual JDBC property name. The property is passed to the driver after removing the driver. prefix and takes precedence over a same-named driver property under database.. Property availability, driver defaults, and optional authentication dependencies follow the same boundaries as the preceding pass-through family.

Topics and Synchronization Scope

`topic.prefix`

Provides an identifier and namespace for the source server and its event topics.

Type: string
Default: None (null)
Importance: High
Valid values / Notes: Required. Use letters, numbers, hyphens, periods, and underscores, and keep the value unique. Changing it after deployment changes the topic and source partition identity and is not a transparent rename.

`topic.naming.strategy`

Determines the naming strategy for data, schema change, transaction, and heartbeat topics.

Type: class
Default: io.debezium.schema.SchemaTopicNamingStrategy
Importance: Medium
Valid values / Notes: A custom implementation must be loadable from the plugin class path. Default data topic names contain the database, schema, and table names.

`table.include.list`

Limits capture to selected business tables.

Type: list
Default: None (null)
Importance: High
Valid values / Notes: A comma-separated list of fully qualified name regular expressions. SQL Server table filtering uses schema.table, and periods must be escaped. Do not set it together with table.exclude.list. A match does not automatically enable CDC for the table.

`table.exclude.list`

Excludes tables that do not need to be captured.

Type: list
Default: None (null)
Importance: Medium
Valid values / Notes: A comma-separated list of schema.table regular expressions. Mutually exclusive with table.include.list.

`table.ignore.builtin`

Controls whether built-in tables are ignored.

Type: boolean
Default: true
Importance: Low
Valid values / Notes: true / false. This does not change external system restrictions on CDC base tables.

`column.include.list`

Selects the columns retained in event values.

Type: list
Default: None (null)
Importance: Medium
Valid values / Notes: A comma-separated list of schema.table.column regular expressions. Mutually exclusive with column.exclude.list. Message keys are formed separately, so column filtering does not filter fields in the key.

`column.exclude.list`

Excludes columns from event values.

Type: list
Default: None (null)
Importance: Medium
Valid values / Notes: A comma-separated list of fully qualified column name regular expressions. Mutually exclusive with column.include.list; this setting alone cannot guarantee that sensitive fields do not appear in the message key.

`skip.messages.without.change`

Skips message publication when none of the included columns changed.

Type: boolean
Default: false
Importance: Medium
Valid values / Notes: true / false. Interpret its behavior together with column filtering; it is not an arbitrary row predicate.

`skipped.operations`

Selects operation types to skip during streaming.

Type: list
Default: t
Importance: Low
Valid values / Notes: c, u, d, t, or none, meaning create, update, delete, truncate, or skip none, respectively. The presence of the t option does not mean that SQL Server CDC can produce TRUNCATE events. Skipping business operations leaves downstream state incomplete.

Initial Snapshot

`snapshot.mode`

Determines how existing data is baselined at startup and how incremental reading begins.

Type: string
Default: initial
Importance: Low
Valid values / Notes: initial takes a snapshot when no Offset exists and then begins streaming. initial_only only takes a snapshot. always takes a snapshot on every start. no_data captures only the schema before streaming. schema_only is a deprecated alias for no_data. when_needed decides whether to take a snapshot based on conditions such as a missing or unavailable position, but does not guarantee automatic recovery from every CDC cleanup scenario. recovery rebuilds schema history and should be considered only under safe conditions, such as no relevant schema changes while the Connector was stopped; it cannot recover historical events that have been cleaned up. configuration_based is controlled by the switches below. custom requires a custom Snapshotter.

`snapshot.isolation.mode`

Determines the snapshot isolation level and its concurrency effects.

Type: string
Default: repeatable_read
Importance: Low
Valid values / Notes: repeatable_read, exclusive, snapshot, read_committed, or read_uncommitted. Weaker isolation does not guarantee a consistent snapshot. snapshot requires ALLOW_SNAPSHOT_ISOLATION to be enabled on the database. Locking mode and lock timeout affect locking behavior. When database.applicationIntent is exactly ReadOnly, the implementation uses snapshot isolation.

`snapshot.lock.timeout.ms`

The maximum time to wait for table locks when a snapshot starts, in milliseconds.

Type: long
Default: 10000
Importance: Medium
Valid values / Notes: A lock acquisition timeout terminates that snapshot attempt. Check concurrent source database load before adjusting it.

`snapshot.locking.mode`

Controls locks during the schema-reading phase for snapshot isolation modes that require locking.

Type: string
Default: exclusive (the default defined by the public runtime parameter)
Importance: Low
Valid values / Notes: exclusive, none, or custom. Applies to the repeatable_read and exclusive isolation modes. Consider none only under safe conditions, such as no schema changes during the snapshot; it does not guarantee that every snapshot phase is lock-free. custom requires a corresponding lock implementation.

`snapshot.locking.mode.custom.name`

Selects a custom snapshot lock implementation.

Type: string
Default: None (null)
Importance: Medium
Valid values / Notes: Required when snapshot.locking.mode=custom. The implementation must be loadable and expose the matching name.

`snapshot.delay.ms`

The delay before a snapshot starts after startup, in milliseconds.

Type: long
Default: 0
Importance: Low
Valid values / Notes: A non-negative integer. This is not the streaming read interval.

`snapshot.fetch.size`

The number of records read into memory at a time during a snapshot.

Type: int
Default: None (null)
Importance: Medium
Valid values / Notes: A non-negative integer. When unset, implementation and JDBC fetch behavior applies; do not treat a batch size used in an example as the default.

`snapshot.max.threads`

The maximum number of threads used to execute a snapshot.

Type: int
Default: 1
Importance: Medium
Valid values / Notes: A positive integer. The snapshot thread count is distinct from the number of Connect Tasks. Increasing it increases database load.

`snapshot.include.collection.list`

Limits the data collections read by the snapshot.

Type: list
Default: None (null)
Importance: Medium
Valid values / Notes: A list of fully qualified collection name regular expressions within the existing capture scope. It does not replace table filtering during streaming.

`snapshot.select.statement.overrides`

Declares tables whose snapshot SELECT statements are overridden.

Type: string
Default: None (null)
Importance: Medium
Valid values / Notes: A comma-separated list of table identifiers. SQL Server uses <database>.<schema>.<table>; also provide the corresponding dynamic SQL properties. A WHERE clause affects only the snapshot and is not a continuous row-level filter.

`snapshot.select.statement.overrides.<table-id>`

Specifies the snapshot SELECT statement for a declared table.

Type: string
Default: None
Importance: Not declared
Valid values / Notes: Replace <table-id> with <database>.<schema>.<table> and list it in snapshot.select.statement.overrides. The returned columns must satisfy the table schema and event conversion requirements.

`snapshot.tables.order.by.row.count`

Orders initial snapshot tables by row count.

Type: string
Default: disabled
Importance: Medium
Valid values / Notes: disabled, ascending, or descending. This does not change the synchronization scope.

`snapshot.query.mode`

Selects how snapshot queries are generated.

Type: string
Default: select_all
Importance: Low
Valid values / Notes: select_all or custom. The custom mode requires a query provider.

`snapshot.query.mode.custom.name`

Selects a custom snapshot query implementation.

Type: string
Default: None (null)
Importance: Medium
Valid values / Notes: When snapshot.query.mode=custom, supply the matching implementation name. The class and its dependencies must be loadable.

`snapshot.mode.custom.name`

Selects a custom Snapshotter.

Type: string
Default: None (null)
Importance: Medium
Valid values / Notes: Required when snapshot.mode=custom and must match the implementation’s name().

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

Controls whether the configuration-based snapshot mode reads table data.

Type: boolean
Default: false
Importance: Medium
Valid values / Notes: Applies only when snapshot.mode=configuration_based.

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

Controls whether the configuration-based snapshot mode reads table schemas.

Type: boolean
Default: false
Importance: Medium
Valid values / Notes: Applies only when snapshot.mode=configuration_based. Coordinate it with data reading and history recovery requirements.

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

Controls whether the configuration-based snapshot mode begins streaming after the snapshot.

Type: boolean
Default: false
Importance: Medium
Valid values / Notes: Applies only when snapshot.mode=configuration_based.

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

Controls whether the configuration-based snapshot mode takes a new snapshot after a data position error.

Type: boolean
Default: false
Importance: Medium
Valid values / Notes: Applies only when snapshot.mode=configuration_based. Taking a new snapshot cannot restore the complete history of changes that have already been cleaned up.

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

Controls whether the configuration-based snapshot mode takes a new snapshot after a schema history error.

Type: boolean
Default: false
Importance: Medium
Valid values / Notes: Applies only when snapshot.mode=configuration_based. Verify that schema changes and retained positions are compatible.

Incremental Snapshots and Signals

`incremental.snapshot.chunk.size`

The number of rows in each incremental snapshot chunk.

Type: int
Default: 1024
Importance: Medium
Valid values / Notes: The definition accepts non-negative integers. Actual chunking requires a size that allows progress; evaluate it together with row size and database load.

`incremental.snapshot.allow.schema.changes`

Controls handling of schema changes during an incremental snapshot.

Type: boolean
Default: false
Importance: Low
Valid values / Notes: This switch does not replace CDC capture instance maintenance. Do not infer that arbitrary concurrent SQL Server DDL is safe, especially primary key changes during an incremental snapshot.

`incremental.snapshot.option.recompile`

Adds OPTION(RECOMPILE) to incremental snapshot SELECT statements.

Type: boolean
Default: false
Importance: Low
Valid values / Notes: This can avoid parameter-sniffing effects, but increases compilation and CPU load on the source database.

`incremental.snapshot.watermarking.strategy`

Controls how watermark signals for incremental snapshot windows are written.

Type: string
Default: INSERT_INSERT (corresponding configuration value: insert_insert)
Importance: Low
Valid values / Notes: insert_insert writes both opening and closing signals. insert_delete writes an opening signal and deletes the corresponding signal when closing. A signal data collection and corresponding write permissions are required.

`signal.data.collection`

The fully qualified name of the source signal table.

Type: string
Default: None (null)
Importance: Medium
Valid values / Notes: SQL Server uses <database>.<schema>.<table>. The source signal flow requires the prescribed schema, CDC, and corresponding access permissions. This source signal entry point is not enabled when the setting is absent.

`signal.enabled.channels`

The enabled signal channels.

Type: list
Default: source
Importance: Medium
Valid values / Notes: Select from available channels such as source, kafka, file, and jmx. Each selected channel requires its supporting resources and parameters.

`signal.poll.interval.ms`

The interval for checking registered signal channels, in milliseconds.

Type: long
Default: 5000
Importance: Medium
Valid values / Notes: A positive integer. This is distinct from the business CDC polling interval.

`signal.file`

The file read by the file signal channel.

Type: string
Default: None (null)
Importance: High
Valid values / Notes: Required when the file channel is enabled. The Worker running the Task must be able to access it.

`signal.kafka.bootstrap.servers`

The cluster addresses for the Kafka signal channel.

Type: string
Default: None (null)
Importance: High
Valid values / Notes: Required when the kafka channel is enabled. It must point to the Kafka cluster used by Connect.

`signal.kafka.topic`

The topic read by the Kafka signal channel.

Type: string
Default: None (null)
Importance: High
Valid values / Notes: Required when the kafka channel is enabled. Prepare consume permissions.

`signal.kafka.groupId`

The group identifier for the Kafka signal consumer.

Type: string
Default: kafka-signal
Importance: Low
Valid values / Notes: Preserve the uppercase I in the property name. Coordinate group isolation with the signal delivery method.

`signal.kafka.poll.timeout.ms`

The timeout for a single Kafka signal poll, in milliseconds.

Type: int
Default: 0
Importance: Low
Valid values / Notes: A non-negative integer. Applies only to the Kafka signal channel.

`signal.consumer.*`

Passes consumer properties through to the Kafka signal channel.

Type: string
Default: No fixed default
Importance: Not declared
Valid values / Notes: The property is passed to the Kafka client after removing the prefix and can be used for signal consumer authentication. It is not equivalent to all Worker consumer settings.

Read Batches and Memory

`max.batch.size`

The maximum batch of records returned to Connect at a time.

Type: int
Default: 2048
Importance: Medium
Valid values / Notes: A positive integer smaller than max.queue.size.

`max.queue.size`

The capacity of the queue holding change events waiting to be sent.

Type: int
Default: 8192
Importance: Medium
Valid values / Notes: Must be greater than max.batch.size. A full queue creates backpressure; it is not an unlimited buffer.

`max.queue.size.in.bytes`

The byte capacity limit for the queue holding events waiting to be sent.

Type: long
Default: 0
Importance: Medium
Valid values / Notes: A non-negative integer. 0 disables this byte limit, while the event count limit still applies.

`max.iteration.transactions`

Controls the number of transactions included in each CDC query iteration and constrains memory requirements for multi-table streaming.

Type: int
Default: 500
Importance: Medium
Valid values / Notes: A non-negative integer. It is not a row count or Kafka message batch size; evaluate it together with transaction sizes.

`query.fetch.size`

The number of records read into memory in each JDBC query batch.

Type: int
Default: 10000 (the default defined by the public SQL Server runtime parameter)
Importance: Medium
Valid values / Notes: 0 delegates to the JDBC default fetch behavior. This is distinct from the per-table read limit set by streaming.fetch.size.

`streaming.fetch.size`

The maximum number of rows read from each table at a time during streaming.

Type: int
Default: 0
Importance: Low
Valid values / Notes: 0 means unlimited. A nonzero value affects per-table batching and must not be treated as a transaction boundary.

`poll.interval.ms`

The time to wait before the next poll when no new change events are available, in milliseconds.

Type: long
Default: 500
Importance: Medium
Valid values / Notes: A positive integer. More frequent polling increases query overhead.

`streaming.delay.ms`

The delay before entering the streaming phase after a snapshot completes, in milliseconds.

Type: long
Default: 0
Importance: Low
Valid values / Notes: A non-negative integer. Required CDC data must remain retained during the delay.

`executor.shutdown.timeout.ms`

The maximum time to wait for the Task executor to shut down, in milliseconds.

Type: long
Default: 4000
Importance: Medium
Valid values / Notes: A positive integer. This does not replace Offset and history persistence.

Message Keys, Types, and Fields

`message.key.columns`

Specifies message key columns for matching tables.

Type: string
Default: None (null)
Importance: Medium
Valid values / Notes: A semicolon-separated list in the form <table-regex>:<column-1>,<column-2>. SQL Server table matching uses schema.table. Tables not overridden use their primary keys. A custom key does not create a unique constraint in the database; you must ensure that it is stable and unique.

`tombstones.on.delete`

Controls whether a same-key, null-value tombstone is sent after a DELETE event.

Type: boolean
Default: true
Importance: Medium
Valid values / Notes: When disabled, the delete event is still sent but no subsequent tombstone is produced. Consumers must distinguish between the two.

`decimal.handling.mode`

Controls the event representation of DECIMAL and NUMERIC values.

Type: string
Default: precise
Importance: Medium
Valid values / Notes: precise uses the precise Connect Decimal representation, string uses strings, and double can lose precision. money and smallmoney also use the decimal conversion path.

`binary.handling.mode`

Controls the event representation of binary columns.

Type: string
Default: bytes
Importance: Low
Valid values / Notes: bytes, base64, base64-url-safe, or hex. Keep it consistent with the downstream deserialization contract.

`time.precision.mode`

Controls the precision used to represent dates, times, and timestamps.

Type: string
Default: adaptive
Importance: Medium
Valid values / Notes: adaptive, adaptive_time_microseconds, isostring, microseconds, nanoseconds, or connect. connect uses Connect millisecond precision. Other modes must be interpreted together with the actual column type and conversion path; do not assume that every SQL Server type produces the same representation.

`schema.name.adjustment.mode`

Adjusts message Schema names for Converter compatibility.

Type: string
Default: none
Importance: Low
Valid values / Notes: none, avro, or avro_unicode. The latter two replace invalid characters or apply Unicode escaping, respectively.

`field.name.adjustment.mode`

Adjusts message field names for Converter compatibility.

Type: string
Default: none
Importance: Low
Valid values / Notes: none, avro, or avro_unicode. Name adjustment can affect downstream field mappings.

`column.propagate.source.type`

Adds source type and length information to the field Schema for matching columns.

Type: list
Default: None (null)
Importance: Medium
Valid values / Notes: A list of fully qualified column name regular expressions. This does not change the column value representation.

`datatype.propagate.source.type`

Selects fields for which original type information is propagated by matching source data type names.

Type: list
Default: None (null)
Importance: Medium
Valid values / Notes: A list of regular expressions for native database type names. This differs from matching by column name.

`column.truncate.to.<N>.chars`

Truncates matching string columns to the specified number of characters.

Type: The dynamic property value is a string (a list of column regular expressions)
Default: None (null)
Importance: Medium
Valid values / Notes: N is a non-negative integer in the property name, and the value is a comma-separated list of fully qualified column name regular expressions. The type metadata for the static pattern entry column.truncate.to.(d+).chars is int, but the pattern name must not be used as an actual configuration key; the value of a concrete numbered property is not the character count.

`column.mask.with.<N>.chars`

Replaces matching column content with a mask containing the specified number of asterisks.

Type: string
Default: None (null)
Importance: Medium
Valid values / Notes: N is a non-negative integer, and the value is a list of fully qualified column name regular expressions. The static pattern name column.mask.with.(d+).chars is not used directly as a configuration key. Masking does not replace a security review of message key fields.

`column.mask.hash.<algorithm>.with.salt.<salt>`

Replaces matching column content with a salted hash.

Type: string
Default: None (null)
Importance: Medium
Valid values / Notes: Corresponds to the pattern entry column.mask.hash.([^.]+).with.salt.(.+). Replace the algorithm and salt with actual names and values; the property value is a list of column regular expressions. The algorithm must be supported by the JVM, and the salt must be managed securely.

`column.mask.hash.v2.<algorithm>.with.salt.<salt>`

Uses the v2 salted hash mapping for matching columns.

Type: string
Default: None
Importance: Not declared
Valid values / Notes: The value is a list of fully qualified column name regular expressions, and the algorithm must be supported by the JVM. The resulting downstream identifiers are not unconditionally interchangeable with those produced by the original hash method.

`unavailable.value.placeholder`

Marks original column values that the source did not provide.

Type: string
Default: __debezium_unavailable_value
Importance: Medium
Valid values / Notes: Distinguish it from actual business values and null; downstream systems must not treat the placeholder as real content.

`include.schema.changes`

Controls whether schema messages are published to the schema change topic.

Type: boolean
Default: true
Importance: Medium
Valid values / Notes: The default topic name is topic.prefix. This is independent of internal schema history and does not automatically refresh SQL Server CDC capture instances.

`include.schema.comments`

Controls whether table and column comments are included in schema metadata.

Type: boolean
Default: false
Importance: Medium
Valid values / Notes: Enabling it increases the memory footprint of schema metadata.

`provide.transaction.metadata`

Controls whether transaction metadata and event counts are generated.

Type: boolean
Default: false
Importance: Low
Valid values / Notes: This does not provide global ordering across databases or partitions and is not equivalent to end-to-end exactly-once delivery.

`transaction.metadata.factory`

The factory class that creates transaction context and transaction message structures.

Type: class
Default: io.debezium.pipeline.txmetadata.DefaultTransactionMetadataFactory
Importance: Low
Valid values / Notes: A custom class and its dependencies must be loadable from the plugin class path.

`sourceinfo.struct.maker`

The implementation class that creates the event source structure.

Type: class
Default: io.debezium.connector.sqlserver.SqlServerSourceInfoStructMaker
Importance: Low
Valid values / Notes: A custom implementation can change source position fields used by downstream systems. Keep the consumption contract compatible.

`extended.headers.enabled`

Controls whether Debezium context headers are added to messages.

Type: boolean
Default: true
Importance: Low
Valid values / Notes: The headers carry source context and do not replace business fields in the message value.

Schema History and Recovery

`schema.history.internal`

The implementation that stores and restores database schema history.

Type: class
Default: io.debezium.storage.kafka.history.KafkaSchemaHistory
Importance: Low
Valid values / Notes: The implementation class and its dependencies must be loadable, and storage properties for the implementation must be configured.

`schema.history.internal.kafka.bootstrap.servers`

The cluster addresses used by the Kafka schema history client.

Type: string
Default: None (null)
Importance: High
Valid values / Notes: Required for the default Kafka history implementation. Use a host:port list for the Kafka cluster used by Connect.

`schema.history.internal.kafka.topic`

The topic that stores internal schema history.

Type: string
Default: None (null)
Importance: High
Valid values / Notes: Required for the default Kafka history implementation. Keep it single-partitioned, retain the complete history indefinitely, and do not share it with other independent Connectors. This requirement does not limit the partition count of business data topics.

`schema.history.internal.kafka.create.timeout.ms`

The timeout for creating the history topic, in milliseconds.

Type: long
Default: 30000
Importance: Low
Valid values / Notes: A positive integer. Applies only to the Kafka history implementation.

`schema.history.internal.kafka.query.timeout.ms`

The timeout for querying Kafka cluster information, in milliseconds.

Type: long
Default: 3000
Importance: Low
Valid values / Notes: A positive integer. Applies only to the Kafka history implementation.

`schema.history.internal.kafka.recovery.attempts`

The number of consecutive polls that return no records during history recovery.

Type: int
Default: 100
Importance: Low
Valid values / Notes: Together with the recovery polling interval, this determines how long recovery waits when no records are available. It is not the number of database connection retries.

`schema.history.internal.kafka.recovery.poll.interval.ms`

The polling interval for recovering history data, in milliseconds.

Type: int
Default: 100
Importance: Low
Valid values / Notes: A non-negative integer. Applies only to the Kafka history implementation.

`schema.history.internal.file.filename`

The file path for file-based schema history storage.

Type: string
Default: None (null)
Importance: Medium
Valid values / Notes: Required when using io.debezium.storage.file.history.FileSchemaHistory. The file must persist, and its accessibility after the Task is scheduled to another Worker must be considered.

`schema.history.internal.skip.unparseable.ddl`

Controls whether unparseable DDL is skipped during history processing.

Type: boolean
Default: false
Importance: Low
Valid values / Notes: Skipping can lose schema metadata. This general history option does not mean that every DDL operation in a SQL Server stream can be processed automatically.

`schema.history.internal.store.only.captured.databases.ddl`

Stores schema history only for captured databases.

Type: boolean
Default: false
Importance: Low
Valid values / Notes: Narrowing the history scope can affect recovery preparation when synchronization scope is expanded later.

`schema.history.internal.store.only.captured.tables.ddl`

Stores schema history only for captured tables.

Type: boolean
Default: false
Importance: Low
Valid values / Notes: Verify history completeness before adding new tables later; changing the filter alone does not backfill existing data.

`schema.history.internal.producer.*`

Passes properties through to the Kafka history producer client.

Type: string
Default: No fixed default
Importance: Not declared
Valid values / Notes: The property is passed to the Kafka client after removing the prefix and can provide secure connection settings. It is not equivalent to business message producer configuration.

`schema.history.internal.consumer.*`

Passes properties through to the Kafka history recovery consumer client.

Type: string
Default: No fixed default
Importance: Not declared
Valid values / Notes: The property is passed to the Kafka client after removing the prefix. Coordinate authentication with the history producer; do not assume that all Worker authentication settings are inherited.

Heartbeats and Error Handling

`heartbeat.interval.ms`

The interval for sending heartbeat messages, in milliseconds.

Type: int
Default: 0
Importance: Medium
Valid values / Notes: A non-negative integer. 0 disables heartbeats. Heartbeats alone cannot guarantee that the recovery position remains within the CDC retention window.

`heartbeat.topics.prefix`

The naming prefix for heartbeat topics.

Type: string
Default: __debezium-heartbeat
Importance: Low
Valid values / Notes: Corresponding messages are produced only when heartbeats are enabled. This is not the business table topic prefix.

`heartbeat.action.query`

The database query executed for each heartbeat.

Type: string
Default: None (null)
Importance: Low
Valid values / Notes: The account must have permission to execute the query. Evaluate its write behavior and effect on source database load.

`errors.max.retries`

The retry budget for retriable Connector errors.

Type: int
Default: -1
Importance: Low
Valid values / Notes: -1 means unlimited retries, 0 means no retries, and a positive value is the limit. Not every initialization, conversion, or Kafka write error is handled by this setting. Permission and configuration errors cannot heal through retries.

`retriable.restart.connector.wait.ms`

The wait time before restart after a retriable exception, in milliseconds.

Type: long
Default: 10000
Importance: Low
Valid values / Notes: A positive integer. Evaluate long outage risks together with the CDC retention period.

`event.processing.failure.handling.mode`

Controls how corrupt or unprocessable events are handled.

Type: string
Default: fail
Importance: Medium
Valid values / Notes: fail stops processing. warn logs and skips the event. ignore / skip skips it. Skipping loses the corresponding event and is not equivalent to lossless fault tolerance.

`event.converting.failure.handling.mode`

Controls how column value conversion failures are handled.

Type: string
Default: warn
Importance: Medium
Valid values / Notes: fail throws an exception. warn sets the failed column to null and logs a warning. skip sets the failed column to null and logs at debug level. Do not interpret skip here as meaning that the complete event is necessarily discarded.

Extension Processing and Notifications

`converters`

The alias list for Debezium custom type converters.

Type: string
Default: None (null)
Importance: Low
Valid values / Notes: Provide <converter-alias>.type for each alias. These are distinct from Kafka Connect’s key.converter and value.converter.

`<converter-alias>.type`

Specifies the Debezium custom converter class.

Type: class
Default: None
Importance: Not declared
Valid values / Notes: The alias must be listed in converters, and the implementation and its dependencies must be on the plugin class path.

`<converter-alias>.<option>`

Provides implementation-specific options for a custom converter.

Type: string
Default: No fixed default
Importance: Not declared
Valid values / Notes: The option is passed to the selected implementation after removing the alias prefix. Verify properties against that implementation.

`post.processors`

The alias list for Debezium post-processors.

Type: string
Default: None (null)
Importance: Low
Valid values / Notes: Configure an implementation class for each alias. This is not the Connect SMT list.

`<post-processor-alias>.type`

Specifies the post-processor implementation class.

Type: class
Default: None
Importance: Not declared
Valid values / Notes: The alias must be listed in post.processors, and the implementation class must be loadable.

`<post-processor-alias>.<option>`

Provides implementation-specific options for a post-processor.

Type: string
Default: No fixed default
Importance: Not declared
Valid values / Notes: Applies only to the selected post-processor. Configure it according to the implementation’s interface and dependencies.

`notification.enabled.channels`

The list of enabled notification channels.

Type: list
Default: None (null)
Importance: Medium
Valid values / Notes: Configure available channels. Here, sink refers to a notification write channel, not the Sink Connector direction.

`notification.sink.topic.name`

The topic used by the notification write channel.

Type: string
Default: None (null)
Importance: High
Valid values / Notes: Required when the sink notification channel is enabled. Prepare the corresponding write permission.

`custom.metric.tags`

Adds tags to the Connector’s MBean object name.

Type: list
Default: None (null)
Importance: Low
Valid values / Notes: A comma-separated list of <key>=<value> entries. Avoid credentials and high-cardinality values, and coordinate with existing collection labels.

Data Lineage

`openlineage.integration.enabled`

Controls whether data lineage metadata is published through OpenLineage.

Type: boolean
Default: false
Importance: Low
Valid values / Notes: Enabling it requires a corresponding configuration file and usable OpenLineage service settings.

`openlineage.integration.config.file.path`

The path to the OpenLineage client configuration file.

Type: string
Default: ./openlineage.yml
Importance: Low
Valid values / Notes: When lineage integration is enabled, the Worker running the Task must be able to read the file. A relative path depends on the working directory.

`openlineage.integration.job.description`

The description of the lineage job.

Type: string
Default: Debezium change data capture job
Importance: Low
Valid values / Notes: Applies only when lineage integration is enabled.

`openlineage.integration.job.namespace`

The lineage job namespace.

Type: string
Default: None (null)
Importance: Low
Valid values / Notes: Set it according to the job identity conventions of the lineage system.

`openlineage.integration.job.owners`

The lineage job owner information.

Type: list
Default: None (null)
Importance: Low
Valid values / Notes: A comma-separated list of <key>=<value> entries. Avoid unnecessary personal sensitive information.

`openlineage.integration.job.tags`

The lineage job tags.

Type: list
Default: None (null)
Importance: Low
Valid values / Notes: A comma-separated list of <key>=<value> entries. This does not affect the CDC capture scope.

Connect Framework

`connector.class`

The implementation class that loads this Connector.

Type: string
Default: None
Importance: High
Valid values / Notes: Required. Use io.debezium.connector.sqlserver.SqlServerConnector.

`tasks.max`

The maximum number of Tasks that Connect can create.

Type: int
Default: 1
Importance: High
Valid values / Notes: At least 1. When database.names is used, databases are assigned in a round-robin manner, and the actual Task count is the smaller of the database count and this limit. A single database is not split across multiple Tasks. Increasing this value for one database does not add table-level parallelism. This limit does not guarantee global ordering across databases.

`key.converter`

The Converter that serializes Connect message keys into Kafka message keys.

Type: class
Default: null (not overridden at the Connector level)
Importance: Low
Valid values / Notes: When unset, the Worker configuration is inherited. When set, the Converter class must be loadable and consistent with the downstream deserialization contract.

`value.converter`

The Converter that serializes Connect message values into Kafka message values.

Type: class
Default: null (not overridden at the Connector level)
Importance: Low
Valid values / Notes: When unset, the Worker configuration is inherited. Before retaining or changing event structure, verify downstream handling of Schema, DELETE events, and tombstones.

`errors.tolerance`

Controls error tolerance for Connect conversion and related framework-managed steps.

Type: string
Default: none
Importance: Medium
Valid values / Notes: none stops on failure. all can skip problematic records and risks data loss. This does not replace the Connector’s database error retries or event processing policies.

Best Practices

Define the Business Table Scope During Initial Onboarding

Applicable scenario: You are synchronizing an existing business database for the first time and want to establish a baseline for selected business tables before continuously receiving their changes, without including unrelated CDC tables. Configuration example: Add the following setting to the Quick Start configuration. If table.exclude.list is already present, remove the conflicting setting. Replace the placeholder with fully qualified regular expressions for the tables to synchronize. Escape periods with a backslash, for example, the literal form dbo\.orders.

table.include.list=<schema-table-regex-list>

Key considerations: First, have the DBA enable CDC for the target tables and verify account read permissions, and then create the Connector. The default initial mode already provides an initial snapshot followed by continuous incremental reading, so the default does not need to be set again. A snapshot affects database load and can acquire locks. Choose an appropriate onboarding window and verify that snapshot rows and subsequent inserts, updates, and deletes all reach the corresponding topics. Do not assume that a table without a stable key can be updated idempotently by row. Expanding the filter later is not a simple rerun of initial onboarding: when an Offset already exists, changing the list does not automatically backfill existing rows for newly added tables, so plan that backfill separately. Continue to reuse this configuration for planned maintenance. Do not create another Connector, clear the Offset, or change the prefix. Before stopping, confirm that the Offset has been committed, back up and retain Offset storage and schema history, and ensure that the CDC retention window covers the downtime and catch-up period. After recovery, verify Task status, catch-up progress, and delivery of new changes. Normal at-least-once operation can still replay events that were sent before their positions were committed. Downstream systems must implement idempotency using stable business keys together with source LSNs and event sequence numbers; do not deduplicate solely by a commit_lsn shared within a transaction.

Monitoring

What to Monitor

Monitor Kafka Connect service health, Connector and Task status, throughput and end-to-end latency, Offset commit progress and failures, errors and retries, and Worker JVM heap, GC, and thread state. Continuously check whether processing resumes after a pause and whether backlogs converge after failures. Monitor DLQ activity only when the deployment enables corresponding error handling and a DLQ; do not treat a DLQ as a general fallback for database CDC read failures.

Import the Grafana Dashboard

Download the AutoMQ Connect Cluster Dashboard. Confirm that the Grafana data source can query the collected Connect and Worker metrics and that labels such as cluster, Connector, and Task match the collection configuration. Then import the JSON into Grafana and select the corresponding data source.

Limitations

CDC capture applies to base tables for which CDC has been enabled. Views and arbitrary database objects cannot be treated as equivalent capture targets, and SQL Server CDC cannot be relied on to produce TRUNCATE row events.
A single database is assigned to only one Task, while multiple databases can be assigned by database. Increasing tasks.max cannot parallelize a database by table, and global ordering across databases, Tasks, or Kafka partitions is not guaranteed.
Table include and exclude settings are mutually exclusive, as are column include and exclude settings. A WHERE clause in a snapshot SELECT statement cannot act as a continuous row-level filter during streaming.
Table DDL does not automatically refresh CDC capture instances. A DBA must coordinate creation of a new instance and a safe switchover, and the old instance must not be removed before the Connector finishes reading it. Do not assume that all DDL changes are synchronized transparently.
Offsets and schema history are separate recovery assets. Losing either one means seamless continuation cannot be guaranteed. The single-partition requirement for schema history does not apply to business data topics.
Historical changes removed by CDC cleanup cannot be fully recovered by taking a new snapshot. A new snapshot can only establish current row state, and rerunning an interrupted snapshot can publish rows again.
Standard non-transactional Source operation must be designed for at-least-once delivery. It does not promise end-to-end exactly-once delivery; downstream idempotency and event deduplication remain application responsibilities.

FAQ

What should I do if the connection reports an untrusted certificate or hostname mismatch?

Confirm that the connection hostname matches the SQL Server certificate, that the server provides the complete certificate chain, and that the Worker JVM trusts its CA. Add a private CA to the Worker’s trust configuration and confirm that the Worker actually running the Task uses that configuration. The driver encrypts the connection and validates the certificate by default. Fix the certificate and trust chain instead of hiding the problem by disabling encryption or bypassing validation.

Why can the Connector read existing data but not receive new incremental messages?

Check whether initial_only is selected, whether CDC is enabled for the target database and tables, whether SQL Server Agent and the capture job are running, whether the table passes the filters, and whether the runtime account can read CDC data. The absence of a maximum LSN can also mean that no captured change has occurred yet. Insufficient permission to query Agent status means only that the diagnostic cannot be completed; it does not prove that Agent has stopped. First confirm that new records appear in the source capture table, and then inspect Task status and write errors.

Why does increasing the Task limit not improve throughput for a single database?

The database is the Task assignment boundary, and one database is not split across multiple Tasks. Determine whether the bottleneck is source capture, querying, the event queue, or Kafka writes, and then adjust the relevant batch settings based on load and memory measurements. Do not use more Tasks as a substitute for within-table parallelism, and do not share one history topic among independent Connectors.

What should I do if rows or changes are duplicated after a restart?

Check whether the restart occurred before the initial snapshot completed, whether always is used, and whether the Offset was persisted successfully. Preserve the original identity, database selection, and schema history, and avoid resetting positions in a way that reimports data. Duplicates within a failure window are a condition that at-least-once operation must handle. Apply changes downstream by stable key and identify events using source positions and event sequence numbers, while correctly handling DELETE events, primary key changes, and tombstones.

Why is a newly added column still missing from messages?

Check column filters and the event Converter, and then verify whether the CDC capture instance still uses the old column schema. Adding a column to the base table does not automatically update the old capture table. The DBA must prepare a new instance using a server-supported procedure, wait for the Connector to switch safely, retain the required schema history, and then clean up the old instance. Do not delete the history topic or clear the Offset to address this issue.

What should I do if CDC positions become unavailable after an extended outage?

First verify the retention window, cleanup job, and currently readable positions, and retain backups of the Offset and history. The complete history of changes that has already been cleaned up cannot be reconstructed from the current table. Decide whether downstream systems need current state or a complete audit history, and then plan a new baseline or another backfill method. Do not assume that an automatic snapshot mode covers every invalid-position path, or treat schema history recovery as log recovery.

​Overview

​Prerequisites

​License

​Quick Start

​Configuration

​Connection and Database Scope

​database.hostname

​database.port

​database.instance

​database.names

​database.user

​database.password

​database.query.timeout.ms

​connection.validation.timeout.ms

​data.query.mode

​database.sqlserver.agent.status.query

​database.<JDBC-property-name>

​driver.*

​Topics and Synchronization Scope

​topic.prefix

​topic.naming.strategy

​table.include.list

​table.exclude.list

​table.ignore.builtin

​column.include.list

​column.exclude.list

​skip.messages.without.change

​skipped.operations

​Initial Snapshot

​snapshot.mode

​snapshot.isolation.mode

​snapshot.lock.timeout.ms

​snapshot.locking.mode

​snapshot.locking.mode.custom.name

​snapshot.delay.ms

​snapshot.fetch.size

​snapshot.max.threads

​snapshot.include.collection.list

​snapshot.select.statement.overrides

​snapshot.select.statement.overrides.<table-id>

​snapshot.tables.order.by.row.count

​snapshot.query.mode

​snapshot.query.mode.custom.name

​snapshot.mode.custom.name

​snapshot.mode.configuration.based.snapshot.data

​snapshot.mode.configuration.based.snapshot.schema

​snapshot.mode.configuration.based.start.stream

​snapshot.mode.configuration.based.snapshot.on.data.error

​snapshot.mode.configuration.based.snapshot.on.schema.error

​Incremental Snapshots and Signals

​incremental.snapshot.chunk.size

​incremental.snapshot.allow.schema.changes

​incremental.snapshot.option.recompile

​incremental.snapshot.watermarking.strategy

​signal.data.collection

​signal.enabled.channels

​signal.poll.interval.ms

​signal.file

​signal.kafka.bootstrap.servers

​signal.kafka.topic

​signal.kafka.groupId

​signal.kafka.poll.timeout.ms

​signal.consumer.*

​Read Batches and Memory

​max.batch.size

​max.queue.size

​max.queue.size.in.bytes

​max.iteration.transactions

​query.fetch.size

​streaming.fetch.size

​poll.interval.ms

​streaming.delay.ms

​executor.shutdown.timeout.ms

​Message Keys, Types, and Fields

​message.key.columns

​tombstones.on.delete

​decimal.handling.mode

​binary.handling.mode

​time.precision.mode

​schema.name.adjustment.mode