YugabyteDB Source Connector - AutoMQ Documentation

Overview

The YugabyteDB Source Connector reads changes from YugabyteDB tables through YSQL logical replication and writes INSERT, UPDATE, DELETE, and TRUNCATE events to Kafka topics. Changes from regular tables are written to topic names composed of topic.prefix, the schema, and the table name. Message values use the Debezium event structure, which includes fields such as before, after, source, op, and timestamps. The connector can first create a snapshot of existing data and then continuously read incremental changes, or it can begin streaming directly when a data baseline already exists. It is suitable for synchronizing business data in YugabyteDB to stream processing, search, analytics, or heterogeneous data systems. Kafka record keys usually come from table primary keys; when no usable primary key exists, the record key may be null.

Prerequisites

Use a dedicated YugabyteDB YSQL account with the LOGIN and REPLICATION attributes. Depending on the deployment, pre-create or allow the connector to create the matching replication slot and publication. If the connector is allowed to create the publication, also grant it CREATE on the database and the required ownership of the captured tables. Confirm that the server supports the logical decoding plugin specified by plugin.name. The target connector artifact declares compatibility with YugabyteDB 2024.1.x.

License

Licensed under the Apache License 2.0.

Quick Start

Prepare a Connect cluster, Kafka, and a YugabyteDB YSQL database in advance, and confirm network connectivity and access permissions. For preparation and management procedures, see Manage Connectors.

connector.class=io.debezium.connector.postgresql.YugabyteDBConnector
topic.prefix=ybdb
database.hostname=<database-host>
database.port=5433
database.user=<database-user>
database.password=<database-password>
database.dbname=<database-name>
plugin.name=yboutput
slot.name=debezium
publication.name=dbz_publication
snapshot.mode=initial

Replace the database address, account, password, and database name placeholders with values for your environment, and confirm that the port, logical decoding plugin, slot, and publication match the target cluster. Provide the database password through a secrets management mechanism. After you apply the configuration, the connector first creates an initial snapshot and then continuously sends change events that occur after that snapshot.

Configuration

Runtime and Serialization

`connector.class`

Specifies the YugabyteDB Source Connector implementation class that Kafka Connect loads.

Type: string
Default: None
Importance: High
Valid values / Notes: This ConfigDef declares no additional public value range.
Required: Yes

`tasks.max`

Maximum number of tasks to use for this connector.

Type: int
Default: 1
Importance: High
Valid values / Notes: Must be an integer greater than or equal to 1.

`tasks.max.enforce`

(Deprecated) Whether to enforce that the tasks.max property is respected by the connector. By default, connectors that generate too many tasks will fail, and existing sets of tasks that exceed the tasks.max property will also be failed. If this property is set to false, then connectors will be allowed to generate more than the maximum number of tasks, and existing sets of tasks that exceed the tasks.max property will be allowed to run. This property is deprecated and will be removed in an upcoming major release.

Type: boolean
Default: true
Importance: Low
Valid values / Notes: Kafka Connect 3.9.1 marks this property as deprecated but still accepts it; this ConfigDef declares no replacement.
Deprecated: Yes

`key.converter`

Overrides the worker-level setting and specifies the converter for record keys.

Type: class
Default: None
Importance: Low
Valid values / Notes: Must be a concrete Converter implementation with a public no-argument constructor.

`value.converter`

Overrides the worker-level setting and specifies the converter for record values.

Type: class
Default: None
Importance: Low
Valid values / Notes: Must be a concrete Converter implementation with a public no-argument constructor.

`header.converter`

Overrides the worker-level setting and specifies the converter for record headers.

Type: class
Default: None
Importance: Low
Valid values / Notes: Must be a concrete HeaderConverter implementation with a public no-argument constructor.

`transforms`

Aliases for the transformations to be applied to records.

Type: list
Default: []
Importance: Low
Valid values / Notes: Use unique, non-empty transformation aliases; configure transforms.<alias>.type for each alias.

`predicates`

Aliases for the predicates used by transformations.

Type: list
Default: []
Importance: Low
Valid values / Notes: Use unique, non-empty predicate aliases; configure predicates.<alias>.type for each alias.

`config.action.reload`

The action that Connect should take on the connector when changes in external configuration providers result in a change in the connector’s configuration properties. A value of ‘none’ indicates that Connect will do nothing. A value of ‘restart’ indicates that Connect should restart/reload the connector with the updated configuration properties.The restart may actually be scheduled in the future if the external configuration provider indicates that a configuration value will expire in the future.

Type: string
Default: restart
Importance: Low
Valid values / Notes: Allowed values: none, restart.

Error Handling and Topic Creation

`errors.retry.timeout`

