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

# Debezium PostgreSQL Source Connector

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

## Overview

Debezium PostgreSQL Source Connector sends existing data and committed row-level changes from PostgreSQL tables to Kafka. It supports incremental data synchronization, search index updates, cache refreshes, and change event consumption. The Connector handles data capture in the PostgreSQL-to-Kafka pipeline, establishing a baseline through a snapshot and then continuously capturing inserts, updates, and deletes through logical replication.

One Connector captures one database. By default, events from each table are sent to a topic whose name combines the topic prefix, schema name, and table name; message keys typically come from the table's primary key. Events include the operation type, before and after data, and source position. The actual serialization format is determined by Kafka Connect converters. Snapshot records and subsequent changes together allow downstream consumers to reconstruct the current state, but consumers must still handle duplicates, deletes, and unavailable field values correctly.

## Prerequisites

* PostgreSQL must have logical replication enabled with `wal_level=logical` and sufficient capacity in `max_replication_slots` and `max_wal_senders`. Server authentication rules must allow the capture account to establish database and logical replication connections. `pgoutput` uses PostgreSQL's built-in logical decoding plugin; `decoderbufs` requires the corresponding plugin to be installed on the server.
* The capture account needs `LOGIN` and `REPLICATION` privileges, or the equivalent replication role provided by a managed PostgreSQL service. Snapshots and metadata reads also require `CONNECT` on the target database, `USAGE` on the target schemas, and `SELECT` on captured tables. Replication privileges do not automatically grant table read access.
* With `pgoutput`, the publication's table set and published operations must cover the required changes. The Quick Start uses a table-specific publication created in advance by a DBA; the Connector does not create or modify it. With `filtered` automatic management, the account also needs the appropriate database `CREATE` privilege, publication ownership, and table ownership. Creating a `FOR ALL TABLES` publication requires superuser privileges; do not grant them to the capture account merely to start the Connector.
* Tables whose updates and deletes must be synchronized need an appropriate `REPLICA IDENTITY`. Tables with a primary key can usually use `DEFAULT`; evaluate `FULL` and its WAL overhead only when complete old rows are needed. Configuring message keys does not automatically change the database replica identity. Without a usable replica identity, PostgreSQL may reject updates or deletes on published tables.

## License

Licensed under Apache License 2.0.

## Quick Start

Prepare a Connect Cluster, Kafka, the source PostgreSQL database, and the tables to capture. Confirm network connectivity and access permissions, and have a DBA prepare a publication for these tables. For Connector creation and management, see [Manage Connectors](../manage-connectors).

```properties theme={null}
connector.class=io.debezium.connector.postgresql.PostgresConnector
database.hostname=<database-host>
database.user=<replication-user>
database.password=<database-password>
database.dbname=<database-name>
topic.prefix=<topic-prefix>
plugin.name=pgoutput
slot.name=<replication-slot-name>
publication.name=<publication-name>
publication.autocreate.mode=disabled
table.include.list=<schema-table-regex>
```

Replace the placeholders for the database address, account, password, database name, and capture resources. Slot names must use lowercase letters, digits, and underscores, with a maximum of 63 characters. Choose a separate slot name for each independent Connector. Table filter expressions match the complete `schema.table` name. In properties, use `[.]` to match a literal dot, for example `public[.]orders`. The pre-created publication must contain the matching tables; otherwise, snapshots may work while subsequent changes are not captured. The password is only a placeholder. In an actual deployment, use a configured ConfigProvider reference to avoid storing plaintext credentials in configuration copies.

The default snapshot mode is `initial`: when no saved offset exists or the first snapshot is incomplete, the Connector reads existing data and then continuously captures changes. The default database port is `5432`, and the task limit is `1`, so these defaults are omitted from the example. `pgoutput` is an explicit choice, not the default logical decoding plugin. If no Connector-level converters are configured, the Worker's serialization settings apply.

## Configuration

### Connector and Serialization

#### `connector.class`

Specifies the Source Connector implementation.

* **Type**: `STRING`
* **Default**: None
* **Importance**: High
* **Required**: Yes
* **Valid Values / Notes**: Use `io.debezium.connector.postgresql.PostgresConnector`.

#### `tasks.max`

Sets the maximum number of tasks.

* **Type**: `INT`
* **Default**: `1`
* **Importance**: High
* **Valid Values / Notes**: Must be at least `1`. This Connector creates only one task; increasing the limit does not split tables or the replication stream for parallel processing.

#### `key.converter`

Specifies the converter for serializing message keys.

* **Type**: `CLASS`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Uses the Worker converter when unset. The specified class must implement the Kafka Connect `Converter` interface.

#### `value.converter`

Specifies the converter for serializing message values.

* **Type**: `CLASS`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Uses the Worker converter when unset. The specified class must implement the Kafka Connect `Converter` interface. Events are not necessarily JSON text.

### Database Connection and Security

#### `database.hostname`

The source database hostname or IP address.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: High
* **Required**: Yes
* **Valid Values / Notes**: Use a resolvable database address.

#### `database.port`

The source database port.

* **Type**: `INT`
* **Default**: `5432`
* **Importance**: High
* **Valid Values / Notes**: Set to the actual PostgreSQL service port.

#### `database.user`

The database capture account.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: High
* **Required**: Yes
* **Valid Values / Notes**: Must meet the privilege requirements for replication, snapshot reads, and the selected publication management mode.

#### `database.password`

The database account password.

* **Type**: `PASSWORD`
* **Default**: `null`
* **Importance**: High
* **Valid Values / Notes**: Whether a password is required depends on database authentication. Use a secret placeholder or ConfigProvider reference; do not retain real passwords in shared configurations or logs.

#### `database.dbname`

