PolarDB:Flink CDC compatible with PolarDB for PostgreSQL (Compatible with Oracle)

Pipeline connector (release-3.5+)

Release-3.5 and later also support the Pipeline connector mode, which reads both snapshot and incremental data for end-to-end full-database synchronization. Note that Pipeline connectors do not currently support synchronizing table schema changes. See the Postgres CDC Pipeline connector documentation.

Determine dependency versions

Before cloning any code, identify the correct versions for three dependencies.

Flink-CDC version

For Realtime Compute for Apache Flink, find the community Flink-CDC version that maps to your VVR version. See CDC and VVR version mapping. For the Flink-CDC repository, see Flink-CDC on GitHub.

Debezium version

In the pom.xml of your target Flink-CDC version, search for debezium.version. For the Debezium repository, see Debezium on GitHub.

PgJDBC version

In the pom.xml of the Postgres CDC module, search for org.postgresql. The file path differs by release:

• Releases earlier than 3.0: flink-connector-postgres-cdc/pom.xml

• Release 3.0 and later: flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml

For the PgJDBC repository, see PgJDBC on GitHub.

Build release-3.5

Flink-CDC release-3.5 is compatible with vvr-11.4-jdk11-flink-1.20.

1. Clone Flink-CDC, PgJDBC, and Debezium at the required versions.

   git clone -b release-3.5 --depth=1 https://github.com/apache/flink-cdc.git
   git clone -b REL42.7.3 --depth=1 https://github.com/pgjdbc/pgjdbc.git
   git clone -b v1.9.8.Final --depth=1 https://github.com/debezium/debezium.git

2. Copy the required files from PgJDBC and Debezium into the Flink-CDC source tree.

   mkdir -p flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3
   mkdir -p flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc
   cp pgjdbc/pgjdbc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3
   cp pgjdbc/pgjdbc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3
   cp pgjdbc/pgjdbc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc
   cp pgjdbc/pgjdbc/src/main/java/org/postgresql/core/Oid.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core
   cp debezium/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql

3. From the flink-cdc directory, apply two upstream bug fixes. These fixes will be merged into the community 3.6 release.

   cd flink-cdc
   
   # Timestamp conversion bug fix
   git fetch origin 2f32836a783f80f295c9dce339c11afec2a32dc2
   git cherry-pick 2f32836a783f80f295c9dce339c11afec2a32dc2
   
   # Implicit behavior (SELECT *) bug fix
   git fetch origin 0d86de24494a855c2d83f9b1052c2e888e182cb1
   git cherry-pick 0d86de24494a855c2d83f9b1052c2e888e182cb1

   For the full diff of each fix, see the timestamp conversion fix and the implicit behavior fix.

4. Apply the PolarDBO compatibility patch.

   git apply release-3.5_support_polardbo.patch

   Download the patch file: release-3.5_support_polardbo.patch.

5. Build the connector JAR with Maven (requires JDK 11).

   mvn clean install -DskipTests -Dcheckstyle.skip=true -Dspotless.check.skip
   
   # The JAR is output to:
   # flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/target

Prebuilt JAR (JDK 11): flink-cdc-pipeline-connector-polardbo-3.5-SNAPSHOT-20260212.jar

Build release-3.1

Flink-CDC release-3.1 is compatible with vvr-8.0.x-flink-1.17.

1. Clone Flink-CDC, PgJDBC, and Debezium at the required versions.

   git clone -b release-3.1 --depth=1 https://github.com/apache/flink-cdc.git
   git clone -b REL42.5.1 --depth=1 https://github.com/pgjdbc/pgjdbc.git
   git clone -b v1.9.8.Final --depth=1 https://github.com/debezium/debezium.git

2. Copy the required files from PgJDBC and Debezium into the Flink-CDC source tree.

   mkdir -p flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3
   mkdir -p flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc
   cp pgjdbc/pgjdbc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3
   cp pgjdbc/pgjdbc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3
   cp pgjdbc/pgjdbc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc
   cp debezium/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql

3. Apply the PolarDBO compatibility patch.

   git apply release-3.1_support_polardbo.patch

   Download the patch file: release-3.1_support_polardbo.patch.

4. Build the connector JAR with Maven (requires JDK 8).

   mvn clean install -DskipTests -Dcheckstyle.skip=true -Dspotless.check.skip -Drat.skip=true
   
   # The JAR is output to the target folder of flink-sql-connector-postgres-cdc

Prebuilt JAR (JDK 8): flink-sql-connector-postgres-cdc-3.1-SNAPSHOT.jar

Build release-2.3

Flink-CDC release-2.3 is compatible with vvr-4.0.15-flink-1.13 through vvr-6.0.2-flink-1.15.