The maximum duration in milliseconds that a failed operation will be reattempted. The default is 0, which means no retries will be attempted. Use -1 for infinite retries.

Type: long
Default: 0
Importance: Medium
Valid values / Notes: 0 disables retries, -1 retries indefinitely, and a positive value sets the total retry duration.

`errors.retry.delay.max.ms`

The maximum duration in milliseconds between consecutive retry attempts. Jitter will be added to the delay once this limit is reached to prevent thundering herd issues.

Type: long
Default: 60000
Importance: Medium
Valid values / Notes: Must be a non-negative millisecond value; jitter is added after the delay reaches this limit.

`errors.tolerance`

Behavior for tolerating errors during connector operation. ‘none’ is the default value and signals that any error will result in an immediate connector task failure; ‘all’ changes the behavior to skip over problematic records.

Type: string
Default: none
Importance: Medium
Valid values / Notes: Allowed values: none, all.

`errors.log.enable`

If true, write each error and the details of the failed operation and problematic record to the Connect application log. This is ‘false’ by default, so that only errors that are not tolerated are reported.

Type: boolean
Default: false
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`errors.log.include.messages`

Controls whether error logs include the source record content and metadata that caused the failure.

Type: boolean
Default: false
Importance: Medium
Valid values / Notes: Enabling this can write record keys, values, headers, and source offsets to logs; assess exposure of sensitive data.

`topic.creation.groups`

Groups of configurations for topics created by source connectors

Type: list
Default: []
Importance: Low
Valid values / Notes: Use unique, non-empty topic creation group aliases.

Source Transactions and Offsets

`exactly.once.support`

Permitted values are requested, required. If set to “required”, forces a preflight check for the connector to ensure that it can provide exactly-once semantics with the given configuration. Some connectors may be capable of providing exactly-once semantics but not signal to Connect that they support this; in that case, documentation for the connector should be consulted carefully before creating it, and the value for this property should be set to “requested”. Additionally, if the value is set to “required” but the worker that performs preflight validation does not have exactly-once support enabled for source connectors, requests to create or validate the connector will fail.

Type: string
Default: requested
Importance: Medium
Valid values / Notes: Allowed values are case-insensitive: required, requested.

`transaction.boundary`

Permitted values are: poll, interval, connector. If set to ‘poll’, a new producer transaction will be started and committed for every batch of records that each task from this connector provides to Connect. If set to ‘connector’, relies on connector-defined transaction boundaries; note that not all connectors are capable of defining their own transaction boundaries, and in that case, attempts to instantiate a connector with this value will fail. Finally, if set to ‘interval’, commits transactions only after a user-defined time interval has passed.

Type: string
Default: poll
Importance: Medium
Valid values / Notes: Allowed values are case-insensitive: interval, poll, connector.

`transaction.boundary.interval.ms`

If ‘transaction.boundary’ is set to ‘interval’, determines the interval for producer transaction commits by connector tasks. If unset, defaults to the value of the worker-level ‘offset.flush.interval.ms’ property. It has no effect if a different transaction.boundary is specified.

Type: long
Default: None
Importance: Low
Valid values / Notes: Must be at least 0; applies only when transaction.boundary=interval, and otherwise defaults to the worker offset.flush.interval.ms value when unset.

`offsets.storage.topic`

The name of a separate offsets topic to use for this connector. If empty or not specified, the worker’s global offsets topic name will be used. If specified, the offsets topic will be created if it does not already exist on the Kafka cluster targeted by this connector (which may be different from the one used for the worker’s global offsets topic if the bootstrap.servers property of the connector’s producer has been overridden from the worker’s). Only applicable in distributed mode; in standalone mode, setting this property will have no effect.

Type: string
Default: None
Importance: Low
Valid values / Notes: Must be a non-empty topic name; applies only in distributed mode, and uses the worker global offsets topic when unset.

Topics and Database Connection

`topic.prefix`

Topic prefix that identifies and provides a namespace for the particular database server/cluster is capturing changes. The topic prefix should be unique across all other connectors, since it is used as a prefix for all Kafka topic names that receive events emitted by this connector. Only alphanumeric characters, hyphens, dots and underscores must be accepted.

Type: string
Default: None
Importance: High
Valid values / Notes: This ConfigDef declares no additional public value range.
Required: Yes

`database.hostname`

Resolvable hostname or IP address of the database server.

Type: string
Default: None
Importance: High
Valid values / Notes: This ConfigDef declares no additional public value range.
Required: Yes

`database.port`

Port of the database server.

Type: int
Default: 5433
Importance: High
Valid values / Notes: This ConfigDef declares no additional public value range.

`database.user`

Name of the database user to be used when connecting to the database.

Type: string
Default: None
Importance: High
Valid values / Notes: This ConfigDef declares no additional public value range.
Required: Yes