The name of the database to capture.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: High
* **Required**: Yes
* **Valid Values / Notes**: One Connector connects to one database. Table filters use `schema.table`, without a database name prefix.

#### `database.query.timeout.ms`

The database query execution timeout, in milliseconds.

* **Type**: `INT`
* **Default**: `600000`
* **Importance**: Low
* **Valid Values / Notes**: `0` means no query duration limit. This is distinct from replication connection status updates and LSN flush timeouts.

#### `database.initial.statements`

Executes initialization statements when establishing a JDBC connection.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Separate statements with semicolons and escape literal semicolons with double semicolons. Connections may be established multiple times; this setting is generally for session configuration, not business DML.

#### `database.sslmode`

Controls PostgreSQL TLS connections and certificate validation.

* **Type**: `STRING`
* **Default**: `prefer`
* **Importance**: Medium
* **Valid Values / Notes**: `disable`, `allow`, `prefer`, `require`, `verify-ca`, `verify-full`. `prefer` allows fallback to an unencrypted connection. `require` requires encryption but does not imply host identity verification; `verify-full` also validates the certificate hostname.

#### `database.sslcert`

The path to the client TLS certificate file.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Set when the database requires a client certificate. The file must be readable by the process executing the task.

#### `database.sslkey`

The path to the client TLS private key file.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Must meet PostgreSQL JDBC private key format requirements. Restrict file access and do not put private key contents in the configuration.

#### `database.sslpassword`

The password for accessing the client private key.

* **Type**: `PASSWORD`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Used with `database.sslkey`. Supply it only through secure credential management.

#### `database.sslrootcert`

The path to the root certificate file used to validate the server certificate.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Used with certificate validation modes. The process executing the task must be able to read the file.

#### `database.sslfactory`

Specifies the factory class that creates SSL sockets.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Use an installed factory class that meets PostgreSQL JDBC requirements. Do not use a factory that skips certificate validation to bypass identity verification in production.

#### `database.tcpKeepAlive`

Controls TCP keepalive probes.

* **Type**: `BOOLEAN`
* **Default**: `true`
* **Importance**: Medium
* **Valid Values / Notes**: `true` or `false`. Actual probe intervals depend on the operating system and connection environment.

#### `connection.validation.timeout.ms`

The maximum wait for connection validation, in milliseconds.

* **Type**: `LONG`
* **Default**: `60000`
* **Importance**: Low
* **Valid Values / Notes**: Successful connection validation does not mean that all snapshot, publication creation, or update operations are authorized.

#### `driver.<PostgreSQL JDBC property>`

Passes through PostgreSQL JDBC connection properties.

* **Type**: `STRING` property passthrough
* **Default**: No ConfigDef default
* **Importance**: Not declared
* **Valid Values / Notes**: Replace the placeholder in the name with the actual JDBC property name; the `driver.` prefix is removed. Types, defaults, and validation rules are defined by PostgreSQL JDBC. Do not embed plaintext credentials in connection properties.

### Logical Replication and Publications

#### `plugin.name`

Selects the server-side logical decoding plugin.

* **Type**: `STRING`
* **Default**: `decoderbufs`
* **Importance**: Medium
* **Valid Values / Notes**: `decoderbufs` or `pgoutput`. Set explicitly to use the built-in `pgoutput` plugin.

#### `slot.name`

Specifies the logical replication slot used to read changes.

* **Type**: `STRING`
* **Default**: `debezium`
* **Importance**: Medium
* **Valid Values / Notes**: Must match `[a-z0-9_]{1,63}`. Independent Connectors must use separate slots; two active replication connections cannot share one slot.

#### `slot.drop.on.stop`

Controls whether to drop the replication slot on a normal stop.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Retaining the slot supports recovery for long-running capture. Enabling this option may lose WAL needed during downtime. It does not mean that a temporary slot is created.

#### `slot.failover`

Controls whether to create a failover slot.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Creating a failover slot requires a connection to a PostgreSQL 17 or newer primary. Otherwise, a regular slot is created. This option does not automatically upgrade existing slots or configure standby synchronization and failover.

#### `slot.stream.params`

Supplies optional parameters to the logical decoding plugin.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Separate parameters with semicolons. Use only parameters supported by the selected plugin; do not apply parameters from other decoding plugins to `pgoutput`.

#### `slot.max.retries`

The maximum number of retries when connecting to the replication slot.

* **Type**: `INT`
* **Default**: `6`
* **Importance**: Low
* **Valid Values / Notes**: Used with `slot.retry.delay.ms`. Retries cannot resolve insufficient privileges, incorrect slot-sharing designs, or lost WAL.

#### `slot.retry.delay.ms`

The interval between replication slot connection retries, in milliseconds.

* **Type**: `LONG`
* **Default**: `10000`
* **Importance**: Low
* **Valid Values / Notes**: Set an integer number of milliseconds. The runtime reads this value as an integer; avoid values outside the integer range.

#### `publication.name`

Specifies the publication used by `pgoutput`.

* **Type**: `STRING`
* **Default**: `dbz_publication`
* **Importance**: Medium
* **Valid Values / Notes**: Applies only to `pgoutput`. Coordinate the publication scope with the Connector filters.

#### `publication.autocreate.mode`

Controls publication creation and table set management.

* **Type**: `STRING`
* **Default**: `all_tables`
* **Importance**: Medium
* **Valid Values / Notes**: `disabled` requires a pre-created publication. `all_tables` creates an all-table publication if none exists. `filtered` creates or updates the table set of a table-specific publication. `no_tables` creates an empty publication if none exists. `filtered` rejects an existing `FOR ALL TABLES` publication and an empty matching table set. Updates may affect other users and require database-side privileges.

#### `replica.identity.autoset.values`

