Skip to main content

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