`database.password`

Password of the database user to be used when connecting to the database.

Type: password
Default: None
Importance: High
Valid values / Notes: Provide it through secrets management and do not write the actual password to ordinary configuration files or logs.

`database.dbname`

The name of the database from which the connector should capture changes

Type: string
Default: None
Importance: High
Valid values / Notes: This ConfigDef declares no additional public value range.
Required: Yes

`database.initial.statements`

A semicolon separated list of SQL statements to be executed when a JDBC connection to the database is established. Note that the connector may establish JDBC connections at its own discretion, so this should typically be used for configuration of session parameters only, but not for executing DML statements. Use doubled semicolon (’;;’) to use a semicolon as a character and not as a delimiter.

Type: string
Default: None
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`database.tcpKeepAlive`

Enable or disable TCP keep-alive probe to avoid dropping TCP connection

Type: boolean
Default: true
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

SSL and Connection Routing

`database.sslmode`

Controls the SSL mode used for the YugabyteDB YSQL connection.

Type: string
Default: prefer
Importance: Medium
Valid values / Notes: Allowed values: allow, prefer, disable, verify-ca, require, verify-full.

`database.sslcert`

File containing the SSL Certificate for the client. See the Postgres SSL docs for further information

Type: string
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`database.sslpassword`

Password to access the client private key from the file specified by ‘database.sslkey’. See the Postgres SSL docs for further information

Type: password
Default: None
Importance: Medium
Valid values / Notes: Provide it through secrets management and do not write the actual password to ordinary configuration files or logs.

`database.sslrootcert`

File containing the root certificate(s) against which the server is validated. See the Postgres JDBC SSL docs for further information

Type: string
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`database.sslkey`

File containing the SSL private key for the client. See the Postgres SSL docs for further information

Type: string
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`database.sslfactory`

A name of class to that creates SSL Sockets. Use com.yugabyte.ssl.NonValidatingFactory to disable SSL validation in development environments

Type: string
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`yb.load.balance.connections`

This config determines load-balance property in the connection url. Supported values are ‘true’, ‘only-primary’, ‘only-rr’, ‘prefer-primary’, ‘prefer-rr’ and ‘false’

Type: string
Default: only-primary
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

Logical Replication and Slots

`plugin.name`

Specifies the logical decoding plugin installed and supported on the server.

Type: string
Default: yboutput
Importance: Medium
Valid values / Notes: Allowed values: decoderbufs, yboutput, pgoutput. The validated recommended path uses yboutput.

`slot.name`

Specifies the logical decoding replication slot used to stream changes.

Type: string
Default: debezium
Importance: Medium
Valid values / Notes: Use lowercase letters, numbers, and underscores, with a maximum length of 63 characters.

`slot.lsn.type`

LSN type being used with the replication slot

Type: string
Default: SEQUENCE
Importance: Medium
Valid values / Notes: Allowed values: sequence, hybrid_time.

`publication.name`

Specifies the publication used to stream changes.

Type: string
Default: dbz_publication
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`publication.autocreate.mode`

Controls whether and how the connector creates a missing publication.

Type: string
Default: all_tables
Importance: Medium
Valid values / Notes: Allowed values: filtered, disabled, all_tables. Pre-create the publication when using disabled.

`replica.identity.autoset.values`

Applies only when streaming changes using pgoutput.Determines the value for Replica Identity at table level. This option will overwrite the existing value in databaseA comma-separated list of regular expressions that match fully-qualified tables and Replica Identity value to be used in the table. Each expression must match the pattern ’ : ’, where the table names could be defined as (SCHEMA_NAME.TABLE_NAME), and the replica identity values are: DEFAULT - Records the old values of the columns of the primary key, if any. This is the default for non-system tables.INDEX index_name - Records the old values of the columns covered by the named index, that must be unique, not partial, not deferrable, and include only columns marked NOT NULL. If this index is dropped, the behavior is the same as NOTHING.FULL - Records the old values of all columns in the row.NOTHING - Records no information about the old row. This is the default for system tables.

Type: string
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`slot.drop.on.stop`

Whether or not to drop the logical replication slot when the connector finishes orderly. By default the replication is kept so that on restart progress can resume from the last recorded location

Type: boolean
Default: false
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`slot.stream.params`

Any optional parameters used by logical decoding plugin. Semi-colon separated. E.g. ‘add-tables=public.table,public.table2;include-lsn=true’

Type: string
Default: None
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`slot.max.retries`

How many times to retry connecting to a replication slot when an attempt fails.

Type: int
Default: 6
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`slot.retry.delay.ms`

Time to wait between retry attempts when the connector fails to connect to a replication slot, given in milliseconds. Defaults to 10 seconds (10,000 ms).

