Realtime Compute for Apache Flink: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.

    Note

    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.

    Note

    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

The Postgres CDC connector relies on a publication to determine which tables' changes are pushed to a slot. If multiple jobs share the same publication, their configurations will be overwritten.

Cause

The default publication.autocreate.mode is filtered, which includes only tables in the connector configuration. This modifies the publication on job startup, which can affect other jobs.

Solution

1. Create a publication in PostgreSQL that includes all monitored tables, or create a separate publication per job.

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

   Subscribing to all tables is not recommended for large databases due to excessive bandwidth and CPU usage on the Flink cluster.

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 provides complete isolation and prevents new jobs from affecting existing ones.

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