1. Clone Flink-CDC, PgJDBC, and Debezium at the required versions.

   git clone -b release-2.3 --depth=1 https://github.com/apache/flink-cdc.git
   git clone -b REL42.2.26 --depth=1 https://github.com/pgjdbc/pgjdbc.git
   git clone -b v1.6.4.Final --depth=1 https://github.com/debezium/debezium.git

2. Copy the required files from PgJDBC and Debezium into the Flink-CDC source tree.

   mkdir -p flink-cdc/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3
   mkdir -p flink-cdc/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc
   mkdir -p flink-cdc/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql
   cp pgjdbc/pgjdbc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java flink-cdc/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3
   cp pgjdbc/pgjdbc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java flink-cdc/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3
   cp pgjdbc/pgjdbc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java flink-cdc/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc
   cp debezium/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java flink-cdc/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql

3. Apply the PolarDBO compatibility patch.

   git apply release-2.3_support_polardbo.patch

   Download the patch file: release-2.3_support_polardbo.patch.

4. Build the connector JAR with Maven (requires JDK 8).

   mvn clean install -DskipTests -Dcheckstyle.skip=true -Dspotless.check.skip -Drat.skip=true
   
   # The JAR is output to the target folder of flink-sql-connector-postgres-cdc

Prebuilt JAR (JDK 8): flink-sql-connector-postgres-cdc-2.3-SNAPSHOT.jar

Step 1: Set database parameters

Configure the following parameters on the PolarDB cluster. You can set them in the PolarDB console — see Set cluster parameters.

Step 2: Set REPLICA IDENTITY for each subscribed table

Run the following command for every table you want to capture:

ALTER TABLE schema.table REPLICA IDENTITY FULL;

REPLICA IDENTITY is a PostgreSQL table-level setting that controls whether the logical decoding plugin includes the old column values in INSERT and UPDATE events. Setting it to FULL ensures complete data synchronization.

SELECT relreplident = 'f' FROM pg_class WHERE relname = 'tablename';

Step 3: Verify account permissions

Connect using a privileged account or an account that has all of the following:

• LOGIN permission

• REPLICATION permission

• SELECT permission on each subscribed table (required for full snapshot reads)

Step 4: Use the primary endpoint

Connect only to the primary endpoint of the PolarDB cluster. Cluster endpoints do not support logical replication.

Additional configuration for partitioned tables (release-3.5+)

For partitioned tables, configure the following. See the Postgres CDC community documentation for details.

• Set scan.include-partitioned-tables.enabled to true.

• Manually create a PUBLICATION in the database with the publish_via_partition_root=true option, then specify its name using the debezium.publication.name parameter.

• Set table-name to match only parent tables. The regular expression must not match child tables, or full snapshot data will be duplicated.

Source connector

This example synchronizes the shipments table from the flink_source database in a PolarDB for PostgreSQL (Compatible with Oracle) 2.0 cluster to the shipments_sink table in the flink_sink database.

This example validates basic connectivity. For production jobs, tune parameters according to the Postgres CDC documentation.

Prerequisites

PolarDB cluster

1. Purchase a PolarDB for PostgreSQL (Compatible with Oracle) 2.0 cluster on the PolarDB cluster purchase page.

2. Create a privileged account.

3. View the primary endpoint. If Realtime Compute for Apache Flink and the PolarDB cluster are in the same virtual private cloud (VPC), use the private endpoint. Otherwise, apply for a public endpoint.

4. Add the Flink instance address to the cluster whitelist.

5. Create the flink_source and flink_sink databases in the console. See Create a database.

6. In flink_source, create the shipments table and insert a test row.

   CREATE TABLE public.shipments (
     shipment_id INT,
     order_id INT,
     origin TEXT,
     destination TEXT,
     is_arrived BOOLEAN,
     order_time DATE,
     PRIMARY KEY (shipment_id)
   );
   ALTER TABLE public.shipments REPLICA IDENTITY FULL;
   INSERT INTO public.shipments SELECT 1, 1, 'test1', 'test1', false, now();

7. In flink_sink, create the shipments_sink table. Note that order_time uses TIMESTAMP, not DATE.

   CREATE TABLE public.shipments_sink (
     shipment_id INT,
     order_id INT,
     origin TEXT,
     destination TEXT,
     is_arrived BOOLEAN,
     order_time TIMESTAMP,
     PRIMARY KEY (shipment_id)
   );

Realtime Compute for Apache Flink

1. Log on to the Realtime Compute console and purchase a Flink instance. See Purchase Realtime Compute for Apache Flink. Deploy the instance in the same region and VPC as the PolarDB cluster to use the private endpoint.

2. Create a custom connector and upload the PolarDBO Flink CDC JAR. Set Formats to debezium-json. See Create a custom connector.

SELECT * FROM public.shipments_sink;

 shipment_id | order_id | origin | destination | is_arrived |     order_time
-------------+----------+--------+-------------+------------+---------------------
           1 |        1 | test1  | test1       | f          | 2024-09-18 05:45:08
(1 row)