Type: long
Default: 10000
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`status.update.interval.ms`

Frequency for sending replication connection status updates to the server, given in milliseconds. Defaults to 10 seconds (10,000 ms).

Type: int
Default: 10000
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`xmin.fetch.interval.ms`

Specify how often (in ms) the xmin will be fetched from the replication slot. This xmin value is exposed by the slot which gives a lower bound of where a new replication slot could start from. The lower the value, the more likely this value is to be the current ‘true’ value, but the bigger the performance cost. The bigger the value, the less likely this value is to be the current ‘true’ value, but the lower the performance penalty. The default is set to 0 ms, which disables tracking xmin.

Type: long
Default: 0
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`flush.lsn.source`

Controls whether Debezium flushes the processed LSN to the source YugabyteDB database.

Type: boolean
Default: true
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`streaming.mode`

Streaming mode the connector should follow

Type: string
Default: DEFAULT
Importance: Low
Valid values / Notes: Allowed values: default, parallel. parallel also requires matching numbers of slot.names, publication.names, and slot.ranges entries.

`slot.names`

Comma separated values for multiple slot names

Type: string
Default: None
Importance: Low
Valid values / Notes: Used only for parallel streaming. Separate entries with commas; the count must match publication.names and slot.ranges.

`publication.names`

Comma separated values for multiple publication names

Type: string
Default: None
Importance: Low
Valid values / Notes: Used only for parallel streaming. Separate entries with commas; the count must match slot.names and slot.ranges.

`slot.ranges`

Semi-colon separated values for hash ranges to be polled by tasks.

Type: string
Default: None
Importance: Low
Valid values / Notes: Separate hash ranges with semicolons; the count must match the slots and publications, and the ranges must fully cover 0 through 65536.

`ysql.major.upgrade`

Should be used carefully only on YSQL major upgrade. This will set yb_ignore_read_time_in_walsender to true in the walsender session. Setting this is required to get over the catalog read errors.

Type: boolean
Default: false
Importance: High
Valid values / Notes: This ConfigDef declares no additional public value range.

Capture Scope and Message Selection

`skipped.operations`

The comma-separated list of operations to skip during streaming, defined as: ‘c’ for inserts/create; ‘u’ for updates; ‘d’ for deletes, ‘t’ for truncates, and ‘none’ to indicate nothing skipped. By default, only truncate operations will be skipped.

Type: list
Default: ["t"]
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`message.prefix.include.list`

A comma-separated list of regular expressions that match the logical decoding message prefixes to be monitored. All prefixes are monitored by default.

Type: list
Default: None
Importance: Medium
Valid values / Notes: Mutually exclusive with message.prefix.exclude.list.

`message.prefix.exclude.list`

A comma-separated list of regular expressions that match the logical decoding message prefixes to be excluded from monitoring.

Type: list
Default: None
Importance: Medium
Valid values / Notes: Mutually exclusive with message.prefix.include.list.

`table.include.list`

The tables for which changes are to be captured

Type: list
Default: None
Importance: High
Valid values / Notes: Use regular expressions for fully qualified table names; mutually exclusive with table.exclude.list.

`table.exclude.list`

A comma-separated list of regular expressions that match the fully-qualified names of tables to be excluded from monitoring

Type: list
Default: None
Importance: Medium
Valid values / Notes: Use regular expressions for fully qualified table names; mutually exclusive with table.include.list.

`table.ignore.builtin`

Flag specifying whether built-in tables should be ignored.

Type: boolean
Default: true
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`schema.include.list`

The schemas for which events should be captured

Type: list
Default: None
Importance: High
Valid values / Notes: Mutually exclusive with schema.exclude.list.

`schema.exclude.list`

The schemas for which events must not be captured

Type: list
Default: None
Importance: Medium
Valid values / Notes: Mutually exclusive with schema.include.list.

`column.include.list`

Regular expressions matching columns to include in change events

Type: list
Default: None
Importance: Medium
Valid values / Notes: Use regular expressions for fully qualified column names; mutually exclusive with column.exclude.list.

`column.exclude.list`

Regular expressions matching columns to exclude from change events

Type: list
Default: None
Importance: Medium
Valid values / Notes: Use regular expressions for fully qualified column names; mutually exclusive with column.include.list.

`message.key.columns`

A semicolon-separated list of expressions that match fully-qualified tables and column(s) to be used as message key. Each expression must match the pattern ’ : ’, where the table names could be defined as (DB_NAME.TABLE_NAME) or (SCHEMA_NAME.TABLE_NAME), depending on the specific connector, and the key columns are a comma-separated list of columns representing the custom key. For any table without an explicit key configuration the table’s primary key column(s) will be used as message key. Example: dbserver1.inventory.orderlines:orderId,orderLineId;dbserver1.inventory.orders:id

