PostgreSQL CDC

Replication slots

Flink PostgreSQL CDC jobs use replication slots to prevent premature WAL purging and ensure data consistency. Poorly managed slots can cause excessive disk usage or read delays. Best practices:

• Clean up unused slots promptly

  • Flink does not automatically delete replication slots after a job stops or restarts statelessly, to prevent WAL data loss.

  • If a job will not restart, manually delete its replication slot to free disk space.

    Important

    Lifecycle management: Treat replication slots as job-level resources and manage them alongside job starts and stops.

• Avoid reusing old slots

  • Always use a new slot name. Reusing an old slot forces the job to read accumulated historical WAL data at startup, delaying new data processing.

  • PostgreSQL requires one slot per connection. Each job must use a unique slot name.

    Important

    Naming convention: When you customize slot.name, avoid names with numeric suffixes such as my_slot_1, to prevent conflicts with temporary slots.

• Slot behavior with incremental snapshots enabled

  • Prerequisites: Checkpoints must be enabled, and the source table must have a primary key defined.

  • Slot creation rules:

    • Incremental snapshot disabled: Only a parallelism of 1 is supported. One global slot is used.

    • Incremental snapshot enabled:

      • Snapshot phase: Each concurrent source subtask creates a temporary slot. The naming format is ${slot.name}_${task_id}.

      • Incremental phase: All temporary slots are automatically reclaimed. Only one global slot is retained.

  • Maximum number of slots: Source parallelism + 1 (during the snapshot phase)

• Resources and performance

  • If available slots or disk space is limited, reduce snapshot parallelism to use fewer temporary slots. This reduces snapshot read speed.

  • If the downstream sink supports idempotent writes, set scan.incremental.snapshot.backfill.skip = true to skip WAL backfill during the snapshot phase and accelerate startup.

    This provides only at-least-once semantics and is not suitable for stateful computations (aggregations or lookup joins) because required historical changes may be lost.

• When incremental snapshots are disabled, checkpoints are not supported during the snapshot phase.

  Configuration to avoid timeouts during the snapshot phase

  When incremental snapshots are disabled, checkpoints during the snapshot phase may cause timeout failovers. Configure these parameters in Other Configuration (Configure custom running parameters):

  execution.checkpointing.interval: 10min
  execution.checkpointing.tolerable-failed-checkpoints: 100
  restart-strategy: fixed-delay
  restart-strategy.fixed-delay.attempts: 2147483647

  Parameters:

  Parameter	Description	Remarks
  execution.checkpointing.interval	The interval between checkpoints.	The unit is a duration value, such as 10min or 30s.
  execution.checkpointing.tolerable-failed-checkpoints	The number of checkpoint failures that can be tolerated before the job fails.	The product of this parameter and the checkpoint scheduling interval is the allowed snapshot read time. Note If the table is very large, set this parameter to a larger value.
  restart-strategy	The job restart strategy.	Valid values: fixed-delay: Fixed-delay restart strategy. failure-rate: Failure-rate restart strategy. exponential-delay: Exponential-delay restart strategy. Restart Strategies.
  restart-strategy.fixed-delay.attempts	Max restart attempts for the fixed-delay restart strategy.	–

Reuse Postgres subscription

When Postgres CDC uses pgoutput, a publication determines which table changes can be read.

If multiple jobs share the same publication, make sure that the publication covers all tables that the jobs need to read. Otherwise, incremental changes of some tables may not be synchronized.

Considerations

The default value of debezium.publication.autocreate.mode is all_tables. In this mode:

• If the publication does not exist, the connector automatically creates a FOR ALL TABLES publication.

• If the publication already exists, the connector reuses it as is and does not check or modify its table scope.

If this option is set to filtered, the connector creates or updates the publication based on the table filter rules of the current job. When multiple jobs share the same publication, a job that starts later may overwrite the table scope and affect the reads of other jobs.

Recommendations

1. In production, create and maintain publications manually, and disable automatic publication creation and updates.

   -- Create a publication named my_flink_pub that includes all tables (or specified tables, creating one publication per job)
   CREATE PUBLICATION my_flink_pub FOR TABLE table_a, table_b;
   -- Or more simply, include all tables in the database
   CREATE PUBLICATION my_flink_pub FOR ALL TABLES;

   Note

   Do not subscribe to all tables in a database. For a very large database with a large number of tables, this wastes network bandwidth and increases CPU consumption on the Flink side.

2. Add the following Flink configurations:

   • debezium.publication.name = 'my_flink_pub' (Specifies the publication name)

   • debezium.publication.autocreate.mode = 'disabled' (Prevents Flink from trying to create or modify the publication on startup)

This prevents multiple jobs from overwriting each other's publication configurations.

Syntax

CREATE TABLE postgrescdc_source (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = '<user name>',
  'password' = '<password>',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>',
  'decoding.plugin.name'= 'pgoutput',
  'scan.incremental.snapshot.enabled' = 'true',
  -- Skipping backfill can speed up reads and reduce resource usage, but it may cause data duplication. Enable this if the downstream sink is idempotent.
  'scan.incremental.snapshot.backfill.skip' = 'false',
  -- In a production environment, set this to 'filtered' or 'disabled' and manage the publication manually instead of through Flink.
  'debezium-publication.autocreate.mode' = 'disabled'
  -- If you have multiple sources, configure a different publication for each source.
  --'debezium.publication.name' = 'my_flink_pub'
);

Connector options

Type mappings

Example

CREATE TABLE source (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = '<user name>',
  'password' = '<password>',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>'
);

SELECT * FROM source;

Flink CDC

VVR V11.4+ supports the PostgreSQL connector as a Flink CDC source.

Syntax

source:
  type: postgres
  name: PostgreSQL Source
  hostname: localhost
  port: 5432
  username: pg_username
  password: pg_password
  tables: db.scm.tbl
  slot.name: test_slot
  scan.startup.mode: initial
  server-time-zone: UTC
  connect.timeout: 120s
  decoding.plugin.name: decoderbufs

sink:
  type: ...

Connector options