INSERT INTO public.shipments SELECT 2, 2, 'test2', 'test2', false, now();
UPDATE public.shipments SET is_arrived = true WHERE shipment_id = 1;
DELETE FROM public.shipments WHERE shipment_id = 2;
INSERT INTO public.shipments SELECT 3, 3, 'test3', 'test3', false, now();
UPDATE public.shipments SET is_arrived = true WHERE shipment_id = 3;

SELECT * FROM public.shipments_sink;

 shipment_id | order_id | origin | destination | is_arrived |     order_time
-------------+----------+--------+-------------+------------+---------------------
           1 |        1 | test1  | test1       | t          | 2024-09-18 05:45:08
           3 |        3 | test3  | test3       | t          | 2024-09-18 07:33:23
(2 rows)

Pipeline connector

This example uses the PolarDBO Flink CDC Pipeline connector to synchronize the shipments1 and shipments2 tables from the flink_source database in a PolarDB for PostgreSQL (Compatible with Oracle) 2.0 cluster. It uses the Print connector as a sink for debugging.

This example validates basic connectivity. For production jobs, configure parameters according to the Postgres CDC Pipeline connector documentation.

CreateTableEvent{tableId=public.shipments2, schema=columns={`shipment_id` INT NOT NULL,`order_id` INT,`origin` STRING,`destination` STRING,`is_arrived` BOOLEAN,`order_time` TIMESTAMP(6)}, primaryKeys=shipment_id, options=()}
CreateTableEvent{tableId=public.shipments1, schema=columns={`shipment_id` INT NOT NULL,`order_id` INT,`origin` STRING,`destination` STRING,`is_arrived` BOOLEAN,`order_time` TIMESTAMP(6)}, primaryKeys=shipment_id, options=()}
DataChangeEvent{tableId=public.shipments2, before=[], after=[1, 1, test1, test1, false, 2026-01-07T16:30:44], op=INSERT, meta=()}
DataChangeEvent{tableId=public.shipments1, before=[], after=[1, 1, test1, test1, false, 2026-01-07T16:30:44], op=INSERT, meta=()}

INSERT INTO public.shipments1 SELECT 2, 2, 'test2', 'test2', false, now();
UPDATE public.shipments1 SET is_arrived = true WHERE shipment_id = 1;
DELETE FROM public.shipments1 WHERE shipment_id = 2;
INSERT INTO public.shipments1 SELECT 3, 3, 'test3', 'test3', false, now();
UPDATE public.shipments1 SET is_arrived = true WHERE shipment_id = 3;

INSERT INTO public.shipments2 SELECT 2, 2, 'test2', 'test2', false, now();
UPDATE public.shipments2 SET is_arrived = true WHERE shipment_id = 1;
DELETE FROM public.shipments2 WHERE shipment_id = 2;
INSERT INTO public.shipments2 SELECT 3, 3, 'test3', 'test3', false, now();
UPDATE public.shipments2 SET is_arrived = true WHERE shipment_id = 3;

DataChangeEvent{tableId=public.shipments1, before=[], after=[2, 2, test2, test2, false, 2026-01-07T16:44:50], op=INSERT, meta=()}
DataChangeEvent{tableId=public.shipments1, before=[1, 1, test1, test1, false, 2026-01-07T16:30:44], after=[1, 1, test1, test1, true, 2026-01-07T16:30:44], op=UPDATE, meta=()}
DataChangeEvent{tableId=public.shipments1, before=[2, 2, test2, test2, false, 2026-01-07T16:44:50], after=[], op=DELETE, meta=()}
DataChangeEvent{tableId=public.shipments1, before=[], after=[3, 3, test3, test3, false, 2026-01-07T16:44:50], op=INSERT, meta=()}
DataChangeEvent{tableId=public.shipments1, before=[3, 3, test3, test3, false, 2026-01-07T16:44:50], after=[3, 3, test3, test3, true, 2026-01-07T16:44:50], op=UPDATE, meta=()}
DataChangeEvent{tableId=public.shipments2, before=[], after=[2, 2, test2, test2, false, 2026-01-07T16:44:50], op=INSERT, meta=()}
DataChangeEvent{tableId=public.shipments2, before=[1, 1, test1, test1, false, 2026-01-07T16:30:44], after=[1, 1, test1, test1, true, 2026-01-07T16:30:44], op=UPDATE, meta=()}
DataChangeEvent{tableId=public.shipments2, before=[2, 2, test2, test2, false, 2026-01-07T16:44:50], after=[], op=DELETE, meta=()}
DataChangeEvent{tableId=public.shipments2, before=[], after=[3, 3, test3, test3, false, 2026-01-07T16:44:50], op=INSERT, meta=()}
DataChangeEvent{tableId=public.shipments2, before=[3, 3, test3, test3, false, 2026-01-07T16:44:50], after=[3, 3, test3, test3, true, 2026-01-07T16:44:50], op=UPDATE, meta=()}