Type: string
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

Snapshots and Incremental Snapshots

`snapshot.mode`

Controls whether and how the connector snapshots existing data at startup.

Type: string
Default: initial
Importance: Medium
Valid values / Notes: Allowed values: always, never, initial_only, initial, parallel, custom.

`snapshot.custom.class`

When ‘snapshot.mode’ is set as custom, this setting must be set to specify a fully qualified class name to load (via the default class loader). This class must implement the ‘Snapshotter’ interface and is called on each app boot to determine whether to do a snapshot and how to build queries.

Type: string
Default: None
Importance: Medium
Valid values / Notes: Used only when snapshot.mode=custom.

`snapshot.delay.ms`

A delay period before a snapshot will begin, given in milliseconds. Defaults to 0 ms.

Type: long
Default: 0
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`snapshot.include.collection.list`

This setting must be set to specify a list of tables/collections whose snapshot must be taken on creating or restarting the connector.

Type: list
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`snapshot.fetch.size`

The maximum number of records that should be loaded into memory while performing a snapshot.

Type: int
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`snapshot.max.threads`

The maximum number of threads used to perform the snapshot. Defaults to 1.

Type: int
Default: 1
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`snapshot.lock.timeout.ms`

The maximum number of millis to wait for table locks at the beginning of a snapshot. If locks cannot be acquired in this time frame, the snapshot will be aborted. Defaults to 10 seconds

Type: long
Default: 10000
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`yb.consistent.snapshot`

Whether or not to take a consistent snapshot of the tables.Disabling this option may result in duplication of some already snapshot data in the streaming phase.

Type: boolean
Default: true
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`incremental.snapshot.chunk.size`

The maximum size of chunk (number of documents/rows) for incremental snapshotting

Type: int
Default: 1024
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`incremental.snapshot.watermarking.strategy`

Specify the strategy used for watermarking during an incremental snapshot: ‘insert_insert’ both open and close signal is written into signal data collection (default); ‘insert_delete’ only open signal is written on signal data collection, the close will delete the relative open signal;

Type: string
Default: INSERT_INSERT
Importance: Low
Valid values / Notes: Allowed values: insert_delete, insert_insert.

`snapshot.select.statement.overrides`

This property contains a comma-separated list of fully-qualified tables (DB_NAME.TABLE_NAME) or (SCHEMA_NAME.TABLE_NAME), depending on the specific connectors. Select statements for the individual tables are specified in further configuration properties, one for each table, identified by the id ‘snapshot.select.statement.overrides.[DB_NAME].[TABLE_NAME]’ or ‘snapshot.select.statement.overrides.[SCHEMA_NAME].[TABLE_NAME]’, respectively. The value of those properties is the select statement to use when retrieving data from the specific table during snapshotting. A possible use case for large append-only tables is setting a specific point where to start (resume) snapshotting, in case a previous snapshotting was interrupted.

Type: string
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

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

Controls the order in which tables are processed in the initial snapshot. A descending value will order the tables by row count descending. A ascending value will order the tables by row count ascending. A value of disabled (the default) will disable ordering by row count.

Type: string
Default: disabled
Importance: Medium
Valid values / Notes: Allowed values: disabled, ascending, descending.

Queueing, Polling, and Recovery

`event.processing.failure.handling.mode`

Controls how the connector handles failures while processing change events.

Type: string
Default: fail
Importance: Medium
Valid values / Notes: Allowed values: warn, fail, ignore, skip.

`max.batch.size`

Maximum size of each batch of source records. Defaults to 2048.

Type: int
Default: 2048
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`max.queue.size`

Maximum size of the queue for change events read from the database log but not yet recorded or forwarded. Defaults to 8192, and should always be larger than the maximum batch size.

Type: int
Default: 8192
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`poll.interval.ms`

Time to wait for new change events to appear after receiving no events, given in milliseconds. Defaults to 500 ms.

Type: long
Default: 500
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`max.queue.size.in.bytes`

Maximum size of the queue in bytes for change events read from the database log but not yet recorded or forwarded. Defaults to 0. Mean the feature is not enabled

Type: long
Default: 0
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`retriable.restart.connector.wait.ms`

Time to wait before restarting connector after retriable exception occurs. Defaults to 10000ms.

Type: long
Default: 10000
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`query.fetch.size`

The maximum number of records that should be loaded into memory while streaming. A value of ‘0’ uses the default JDBC fetch size.

Type: int
Default: 0
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`errors.max.retries`

The maximum number of retries on connection errors before failing (-1 = no limit, 0 = disabled, > 0 = num of retries).

