> ## Documentation Index
> Fetch the complete documentation index at: https://docs.automq.com/llms.txt
> Use this file to discover all available pages before exploring further.

# YugabyteDB Source Connector

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

## 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](../manage-connectors).

```properties theme={null}
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:

```properties theme={null}
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:

```properties theme={null}
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](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), 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.