Automatically changes replica identities according to per-table rules.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: A comma-separated mapping of table regular expressions to replica identities, in the form `schema.table:replica identity`. Supports `DEFAULT`, `INDEX index_name`, `FULL`, and `NOTHING`. Applies only to `pgoutput` and overrides existing database settings. A specified index must meet PostgreSQL replica identity index requirements, and the account must have permission to alter the table.

#### `publish.via.partition.root`

Controls whether changes are published through the partition root table.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Applies to `pgoutput`. `true` uses the partition root table identity; `false` publishes partition changes directly. Coordinate this setting with the database-side publication settings.

#### `status.update.interval.ms`

The interval for sending replication connection status to PostgreSQL, in milliseconds.

* **Type**: `INT`
* **Default**: `10000`
* **Importance**: Medium
* **Valid Values / Notes**: A positive integer. This is not the Kafka offset commit interval.

#### `flush.lsn.source`

Controls whether to acknowledge processed LSNs to PostgreSQL.

* **Type**: `BOOLEAN`
* **Default**: `true`
* **Importance**: Low
* **Valid Values / Notes**: When `false`, an external mechanism must manage LSN advancement and WAL reclamation. Connector heartbeats alone cannot replace it.

#### `lsn.flush.timeout.ms`

The maximum wait for an LSN flush operation, in milliseconds.

* **Type**: `LONG`
* **Default**: `30000`
* **Importance**: Medium
* **Valid Values / Notes**: A positive integer. `lsn.flush.timeout.action` determines how a timeout is handled.

#### `lsn.flush.timeout.action`

Controls how LSN flush timeouts are handled.

* **Type**: `STRING`
* **Default**: `fail`
* **Importance**: Medium
* **Valid Values / Notes**: `fail` stops the Connector, `warn` logs a warning and continues, and `ignore` ignores the timeout and continues. Continued operation does not mean that the source has acknowledged the consumed position.

#### `xmin.fetch.interval.ms`

The interval for querying the replication slot's `xmin`, in milliseconds.

* **Type**: `LONG`
* **Default**: `0`
* **Importance**: Medium
* **Valid Values / Notes**: `0` disables tracking. More frequent queries increase database overhead.

### Capture Scope and Field Processing

#### `schema.include.list`

Captures only matching schemas.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: High
* **Valid Values / Notes**: Comma-separated valid regular expressions. Cannot be set together with `schema.exclude.list`. Tables must also satisfy table filters and fall within the publication scope.

#### `schema.exclude.list`

Excludes matching schemas.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Comma-separated valid regular expressions. Cannot be set together with `schema.include.list`.

#### `table.include.list`

Captures only matching tables.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: High
* **Valid Values / Notes**: Comma-separated regular expressions matching the complete `schema.table` name. Cannot be set together with `table.exclude.list`. Does not automatically expand a pre-created publication.

#### `table.exclude.list`

Excludes matching tables.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Comma-separated regular expressions matching the complete `schema.table` name. Cannot be set together with `table.include.list`.

#### `table.ignore.builtin`

Controls whether to ignore built-in tables.

* **Type**: `BOOLEAN`
* **Default**: `true`
* **Importance**: Low
* **Valid Values / Notes**: When enabled, excludes `pg_catalog`, `information_schema`, schemas starting with `pg_temp`, and the `spatial_ref_sys` table.

#### `column.include.list`

Outputs only matching columns.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Regular expressions match `schema.table.column`. Cannot be set together with `column.exclude.list`. Column filtering does not replace database access isolation.

#### `column.exclude.list`

Excludes matching columns from event fields.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Regular expressions match `schema.table.column`. Cannot be set together with `column.include.list`. Do not infer that the database does not read these columns.

#### `message.prefix.include.list`

Captures only logical decoding messages with matching prefixes.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: A list of valid regular expressions. Cannot be set together with `message.prefix.exclude.list`. Without filtering, all prefixes are captured.

#### `message.prefix.exclude.list`

Excludes logical decoding messages with matching prefixes.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: A list of valid regular expressions. Cannot be set together with `message.prefix.include.list`.

#### `skipped.operations`

Specifies operations to skip during streaming.

* **Type**: `LIST`
* **Default**: `t`
* **Importance**: Low
* **Valid Values / Notes**: `c` for inserts, `u` for updates, `d` for deletes, `t` for table truncation, and `none` to skip no operations. Skipping operations changes the state that downstream consumers can reconstruct.

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

Masks matching columns with a salted hash.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: The name is a parameterized pattern; replace the pattern portions with a hash algorithm and salt in the actual key. The value is a list of regular expressions matching fully qualified column names. Treat salts as sensitive configuration and do not use real salts in shared examples.

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

Replaces matching columns with the specified number of asterisks.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: The actual key uses a numeric character count, for example `column.mask.with.8.chars`; its value specifies column matching rules. The pattern in the name is not a property name that can be submitted directly.

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

Truncates matching columns to the specified number of characters.

* **Type**: `INT`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: The actual key uses a numeric character count, for example `column.truncate.to.16.chars`. This is a parameterized key pattern: the character count is in the actual key name, and the corresponding matching rule selects columns. Truncation is irreversible; evaluate its impact on business fields.

### Snapshots and Initial Data Loading

#### `snapshot.mode`

Selects the startup strategy for snapshots and subsequent streaming.

* **Type**: `STRING`
* **Default**: `initial`
* **Importance**: Medium
* **Valid Values / Notes**: `initial` performs the first initialization and then captures changes continuously. `always` takes a new snapshot on every startup before capturing changes. `initial_only` performs initialization only. `no_data` does not scan existing rows. `when_needed` allows a new snapshot when the required log position is unavailable. `configuration_based` uses the associated switches. `custom` uses an installed extension. `never` is still accepted but deprecated; use `no_data` instead. A new snapshot restores the current state, not lost historical changes.