Type: int
Default: -1
Importance: Low
Valid values / Notes: -1 means unlimited retries, 0 disables retries, and a positive integer sets the maximum retry count.

Event Format and Data Types

`provide.transaction.metadata`

Enables transaction metadata extraction together with event counting

Type: boolean
Default: false
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`decimal.handling.mode`

Specify how DECIMAL and NUMERIC columns should be represented in change events, including: ‘precise’ (the default) uses java.math.BigDecimal to represent values, which are encoded in the change events using a binary representation and Kafka Connect’s ‘org.apache.kafka.connect.data.Decimal’ type; ‘string’ uses string to represent values; ‘double’ represents values using Java’s ‘double’, which may not offer the precision but will be far easier to use in consumers.

Type: string
Default: precise
Importance: Medium
Valid values / Notes: Allowed values: string, double, precise.

`time.precision.mode`

Time, date, and timestamps can be represented with different kinds of precisions, including: ‘adaptive’ (the default) bases the precision of time, date, and timestamp values on the database column’s precision; ‘adaptive_time_microseconds’ like ‘adaptive’ mode, but TIME fields always use microseconds precision; ‘connect’ always represents time, date, and timestamp values using Kafka Connect’s built-in representations for Time, Date, and Timestamp, which uses millisecond precision regardless of the database columns’ precision.

Type: string
Default: adaptive
Importance: Medium
Valid values / Notes: Allowed values: adaptive, adaptive_time_microseconds, connect.

`primary.key.hash.columns`

A comma separated value having all the hash components of the primary key

Type: string
Default: None
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`hstore.handling.mode`

Specify how HSTORE columns should be represented in change events, including: ‘json’ represents values as string-ified JSON (default); ‘map’ represents values as a key/value map

Type: string
Default: json
Importance: Low
Valid values / Notes: Allowed values: json, map.

`binary.handling.mode`

Specify how binary (blob, binary, etc.) columns should be represented in change events, including: ‘bytes’ represents binary data as byte array (default); ‘base64’ represents binary data as base64-encoded string; ‘base64-url-safe’ represents binary data as base64-url-safe-encoded string; ‘hex’ represents binary data as hex-encoded (base16) string

Type: string
Default: bytes
Importance: Low
Valid values / Notes: Allowed values: bytes, base64, hex, base64-url-safe.

`schema.name.adjustment.mode`

Specify how schema names should be adjusted for compatibility with the message converter used by the connector, including: ‘avro’ replaces the characters that cannot be used in the Avro type name with underscore; ‘avro_unicode’ replaces the underscore or characters that cannot be used in the Avro type name with corresponding unicode like _uxxxx. Note: _ is an escape sequence like backslash in Java;‘none’ does not apply any adjustment (default)

Type: string
Default: none
Importance: Low
Valid values / Notes: Allowed values: none, avro_unicode, avro.

`interval.handling.mode`

Specify how INTERVAL columns should be represented in change events, including: ‘string’ represents values as an exact ISO formatted string; ‘numeric’ (default) represents values using the inexact conversion into microseconds

Type: string
Default: numeric
Importance: Low
Valid values / Notes: Allowed values: string, numeric.

`schema.refresh.mode`

Specify the conditions that trigger a refresh of the in-memory schema for a table. ‘columns_diff’ (the default) is the safest mode, ensuring the in-memory schema stays in-sync with the database table’s schema at all times. ‘columns_diff_exclude_unchanged_toast’ instructs the connector to refresh the in-memory schema cache if there is a discrepancy between it and the schema derived from the incoming message, unless unchanged TOASTable data fully accounts for the discrepancy. This setting can improve connector performance significantly if there are frequently-updated tables that have TOASTed data that are rarely part of these updates. However, it is possible for the in-memory schema to become outdated if TOASTable columns are dropped from the table.

Type: string
Default: columns_diff
Importance: Medium
Valid values / Notes: Allowed values: columns_diff, columns_diff_exclude_unchanged_toast.

`unavailable.value.placeholder`

Specify the constant that will be provided by Debezium to indicate that the original value is a toasted value not provided by the database. If starts with ‘hex:’ prefix it is expected that the rest of the string represents hexadecimal encoded octets.

Type: string
Default: __debezium_unavailable_value
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`converters`

Optional list of custom converters that would be used instead of default ones. The converters are defined using ’ .type’ config option and configured using options ’ . ’

Type: string
Default: None
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`post.processors`

Optional list of post processors. The processors are defined using ’ .type’ config option and configured using options ’ ’

Type: string
Default: None
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`tombstones.on.delete`

Whether delete operations should be represented by a delete event and a subsequent tombstone event (true) or only by a delete event (false). Emitting the tombstone event (the default behavior) allows Kafka to completely delete all events pertaining to the given key once the source record got deleted.

