Postgres CDC Connector #

Notes #

slot.name option #

The slot.name is recommended to set for different tables to avoid the potential PSQLException: ERROR: replication slot "flink" is active for PID 974 error. See more here.

scan.lsn-commit.checkpoints-num-delay option #

When consuming PostgreSQL logs, the LSN offset must be committed to trigger the log data cleanup for the corresponding slot. However, once the LSN offset is committed, earlier offsets become invalid. To ensure access to earlier LSN offsets for job recovery, we delay the LSN commit by scan.lsn-commit.checkpoints-num-delay (default value is 3) checkpoints. This feature is available when config option scan.incremental.snapshot.enabled is set to true.

Incremental Snapshot Options #

The following options is available only when scan.incremental.snapshot.enabled=true:

Available Metadata #

The following format metadata can be exposed as read-only (VIRTUAL) columns in a table definition.

Limitation #

Can’t perform checkpoint during scanning snapshot of tables when incremental snapshot is disabled #

When scan.incremental.snapshot.enabled=false, we have the following limitation.

During scanning snapshot of database tables, since there is no recoverable position, we can’t perform checkpoints. In order to not perform checkpoints, Postgres CDC source will keep the checkpoint waiting to timeout. The timeout checkpoint will be recognized as failed checkpoint, by default, this will trigger a failover for the Flink job. So if the database table is large, it is recommended to add following Flink configurations to avoid failover because of the timeout checkpoints:

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

The extended CREATE TABLE example demonstrates the syntax for exposing these metadata fields:

CREATE TABLE products (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    table_name STRING METADATA  FROM 'table_name' VIRTUAL,
    operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    shipment_id INT,
    order_id INT,
    origin STRING,
    destination STRING,
    is_arrived BOOLEAN
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = 'localhost',
  'port' = '5432',
  'username' = 'postgres',
  'password' = 'postgres',
  'database-name' = 'postgres',
  'schema-name' = 'public',
  'table-name' = 'shipments',
  'slot.name' = 'flink'
);

Features #

Incremental Snapshot Reading (Experimental) #

Incremental snapshot reading is a new mechanism to read snapshot of a table. Compared to the old snapshot mechanism, the incremental snapshot has many advantages, including:

• (1) PostgreSQL CDC Source can be parallel during snapshot reading
• (2) PostgreSQL CDC Source can perform checkpoints in the chunk granularity during snapshot reading
• (3) PostgreSQL CDC Source doesn’t need to acquire global read lock before snapshot reading

During the incremental snapshot reading, the PostgreSQL CDC Source firstly splits snapshot chunks (splits) by user specified chunk key of table, and then PostgreSQL CDC Source assigns the chunks to multiple readers to read the data of snapshot chunk.

Exactly-Once Processing #

The Postgres CDC connector is a Flink Source connector which will read database snapshot first and then continues to read binlogs with exactly-once processing even failures happen. Please read How the connector works.

Startup Reading Position #

The config option scan.startup.mode specifies the startup mode for PostgreSQL CDC consumer. The valid enumerations are:

• initial (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the replication slot.
• latest-offset: Never to perform snapshot on the monitored database tables upon first startup, just read from the end of the replication which means only have the changes since the connector was started.
• committed-offset: Skip snapshot phase and start reading events from a confirmed_flush_lsn offset of replication slot.
• snapshot: Only the snapshot phase is performed and exits after the snapshot phase reading is completed.

DataStream Source #

The Postgres CDC connector can also be a DataStream source. There are two modes for the DataStream source:

• incremental snapshot based, which allows parallel reading
• SourceFunction based, which only supports single thread reading

Incremental Snapshot based DataStream (Experimental) #

import org.apache.flink.cdc.connectors.base.source.jdbc.JdbcIncrementalSource;
import org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder;
import org.apache.flink.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class PostgresParallelSourceExample {

    public static void main(String[] args) throws Exception {

        DebeziumDeserializationSchema<String> deserializer =
                new JsonDebeziumDeserializationSchema();

        JdbcIncrementalSource<String> postgresIncrementalSource =
                PostgresSourceBuilder.PostgresIncrementalSource.<String>builder()
                        .hostname("localhost")
                        .port(5432)
                        .database("postgres")
                        .schemaList("inventory")
                        .tableList("inventory.products")
                        .username("postgres")
                        .password("postgres")
                        .slotName("flink")
                        .decodingPluginName("decoderbufs") // use pgoutput for PostgreSQL 10+
                        .deserializer(deserializer)
                        .splitSize(2) // the split size of each snapshot split
                        .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(3000);

        env.fromSource(
                        postgresIncrementalSource,
                        WatermarkStrategy.noWatermarks(),
                        "PostgresParallelSource")
                .setParallelism(2)
                .print();

        env.execute("Output Postgres Snapshot");
    }
}

SourceFunction-based DataStream #

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.cdc.connectors.postgres.PostgreSQLSource;

public class PostgreSQLSourceExample {
  public static void main(String[] args) throws Exception {
    SourceFunction<String> sourceFunction = PostgreSQLSource.<String>builder()
      .hostname("localhost")
      .port(5432)
      .database("postgres") // monitor postgres database
      .schemaList("inventory")  // monitor inventory schema
      .tableList("inventory.products") // monitor products table
      .username("flinkuser")
      .password("flinkpw")
      .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
      .build();

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env
      .addSource(sourceFunction)
      .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

    env.execute();
  }
}

Available Source metrics #

Metrics can help understand the progress of assignments, and the following are the supported Flink metrics:

Notice:

1. The group name is namespace.schema.table, where namespace is the actual database name, schema is the actual schema name, and table is the actual table name.
2. For PostgreSQL, the group name will be like test_database.test_schema.test_table.

Tables Without primary keys #

Starting from version 3.4.0, Postgres CDC support tables that do not have a primary key. To use a table without primary keys, you must configure the scan.incremental.snapshot.chunk.key-column option and specify one non-null field.

There are two places that need to be taken care of.

1. If there is an index in the table, try to use a column which is contained in the index in scan.incremental.snapshot.chunk.key-column. This will increase the speed of select statement.
2. The processing semantics of a Postgres CDC table without primary keys is determined based on the behavior of the column that are specified by the scan.incremental.snapshot.chunk.key-column.

• If no update operation is performed on the specified column, the exactly-once semantics is ensured.
• If the update operation is performed on the specified column, only the at-least-once semantics is ensured. However, you can specify primary keys at downstream and perform the idempotence operation to ensure data correctness.

Warning #

Using a non-primary key column as the scan.incremental.snapshot.chunk.key-column for a Postgres table with primary keys may lead to data inconsistencies. Below is a scenario illustrating this issue and recommendations to mitigate potential problems.

Problem Scenario #

• Table Structure:

  • Primary Key: id
  • Chunk Key Column: pid (Not a primary key)

• Snapshot Splits:

  • Split 0: 1 < pid <= 3
  • Split 1: 3 < pid <= 5

• Operation:

  • Two different subtasks are reading Split 0 and Split 1 concurrently.
  • An update operation changes pid from 2 to 4 for id=0 while both splits are being read. This update occurs between the low and high watermark of both splits.

• Result:

  • Split 0: Contains the record [id=0, pid=2]
  • Split 1: Contains the record [id=0, pid=4]

Since the order of processing these records cannot be guaranteed, the final value of pid for id=0 may end up being either 2 or 4, leading to potential data inconsistencies.

Data Type Mapping #

Back to top