#### `snapshot.isolation.mode`

Sets the snapshot transaction isolation level.

* **Type**: `STRING`
* **Default**: `serializable`
* **Importance**: Low
* **Valid Values / Notes**: `serializable`, `repeatable_read`, `read_committed`, `read_uncommitted`. Different isolation levels do not provide equivalent consistency. Evaluate the impact of business writes during the snapshot before changing this setting.

#### `snapshot.locking.mode`

Controls the locking strategy used when reading snapshot metadata.

* **Type**: `STRING`
* **Default**: `none`
* **Importance**: Low
* **Valid Values / Notes**: `none`, `shared`, `custom`. `none` requires avoiding schema changes during the snapshot. `shared` acquires `ACCESS SHARE` locks, blocking conflicting DDL without blocking ordinary row writes. `custom` uses an installed locking strategy.

#### `snapshot.lock.timeout.ms`

The maximum wait to acquire table locks at the start of a snapshot, in milliseconds.

* **Type**: `LONG`
* **Default**: `10000`
* **Importance**: Medium
* **Valid Values / Notes**: Applies when the selected locking strategy requires locks. A lock acquisition timeout terminates the snapshot; first check for conflicting transactions or DDL.

#### `snapshot.delay.ms`

The delay before starting a snapshot, in milliseconds.

* **Type**: `LONG`
* **Default**: `0`
* **Importance**: Low
* **Valid Values / Notes**: Delaying startup does not replace resource capacity planning.

#### `streaming.delay.ms`

The delay between snapshot completion and the start of streaming, in milliseconds.

* **Type**: `LONG`
* **Default**: `0`
* **Importance**: Low
* **Valid Values / Notes**: Changes during the delay still depend on WAL retained by the replication slot. Account for log retention space.

#### `snapshot.include.collection.list`

Limits the set of tables included in the snapshot.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Specifies table matching rules; selected tables must be within the capture scope. Affects only the snapshot scope and does not replace streaming table filters.

#### `snapshot.fetch.size`

Controls the fetch size for snapshot reads.

* **Type**: `INT`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: When unset, the PostgreSQL runtime falls back to `10240`. This fallback is not the default in the configuration definition. Consider row size and task memory when adjusting it.

#### `snapshot.max.threads`

Sets the maximum number of snapshot threads.

* **Type**: `INT`
* **Default**: `1`
* **Importance**: Medium
* **Valid Values / Notes**: Affects the snapshot phase, not the number of streaming tasks. Evaluate database read load before increasing parallelism.

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

Controls table processing order during the initial snapshot.

* **Type**: `STRING`
* **Default**: `disabled`
* **Importance**: Medium
* **Valid Values / Notes**: `disabled` does not sort by row count, `ascending` sorts by increasing row count, and `descending` sorts by decreasing row count.

#### `snapshot.select.statement.overrides`

Specifies tables whose snapshot queries are overridden.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: A comma-separated list of `schema.table` names. Each table needs a corresponding dynamic query setting using the same table name. Changes only existing-data reads, not the scope of subsequent logical replication.

#### `snapshot.select.statement.overrides.<schema>.<table>`

Supplies a snapshot SQL query for the specified table.

* **Type**: `STRING`
* **Default**: No ConfigDef default
* **Importance**: Not declared
* **Valid Values / Notes**: The actual table name must appear in `snapshot.select.statement.overrides`, and the value must be an SQL `SELECT`. The override is skipped if no value is provided. Filtering existing data changes the initial downstream baseline; it does not imply that subsequent changes are filtered by the same condition.

#### `snapshot.query.mode`

Selects how snapshot queries are built.

* **Type**: `STRING`
* **Default**: `select_all`
* **Importance**: Low
* **Valid Values / Notes**: `select_all` or `custom`. Custom mode requires the corresponding SPI implementation.

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

Specifies the custom snapshot query implementation name.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: With `snapshot.query.mode=custom`, use an installed `io.debezium.snapshot.spi.SnapshotQuery` implementation's `name()` value, not an arbitrary class name.

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

Specifies the custom snapshot locking implementation name.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: With `snapshot.locking.mode=custom`, use an installed `io.debezium.snapshot.spi.SnapshotLock` implementation's `name()` value.

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

Specifies the custom snapshot strategy name.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: With `snapshot.mode=custom`, use an installed `Snapshotter` implementation's `name()` value.

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

Controls whether the configuration-based strategy snapshots data.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Used only with `snapshot.mode=configuration_based`.

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

Controls whether the configuration-based strategy snapshots schemas.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Used only with `snapshot.mode=configuration_based`.

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

Controls whether the configuration-based strategy starts streaming after the snapshot.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Used only with `snapshot.mode=configuration_based`. Do not assume that this mode automatically captures changes continuously.

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

Controls whether the configuration-based strategy takes a snapshot on a schema error.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Used only with `snapshot.mode=configuration_based`; it does not replace diagnosis of the specific error.

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

Controls whether the configuration-based strategy takes a snapshot on a data position error.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Used only with `snapshot.mode=configuration_based`. Rereading current data does not recover lost historical events.

### Events, Message Keys, and Data Types

#### `message.key.columns`

Specifies custom message key columns per table.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Semicolon-separated mappings in the form `schema.table:column1,column2`. Unspecified tables typically use their primary keys. Custom message keys do not supply old key values missing from the database WAL or change replica identities.

#### `tombstones.on.delete`

Controls whether to send a null-valued record with the same key after a delete event.

* **Type**: `BOOLEAN`
* **Default**: `true`
* **Importance**: Medium
* **Valid Values / Notes**: `true` sends a delete event followed by a tombstone; `false` sends only the delete event. Tombstones support Kafka log compaction and do not guarantee automatic deletes in downstream databases.