Type: boolean
Default: true
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`topic.naming.strategy`

The name of the TopicNamingStrategy class that should be used to determine the topic name for data change, schema change, transaction, heartbeat event etc.

Type: class
Default: io.debezium.schema.SchemaTopicNamingStrategy
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`include.schema.comments`

Whether the connector parse table and column’s comment to metadata object. Note: Enable this option will bring the implications on memory usage. The number and size of ColumnImpl objects is what largely impacts how much memory is consumed by the Debezium connectors, and adding a String to each of them can potentially be quite heavy. The default is ‘false’.

Type: boolean
Default: false
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`include.unknown.datatypes`

Specify whether the fields of data type not supported by Debezium should be processed: ‘false’ (the default) omits the fields; ‘true’ converts the field into an implementation dependent binary representation.

Type: boolean
Default: false
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`sourceinfo.struct.maker`

The name of the SourceInfoStructMaker class that returns SourceInfo schema and struct.

Type: class
Default: io.debezium.connector.postgresql.PostgresSourceInfoStructMaker
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

Signals, Notifications, and Field Processing

`heartbeat.interval.ms`

Length of an interval in milli-seconds in in which the connector periodically sends heartbeat messages to a heartbeat topic. Use 0 to disable heartbeat messages. Disabled by default.

Type: int
Default: 0
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`heartbeat.topics.prefix`

The prefix that is used to name heartbeat topics.Defaults to __debezium-heartbeat.

Type: string
Default: __debezium-heartbeat
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`heartbeat.action.query`

The query executed with every heartbeat.

Type: string
Default: None
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`signal.data.collection`

The name of the data collection that is used to send signals/commands to Debezium. Signaling is disabled when not set.

Type: string
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`signal.poll.interval.ms`

Interval for looking for new signals in registered channels, given in milliseconds. Defaults to 5 seconds.

Type: long
Default: 5000
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`signal.enabled.channels`

List of channels names that are enabled. Source channel is enabled by default

Type: list
Default: ["source"]
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`notification.enabled.channels`

List of notification channels names that are enabled.

Type: list
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`notification.sink.topic.name`

The name of the topic for the notifications. This is required in case ‘sink’ is in the list of enabled channels

Type: string
Default: None
Importance: High
Valid values / Notes: This ConfigDef declares no additional public value range.

`custom.metric.tags`

The custom metric tags will accept key-value pairs to customize the MBean object name which should be appended the end of regular name, each key would represent a tag for the MBean object name, and the corresponding value would be the value of that tag the key is. For example: k1=v1,k2=v2

Type: list
Default: None
Importance: Low
Valid values / Notes: This ConfigDef declares no additional public value range.

`column.mask.hash.([^.]+).with.salt.(.+)`

A comma-separated list of regular expressions matching fully-qualified names of columns that should be masked by hashing the input. Using the specified hash algorithms and salt.

Type: string
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`column.mask.with.(d+).chars`

A comma-separated list of regular expressions matching fully-qualified names of columns that should be masked with configured amount of asterisk (’*’) characters.

Type: string
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`column.truncate.to.(d+).chars`

A comma-separated list of regular expressions matching fully-qualified names of columns that should be truncated to the configured amount of characters.

Type: int
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`column.propagate.source.type`

A comma-separated list of regular expressions matching fully-qualified names of columns that adds the column’s original type and original length as parameters to the corresponding field schemas in the emitted change records.

Type: list
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

`datatype.propagate.source.type`

A comma-separated list of regular expressions matching the database-specific data type names that adds the data type’s original type and original length as parameters to the corresponding field schemas in the emitted change records.

Type: list
Default: None
Importance: Medium
Valid values / Notes: This ConfigDef declares no additional public value range.

Best Practices

Establish a Baseline and Continue Synchronization During Initial Onboarding

Applicable business scenario: You are onboarding existing business tables for the first time and need to write the current data to Kafka before seamlessly receiving subsequent changes. This approach is suitable for the initialization stage when downstream systems do not yet have a complete baseline. Configuration example: Keep the following property in the Quick Start configuration:

snapshot.mode=initial

Key considerations: Before starting the connector, confirm that the slot, publication, logical decoding plugin, and account permissions are ready. Snapshot records use the READ operation type. After the snapshot completes, the connector transitions to streaming. Downstream systems should process records idempotently by primary key and allow record replay during failure recovery or around offset commit boundaries.

Read Only Subsequent Changes When a Data Baseline Already Exists

Applicable business scenario: A downstream system has already completed historical data initialization through another controlled process and needs to receive only new changes from the logical replication position, avoiding another full snapshot from the connector. Configuration example: Change the snapshot mode in the Quick Start configuration to:

snapshot.mode=never

Key considerations: Before using this mode, confirm that the downstream baseline is consistent with the replication starting position and that the replication slot still retains the required history. This configuration does not backfill existing data from before the slot starting position. If the slot is deleted or the required replication history has been removed, recovery from the expected position cannot be guaranteed.

Monitoring

What to Monitor

Monitor Kafka Connect health, connector and task status, throughput, latency, offset commits, errors, retries, and worker JVM signals. Monitor DLQ activity only when the corresponding error handling is enabled.

Import the Grafana Dashboard

Confirm that Connect metrics are available to a Grafana data source and that collected labels meet the dashboard filter requirements. Download the Kafka Connect Dashboard, import the JSON into Grafana, and select the corresponding data source.

Limitations

Parallel streaming requires capturing only one table and creates a task for each slot, publication, and hash range combination. The three lists must contain the same number of entries, and the ranges must completely cover 0 through 65536.
The connector does not guarantee global ordering across topics, Kafka partitions, or multiple parallel tasks.
A failure before Kafka Connect offsets are persisted may cause records to be replayed. Standard operation cannot be treated as providing unconditional end-to-end exactly-once delivery, so downstream systems should provide idempotency or deduplication.
Recovery depends on the replication slot continuing to exist and retaining the required replication history. If the slot is deleted or the history has been removed, the connector cannot guarantee that reading will continue from the original position.
Snapshot modes that do not enter streaming, such as initial_only, provide only an existing-data snapshot and do not provide continuous CDC.

FAQ

Why Can’t the Connector Connect to YugabyteDB?

A mismatched hostname, port, database name, account permissions, or logical decoding plugin can cause startup to fail. Check database.hostname, database.port, database.dbname, and plugin.name. Confirm that the account has the permissions required for login and logical replication and that the worker can access the YSQL service. After correcting the configuration, restart the connector and check the task status.

Why Wasn’t the Expected Data Topic Created?

First, check whether topic.prefix follows the naming rules and whether the connector has actually captured matching data changes. Regular-table topic names also contain the schema and table name. Confirm that the downstream system subscribes to the complete topic name, and check whether the publication includes the target table and whether the replication slot is available.

Why Are There Duplicate Records After a Restart?

If a failure occurs after database changes have been sent to Kafka but before the corresponding offsets are persisted, those changes may be read again during recovery. Check the Connect offset commit status and replication slot position, and have the downstream system perform idempotent writes or deduplication using a stable business key or message key. Do not rely solely on connector restarts to prevent duplicates.

Why Are No Changes Received Continuously After the Initial Snapshot Completes?

Check whether snapshot.mode is set to initial_only, which does not enter continuous streaming after the snapshot completes. If continuous CDC is required, use a mode that includes a streaming phase, and confirm that the slot, publication, plugin, and database permissions remain valid.

​Overview

​Prerequisites

​License

​Quick Start

​Configuration

​Runtime and Serialization

​connector.class

​tasks.max

​tasks.max.enforce

​key.converter

​value.converter

​header.converter

​transforms

​predicates

​config.action.reload

​Error Handling and Topic Creation

​errors.retry.timeout

​errors.retry.delay.max.ms

​errors.tolerance

​errors.log.enable

​errors.log.include.messages

​topic.creation.groups

​Source Transactions and Offsets

​exactly.once.support

​transaction.boundary

​transaction.boundary.interval.ms

​offsets.storage.topic

​Topics and Database Connection

​topic.prefix

​database.hostname

​database.port

​database.user

​database.password

​database.dbname

​database.initial.statements

​database.tcpKeepAlive

​SSL and Connection Routing

​database.sslmode

​database.sslcert

​database.sslpassword

​database.sslrootcert

​database.sslkey

​database.sslfactory

​yb.load.balance.connections

​Logical Replication and Slots

​plugin.name

​slot.name

​slot.lsn.type

​publication.name

​publication.autocreate.mode

​replica.identity.autoset.values

​slot.drop.on.stop

​slot.stream.params

​slot.max.retries

​slot.retry.delay.ms

​status.update.interval.ms

​xmin.fetch.interval.ms

​flush.lsn.source

​streaming.mode

​slot.names

​publication.names

​slot.ranges

​ysql.major.upgrade

​Capture Scope and Message Selection

​skipped.operations

​message.prefix.include.list

​message.prefix.exclude.list

​table.include.list

​table.exclude.list

​table.ignore.builtin

​schema.include.list

​schema.exclude.list

​column.include.list

​column.exclude.list

​message.key.columns

​Snapshots and Incremental Snapshots

​snapshot.mode

​snapshot.custom.class

​snapshot.delay.ms

​snapshot.include.collection.list