#### `provide.transaction.metadata`

Controls the output of transaction metadata and event counts.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Low
* **Valid Values / Notes**: Enabling this setting adds transaction context and metadata output. It does not imply atomic consumption across topics or end-to-end transaction guarantees.

#### `transaction.metadata.factory`

Specifies the factory for transaction context and structures.

* **Type**: `CLASS`
* **Default**: `io.debezium.pipeline.txmetadata.DefaultTransactionMetadataFactory`
* **Importance**: Low
* **Valid Values / Notes**: Custom implementations must be accessible to the plugin class loader and meet transaction metadata interface requirements.

#### `extended.headers.enabled`

Controls whether to add Debezium context headers.

* **Type**: `BOOLEAN`
* **Default**: `true`
* **Importance**: Low
* **Valid Values / Notes**: Headers provide event tracing and source identification information. They do not replace message keys or event positions.

#### `decimal.handling.mode`

Selects the representation of `DECIMAL` and `NUMERIC` values.

* **Type**: `STRING`
* **Default**: `precise`
* **Importance**: Medium
* **Valid Values / Notes**: `precise` uses Connect Decimal and exact numeric values, `string` uses strings, and `double` uses floating-point values that may lose precision. Coordinate this choice with the consumer's type model.

#### `time.precision.mode`

Selects the precision and representation of dates, times, and timestamps.

* **Type**: `STRING`
* **Default**: `adaptive`
* **Importance**: Medium
* **Valid Values / Notes**: `adaptive`, `adaptive_time_microseconds`, `isostring`, `microseconds`, `nanoseconds`, `connect`. `adaptive` converts according to source column precision. `connect` uses Connect millisecond-precision representations and may not preserve higher precision.

#### `hstore.handling.mode`

Selects the representation of `HSTORE` values.

* **Type**: `STRING`
* **Default**: `json`
* **Importance**: Low
* **Valid Values / Notes**: `json` uses JSON strings; `map` uses key-value maps.

#### `binary.handling.mode`

Selects the representation of binary columns.

* **Type**: `STRING`
* **Default**: `bytes`
* **Importance**: Low
* **Valid Values / Notes**: `bytes`, `base64`, `hex`, `base64-url-safe`. Final byte serialization also depends on the converter.

#### `interval.handling.mode`

Selects the representation of `INTERVAL` values.

* **Type**: `STRING`
* **Default**: `numeric`
* **Importance**: Low
* **Valid Values / Notes**: `numeric` approximately converts intervals to microseconds; `string` uses exact ISO-format strings. For intervals containing units such as months, do not assume that numeric conversion preserves the original interval semantics.

#### `include.unknown.datatypes`

Controls whether to output fields with unrecognized data types.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: `false` omits unsupported fields. `true` uses an implementation-dependent binary representation and does not guarantee a corresponding business logical type for every custom type.

#### `unavailable.value.placeholder`

Represents TOAST field values not supplied by the database.

* **Type**: `STRING`
* **Default**: `__debezium_unavailable_value`
* **Importance**: Medium
* **Valid Values / Notes**: This is not SQL NULL or an actual business value. When the value starts with `hex:`, the remainder is parsed as hexadecimal bytes. Consumers must recognize unavailable values and avoid overwriting existing valid field values with the placeholder.

#### `schema.refresh.mode`

Controls when in-memory table schemas are refreshed.

* **Type**: `STRING`
* **Default**: `columns_diff`
* **Importance**: Medium
* **Valid Values / Notes**: `columns_diff` refreshes on column differences. `columns_diff_exclude_unchanged_toast` ignores differences caused by unchanged TOAST values, which may conceal dropped TOAST columns and leave the cache stale.

#### `schema.name.adjustment.mode`

Adjusts output schema names to suit the serialization format.

* **Type**: `STRING`
* **Default**: `none`
* **Importance**: Low
* **Valid Values / Notes**: `none`, `avro`, `avro_unicode`. Evaluate name collisions and downstream compatibility before conversion. This does not change actual database names.

#### `include.schema.comments`

Controls whether to include table and column comments in metadata.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Enabling this setting increases memory use. Evaluate comment size and whether comments contain information that should not be propagated.

#### `column.propagate.source.type`

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

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Comma-separated regular expressions matching fully qualified column names. Propagates type metadata without changing field values.

#### `datatype.propagate.source.type`

Propagates source type and length information by database type name.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: Comma-separated regular expressions matching database type names, not column names.

#### `sourceinfo.struct.maker`

Specifies the implementation that builds the event `source` schema and structure.

* **Type**: `CLASS`
* **Default**: `io.debezium.connector.postgresql.PostgresSourceInfoStructMaker`
* **Importance**: Low
* **Valid Values / Notes**: Custom implementations must conform to the `SourceInfoStructMaker` interface. Structural changes affect consumers that depend on source metadata.

#### `converters`

Registers Debezium custom field conversion extensions.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Use extension aliases with their respective `别名.type` and `别名.选项` settings, where the alias denotes each extension. Required classes must be installed. This setting is distinct from Kafka Connect message serialization converters.

#### `post.processors`

Registers event post-processing extensions.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Use processor aliases with their respective `别名.type` settings and configuration properties, where the alias denotes each processor. Install compatible implementations and evaluate processing overhead.

### Topic Naming

#### `topic.prefix`

Sets the topic namespace for the capture source.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: High
* **Required**: Yes
* **Valid Values / Notes**: Allows only letters, digits, dots, underscores, and hyphens. Use separate prefixes for different capture sources. Retain the original prefix during maintenance to avoid unintentionally changing output identity.

#### `topic.naming.strategy`

Specifies the topic naming strategy.

* **Type**: `CLASS`
* **Default**: `io.debezium.schema.SchemaTopicNamingStrategy`
* **Importance**: Medium
* **Valid Values / Notes**: Default table topic names join the topic prefix, schema, and table name with a delimiter. The following four settings apply to the default strategy; custom strategies may use different settings.

#### `topic.delimiter`

Specifies the delimiter for the default naming strategy.

* **Type**: `STRING`
* **Default**: `.`
* **Importance**: Low
* **Valid Values / Notes**: Must meet topic naming character requirements. Changing it changes topic names.

#### `topic.cache.size`

Sets the topic name cache capacity.

* **Type**: `INT`
* **Default**: `10000`
* **Importance**: Low
* **Valid Values / Notes**: Used to cache mappings from data collections to topic names. This is not a limit on the number of Kafka topics.

#### `topic.heartbeat.prefix`

Sets the heartbeat topic prefix for the default naming strategy.

* **Type**: `STRING`
* **Default**: `__debezium-heartbeat`
* **Importance**: Low
* **Valid Values / Notes**: By default, the heartbeat topic combines this prefix, the delimiter, and `topic.prefix`. Records are sent only when heartbeats are enabled.

#### `topic.transaction`

Sets the transaction topic suffix for the default naming strategy.

* **Type**: `STRING`
* **Default**: `transaction`
* **Importance**: Low
* **Valid Values / Notes**: The transaction topic combines `topic.prefix`, the delimiter, and this suffix. Transaction metadata must be enabled separately.

### Heartbeats, Signals, and Incremental Snapshots

#### `heartbeat.interval.ms`

Sets the interval for sending heartbeat records, in milliseconds.

* **Type**: `INT`
* **Default**: `0`
* **Importance**: Medium
* **Valid Values / Notes**: A non-negative integer; `0` disables heartbeats. Can help advance offsets when replication progresses but table events are filtered out. Kafka heartbeats alone do not guarantee that the database generates new WAL.

#### `heartbeat.topics.prefix`

Sets the generic heartbeat prefix property.

* **Type**: `STRING`
* **Default**: `__debezium-heartbeat`
* **Importance**: Low
* **Valid Values / Notes**: This property remains configurable, but the default `SchemaTopicNamingStrategy` uses the separate `topic.heartbeat.prefix` setting. Do not infer the default strategy's topic name from this property alone.

#### `heartbeat.action.query`

Specifies the query executed for database heartbeats.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Requires a nonzero `heartbeat.interval.ms`. If writes are used to advance WAL in a low-traffic database, the account needs the appropriate write privileges, and the relevant table must be in the `pgoutput` publication. The query target must be in the currently connected database.

#### `signal.data.collection`

Specifies the table used to receive source-side signals.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: PostgreSQL uses `schema.table`. Source-side signaling requires the appropriate signal table, capture scope, and privileges.

#### `signal.enabled.channels`

Selects enabled signaling channels.

* **Type**: `LIST`
* **Default**: `source`
* **Importance**: Medium
* **Valid Values / Notes**: A list of channel names. Selected channels require corresponding implementations and configuration; setting names alone is insufficient without channel resources.

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

The interval for checking registered signaling channels, in milliseconds.

* **Type**: `LONG`
* **Default**: `5000`
* **Importance**: Medium
* **Valid Values / Notes**: Shorter intervals increase polling overhead. This is not the interval for capturing business changes.

#### `incremental.snapshot.chunk.size`

Sets the maximum number of rows per incremental snapshot chunk.

* **Type**: `INT`
* **Default**: `1024`
* **Importance**: Medium
* **Valid Values / Notes**: Incremental snapshots capture additional existing data while streaming. Chunk size affects memory use and read load, not the number of streaming tasks.

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

Selects the management strategy for incremental snapshot watermark signals.

* **Type**: `STRING`
* **Default**: `INSERT_INSERT`
* **Importance**: Low
* **Valid Values / Notes**: Accepts `insert_insert` and `insert_delete`. The former writes both opening and closing watermark signals; the latter writes an opening signal and closes it by deletion. These operations write to the signal table and require appropriate privileges.

#### `notification.enabled.channels`

Selects notification output channels.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Medium
* **Valid Values / Notes**: A list of channel names. Selected channels must have available implementations. A notification topic is required when the list includes `sink`. This is notification output from a Source Connector, not a direction setting for another Connector type.

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

Specifies the output destination for the topic notification channel.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: High
* **Valid Values / Notes**: Required when `notification.enabled.channels` includes `sink`. Does not change business table event topics.

### Queues, Throughput, and Error Recovery

#### `max.batch.size`

Sets the maximum number of records delivered to the Worker in one batch.

* **Type**: `INT`
* **Default**: `2048`
* **Importance**: Medium
* **Valid Values / Notes**: The batch limit should be smaller than `max.queue.size`. Evaluate memory and latency before increasing it; a larger batch does not guarantee improved source-side commit capacity.

#### `max.queue.size`

Sets the maximum number of change events queued before delivery.

* **Type**: `INT`
* **Default**: `8192`
* **Importance**: Medium
* **Valid Values / Notes**: Must be greater than `0` and strictly greater than `max.batch.size`. Reaching the limit causes backpressure rather than silent drops; a backlog may still increase source-side WAL retention pressure.

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

Sets the byte capacity limit for the change event queue.

* **Type**: `LONG`
* **Default**: `0`
* **Importance**: Medium
* **Valid Values / Notes**: `0` disables the byte limit. When enabled, it constrains the queue together with the record count limit. It is not a total memory limit for the Worker JVM.

#### `poll.interval.ms`

The wait time when the queue contains no new events, in milliseconds.

* **Type**: `LONG`
* **Default**: `500`
* **Importance**: Medium
* **Valid Values / Notes**: This does not cause full-table database queries at the specified interval. Reducing the wait changes idle polling overhead.

#### `query.fetch.size`

Controls the fetch size of JDBC queries during streaming.

* **Type**: `INT`
* **Default**: `0`
* **Importance**: Medium
* **Valid Values / Notes**: `0` uses the JDBC default fetch size. This is distinct from snapshot batches and replication message batches.

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

Controls behavior after event processing failures.

* **Type**: `STRING`
* **Default**: `fail`
* **Importance**: Medium
* **Valid Values / Notes**: `fail`, `warn`, `ignore`, `skip`. `fail` stops processing. Skipping policies omit problematic events; warnings or continued operation do not imply data completeness.

#### `errors.max.retries`

Controls the maximum number of retries for connection-related errors.

* **Type**: `INT`
* **Default**: `-1`
* **Importance**: Low
* **Valid Values / Notes**: `-1` means unlimited retries, `0` disables retries, and a positive value specifies the retry count. Applies only to retriable errors and cannot recover WAL that has been removed.

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

The wait before restarting capture logic after a retriable exception, in milliseconds.

* **Type**: `LONG`
* **Default**: `10000`
* **Importance**: Low
* **Valid Values / Notes**: Used with the error retry policy. Permanent privilege errors require authorization changes, not continually shorter waits.

#### `executor.shutdown.timeout.ms`

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

* **Type**: `LONG`
* **Default**: `4000`
* **Importance**: Medium
* **Valid Values / Notes**: Allow time for task cleanup during shutdown. This does not replace retention planning for source slots and offsets.

### Extended Metadata and Data Lineage

#### `custom.metric.tags`

Adds custom tags to MBean object names.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Comma-separated key-value pairs, for example `environment=production`. Limit tag count and do not include credentials or sensitive business content.

#### `openlineage.integration.enabled`

Controls whether to output data lineage metadata through OpenLineage.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Low
* **Valid Values / Notes**: Prepare the corresponding OpenLineage configuration and available integration resources before enabling it.

#### `openlineage.integration.config.file.path`

Specifies the OpenLineage configuration file path.

* **Type**: `STRING`
* **Default**: `./openlineage.yml`
* **Importance**: Low
* **Valid Values / Notes**: The file must be readable by the running process. Protect sensitive information in the file according to credential management requirements.

#### `openlineage.integration.job.namespace`

Sets the namespace for the lineage job.

* **Type**: `STRING`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Used for lineage job identity. Does not change the captured database or topics.

#### `openlineage.integration.job.description`

Sets the lineage job description.

* **Type**: `STRING`
* **Default**: `Debezium change data capture job`
* **Importance**: Low
* **Valid Values / Notes**: Used only for metadata descriptions. Do not include sensitive requests or credentials.

#### `openlineage.integration.job.tags`

Sets lineage job tags.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Comma-separated key-value pairs. Avoid excessive tags and sensitive information.

#### `openlineage.integration.job.owners`

Sets lineage job owner metadata.

* **Type**: `LIST`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Comma-separated key-value pairs. Populate them according to your organization's metadata policy.

## Best Practices

### Establish an Initial Baseline and Continue Synchronizing Changes

**Applicable Scenario**: Connect for the first time to a PostgreSQL database with existing business data, establish a complete initial state for search, caching, or data synchronization, and continue receiving business changes rather than only new data after startup.

**Configuration Example**: Reuse the Quick Start connection, dedicated replication slot, pre-created publication, and table filter settings. Determine the complete capture scope before creating the Connector for the first time; no additional configuration is needed.

```properties theme={null}
connector.class=io.debezium.connector.postgresql.PostgresConnector
database.hostname=<database-host>
database.user=<replication-user>
database.password=<database-password>
database.dbname=<database-name>
topic.prefix=<topic-prefix>
plugin.name=pgoutput
slot.name=<replication-slot-name>
publication.name=<publication-name>
publication.autocreate.mode=disabled
table.include.list=<schema-table-regex>
```

**Key Considerations**: Check the table filters against the publication's table set before startup, then schedule the snapshot read window and avoid DDL during the snapshot. With no offset on first startup, the default `initial` mode emits `r` records and continues with changes after completion. It supports capturing changes made during the snapshot, but initialization reads add database load. Do not assume that expanding table filters automatically initializes new tables: with existing offsets, `initial` does not rescan existing data merely because tables were added. Restarting an incomplete snapshot may resend rows already read; downstream consumers must handle duplicates using stable message keys. For routine stops and starts, retain the Connector identity, topic prefix, offsets, and replication slot, and ensure the required WAL remains available. Do not drop the slot for routine maintenance.

### Keep Publication Tables Aligned When Maintaining the Capture Scope

**Applicable Scenario**: Expand capture from the initial table set to other business tables while keeping table filters aligned with the PostgreSQL publication, so that new table changes are not missed after changing only Connector settings. Applies to deployments where the capture team is authorized to manage a dedicated table-specific publication.

**Configuration Example**: Override the following two Quick Start settings. The table regular expression represents the complete table set after the change, and the publication name still refers to a table-specific publication dedicated to this Connector. Remove `table.exclude.list` first if it was previously set.

```properties theme={null}
publication.autocreate.mode=filtered
table.include.list=<complete-schema-table-regex>
```

**Key Considerations**: First confirm that the account has the privileges needed to manage the publication and add each table, then update the configuration and restart capture. During initialization, `filtered` creates or replaces the publication's table set according to the filters; it does not simply append tables to the original set. It cannot modify a `FOR ALL TABLES` publication and should not be used for a publication shared with other consumers. If added tables contain existing data, plan a separate baseline backfill. Updating the publication only addresses replication scope; it does not rerun a completed `initial` snapshot. For DBA-managed publications, retain `disabled` and have the DBA coordinate table set changes, without granting additional privileges for automatic management.

## Monitoring

### What to Monitor

Monitor Kafka Connect cluster health, Connector and task status, throughput and processing latency, offset commit progress, errors and retries, and Worker JVM heap, GC, and thread signals. Even when a task reports that it is running, confirm that records and offsets continue to advance. For prolonged backlogs, use source-side operational information to check WAL retention pressure from replication slots. Monitor DLQ activity only when the corresponding error handling is enabled; do not assume that every capture exception is written to a DLQ.

### Import the Grafana Dashboard

Download the [AutoMQ Connect Cluster Grafana Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), prepare a matching Prometheus-compatible data source and labels such as cluster and Worker, then import the JSON into Grafana and select the appropriate data source and cluster.

## Limitations

* One Connector captures only one database and creates one streaming task. Increasing `tasks.max` does not split the replication stream.
* Include and exclude settings at the same level are mutually exclusive. Connector table filters cannot compensate for tables missing from the publication.
* PostgreSQL logical decoding does not provide replayable SQL DDL events. Event schema updates do not constitute a complete DDL history.
* The completeness of `before` depends on replica identity and the old tuple supplied by the database. Do not assume that every update or delete contains a complete old row.
* Schema changes are not supported during incremental snapshots. Coordinate DDL windows accordingly.
* `TRUNCATE` is skipped by default. When enabled, it produces keyless `t` events, not per-row deletes or tombstones.
* Global consumption order and atomic visibility across tables are not guaranteed across Kafka topics or partitions. In particular, do not rely on global ordering between keyless `TRUNCATE` events and row events.
* After a replication slot is dropped or required WAL is removed, Kafka offsets cannot regenerate source log history. A new snapshot can rebuild the current state but cannot recover intermediate changes that are no longer available.
* Normal failure recovery and restarts of incomplete snapshots may resend records. The default configuration must not be interpreted as an end-to-end exactly-once synchronization guarantee.

## FAQ

### Existing Data Appears in the Snapshot, but Subsequent Updates Do Not Reach the Topic

First check the logical decoding plugin actually in use. Then verify that the `pgoutput` publication contains the target tables and publishes the required operations, and that `schema`, `table`, and `skipped.operations` filters do not exclude the events. Snapshots read through JDBC, so being able to read existing data does not prove that the publication scope is correct. For DBA-managed publications, have the DBA add the missing coverage. With `filtered`, confirm that the table-specific publication is dedicated to this Connector and that management privileges are available, then restart after updating the configuration. If updates or deletes fail in the database, also check the table's replica identity.

### Startup Fails with Insufficient Publication or Table Privileges

Check that the account has replication, database connection, schema usage, and table read privileges. Automatic publication creation also requires database `CREATE`; modifying a publication and adding tables involves object ownership. The default `all_tables` mode may attempt to create an all-table publication requiring superuser privileges. Have a DBA pre-create the required table-specific publication and set `publication.autocreate.mode=disabled` instead of promoting the capture account directly to superuser. With `filtered`, confirm that the existing publication is not `FOR ALL TABLES` and that the table filters match a nonempty set.

### A Delete Produces Two Records, and the Second Has a Null Value

This is the default delete behavior: a delete event with `op=d` and a null `after` is followed by a null-valued tombstone with the same key. Consumers must distinguish the delete envelope from the tombstone rather than treat a null-valued record as a parsing failure. Applications that reconstruct state must handle deletes. Log compaction uses tombstones to remove history for the same key; it does not automatically delete rows from downstream databases.

### Old Values Are Incomplete in Update Events, or Large Fields Contain an Unavailable Placeholder

Check the table's `REPLICA IDENTITY`. `DEFAULT` typically supplies only old key columns, and unchanged TOAST fields may also omit their actual values. The placeholder means unavailable, not SQL NULL; consumers can retain existing field values when a new value is unavailable. If complete old rows are genuinely required, have a DBA evaluate `FULL` and its additional WAL and processing costs. Do not enable it uniformly for all tables merely to eliminate placeholders. Custom message keys cannot recover old column values that the database did not send.

### Adding Table Filters Does Not Produce Existing Data for New Tables

`table.include.list` controls capture scope; it is not a request to rerun initialization. With existing offsets and a completed first snapshot, the default `initial` mode does not automatically rescan existing rows when tables are added. First verify the publication and privileges for the new tables. Then plan a supported incremental snapshot or a separate initialization capture process according to downstream baseline requirements. Do not try to fix this by clearing offsets or dropping slots directly, as that may affect the existing recovery boundary.

### WAL Usage Keeps Growing During Downtime

Replication slots retain WAL needed for capture. Check Connector and task status, backlogs, offset commits, and LSN acknowledgment progress. Confirm that `flush.lsn.source` has not been disabled without an external reclamation mechanism. Observe progress after consumption resumes, and plan disk capacity and server-side retention limits according to the permitted downtime. Consider heartbeats when many events are filtered out. Low-traffic databases may also need a database heartbeat query that generates WAL, with the relevant table included in the publication. Do not drop a slot that is still needed for recovery merely to free disk space.

### Restarting Produces Duplicates or Fails to Read the Original Log Position

Duplicates may result from offsets that were not yet persisted or rescans of incomplete snapshots. Downstream consumers should use stable keys and source positions to design idempotent processing. If the original position cannot be read, check whether capture identity changed, whether the replication slot was retained, and whether required WAL was removed. Increasing retry counts cannot recover missing logs. If rebuilding is necessary, coordinate the new snapshot baseline, existing downstream state, and historical event gaps. Even a strategy that permits a new snapshot cannot recover history that has already been lost.