This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new b11c961ad [FLINK-38907][postgres] Fix data loss with filtered
publication autocreate mode in backfill phase (#4223)
b11c961ad is described below
commit b11c961adecc6095b9bd97ff63df0bea4b22a1fa
Author: Giovanny Gutiérrez <[email protected]>
AuthorDate: Wed Oct 14 01:05:06 2026 -0500
[FLINK-38907][postgres] Fix data loss with filtered publication autocreate
mode in backfill phase (#4223)
---
.../fetch/PostgresSourceFetchTaskContext.java | 7 ++
.../postgres/source/PostgresSourceITCase.java | 88 +++++++++++++++++++++-
.../source/fetch/PostgresScanFetchTaskTest.java | 20 +++--
.../source/reader/PostgresSourceReaderTest.java | 2 +
.../postgres/testutils/UniqueDatabase.java | 9 +++
.../src/test/resources/ddl/customer.sql | 31 ++++++++
6 files changed, 145 insertions(+), 12 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresSourceFetchTaskContext.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresSourceFetchTaskContext.java
index ff94ecfcc..b518546d6 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresSourceFetchTaskContext.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresSourceFetchTaskContext.java
@@ -76,6 +76,7 @@ import static
io.debezium.connector.AbstractSourceInfo.SCHEMA_NAME_KEY;
import static io.debezium.connector.AbstractSourceInfo.TABLE_NAME_KEY;
import static
io.debezium.connector.postgresql.PostgresConnectorConfig.DROP_SLOT_ON_STOP;
import static
io.debezium.connector.postgresql.PostgresConnectorConfig.PLUGIN_NAME;
+import static
io.debezium.connector.postgresql.PostgresConnectorConfig.PUBLICATION_AUTOCREATE_MODE;
import static
io.debezium.connector.postgresql.PostgresConnectorConfig.SLOT_NAME;
import static
io.debezium.connector.postgresql.PostgresConnectorConfig.SNAPSHOT_MODE;
import static
io.debezium.connector.postgresql.PostgresObjectUtils.createReplicationConnection;
@@ -144,6 +145,12 @@ public class PostgresSourceFetchTaskContext extends
JdbcSourceFetchTaskContext {
.with(DROP_SLOT_ON_STOP.name(), true)
// Disable heartbeat event in snapshot
split fetcher
.with(Heartbeat.HEARTBEAT_INTERVAL, 0)
+ // Never touch the shared publication in
backfill tasks;
+ // filtered mode would shrink it to this
split's single table
+ // and other tables' changes would be
dropped by pgoutput
+ .with(
+ PUBLICATION_AUTOCREATE_MODE.name(),
+
PostgresConnectorConfig.AutoCreateMode.DISABLED)
.build());
} else {
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceITCase.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceITCase.java
index d884d079e..95c15d775 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceITCase.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceITCase.java
@@ -74,6 +74,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import java.util.stream.Collectors;
+import static
io.debezium.connector.postgresql.connection.ReplicationConnection.Builder.DEFAULT_PUBLICATION_NAME;
import static java.lang.String.format;
import static
org.apache.flink.cdc.connectors.postgres.testutils.PostgresTestUtils.hasNextData;
import static
org.apache.flink.cdc.connectors.postgres.testutils.PostgresTestUtils.triggerFailover;
@@ -185,6 +186,24 @@ class PostgresSourceITCase extends PostgresTestBase {
scanStartupMode);
}
+ @Test
+ void testFilteredPublication() throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put("debezium.publication.autocreate.mode", "filtered");
+ options.put("scan.incremental.snapshot.chunk.size", "10");
+ testPostgresParallelSource(
+ 1,
+ DEFAULT_SCAN_STARTUP_MODE,
+ PostgresTestUtils.FailoverType.NONE,
+ PostgresTestUtils.FailoverPhase.NEVER,
+ new String[] {"Customers", "customers_1", "customers_2"},
+ 1,
+ 0,
+ options,
+ this::checkPublicationForBackfillPhase,
+ false);
+ }
+
@ParameterizedTest
@ValueSource(strings = {"initial", "latest-offset"})
void testReadMultipleTableWithMultipleParallelism(String scanStartupMode)
throws Exception {
@@ -905,6 +924,31 @@ class PostgresSourceITCase extends PostgresTestBase {
Map<String, String> otherOptions,
StreamDataChecker streamDataChecker)
throws Exception {
+ testPostgresParallelSource(
+ parallelism,
+ scanStartupMode,
+ failoverType,
+ failoverPhase,
+ captureCustomerTables,
+ restartAttempts,
+ delayBetweenAttempts,
+ otherOptions,
+ streamDataChecker,
+ true);
+ }
+
+ private void testPostgresParallelSource(
+ int parallelism,
+ String scanStartupMode,
+ PostgresTestUtils.FailoverType failoverType,
+ PostgresTestUtils.FailoverPhase failoverPhase,
+ String[] captureCustomerTables,
+ int restartAttempts,
+ long delayBetweenAttempts,
+ Map<String, String> otherOptions,
+ StreamDataChecker streamDataChecker,
+ boolean checkSnapshot)
+ throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
@@ -917,7 +961,7 @@ class PostgresSourceITCase extends PostgresTestBase {
TableResult tableResult = tEnv.executeSql("select * from customers");
// first step: check the snapshot data
- if (DEFAULT_SCAN_STARTUP_MODE.equals(scanStartupMode)) {
+ if (checkSnapshot &&
DEFAULT_SCAN_STARTUP_MODE.equals(scanStartupMode)) {
checkSnapshotData(tableResult, failoverType, failoverPhase,
captureCustomerTables);
}
@@ -1036,6 +1080,48 @@ class PostgresSourceITCase extends PostgresTestBase {
expectedSnapshotData, fetchRows(iterator,
expectedSnapshotData.size()));
}
+ private void checkPublicationForBackfillPhase(
+ TableResult tableResult,
+ PostgresTestUtils.FailoverType failoverType,
+ PostgresTestUtils.FailoverPhase failoverPhase,
+ String[] captureCustomerTables)
+ throws Exception {
+ List<String> expectedTables =
+ Arrays.stream(captureCustomerTables)
+ .map(table -> SCHEMA_NAME + "." + table)
+ .collect(Collectors.toList());
+ // Backfill creates its slot after initializing the shared publication.
+ String query =
+ format(
+ "SELECT schemaname, tablename FROM
pg_publication_tables "
+ + "WHERE pubname = '%s' AND EXISTS ("
+ + "SELECT 1 FROM pg_replication_slots "
+ + "WHERE slot_name = '%s_0' AND database =
current_database())",
+ DEFAULT_PUBLICATION_NAME, slotName);
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
+ try (PostgresConnection connection = getConnection()) {
+ while (System.nanoTime() < deadline) {
+ List<String> publishedTables = new ArrayList<>();
+ connection.query(
+ query,
+ rs -> {
+ while (rs.next()) {
+ publishedTables.add(rs.getString(1) + "." +
rs.getString(2));
+ }
+ });
+ connection.commit();
+ if (!publishedTables.isEmpty()) {
+ assertThat(publishedTables)
+ .as("Publication must retain all captured tables
during backfill")
+
.containsExactlyInAnyOrderElementsOf(expectedTables);
+ return;
+ }
+ Thread.sleep(10L);
+ }
+ }
+ throw new AssertionError("No backfill publication observed within 30
seconds");
+ }
+
private void checkStreamData(
TableResult tableResult,
PostgresTestUtils.FailoverType failoverType,
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTaskTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTaskTest.java
index eb9834b17..edceb8197 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTaskTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTaskTest.java
@@ -45,10 +45,12 @@ import org.apache.flink.table.types.DataType;
import io.debezium.connector.postgresql.connection.PostgresConnection;
import org.apache.kafka.connect.source.SourceRecord;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
+import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -56,6 +58,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Optional;
+import static
io.debezium.connector.postgresql.connection.ReplicationConnection.Builder.DEFAULT_PUBLICATION_NAME;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link PostgresScanFetchTask}. */
@@ -75,10 +78,14 @@ class PostgresScanFetchTaskTest extends PostgresTestBase {
POSTGRES_CONTAINER.getUsername(),
POSTGRES_CONTAINER.getPassword());
- @Test
- void testChangingDataInSnapshotScan() throws Exception {
+ @BeforeEach
+ void before() throws SQLException {
customDatabase.createAndInitialize();
+ customDatabase.createPublication(DEFAULT_PUBLICATION_NAME);
+ }
+ @Test
+ void testChangingDataInSnapshotScan() throws Exception {
TestTableId tableId = new TestTableId(schemaName, tableName);
String[] changingDataSql =
new String[] {
@@ -112,7 +119,6 @@ class PostgresScanFetchTaskTest extends PostgresTestBase {
@Test
void testInsertDataInSnapshotScan() throws Exception {
- customDatabase.createAndInitialize();
TestTableId tableId = new TestTableId(schemaName, tableName);
String[] insertDataSql =
new String[] {
@@ -145,7 +151,6 @@ class PostgresScanFetchTaskTest extends PostgresTestBase {
@Test
void testDeleteDataInSnapshotScan() throws Exception {
- customDatabase.createAndInitialize();
TestTableId tableId = new TestTableId(schemaName, tableName);
String[] deleteDataSql =
new String[] {
@@ -170,8 +175,6 @@ class PostgresScanFetchTaskTest extends PostgresTestBase {
@Test
void testSnapshotScanSkipBackfillWithPostLowWatermark() throws Exception {
- customDatabase.createAndInitialize();
-
TestTableId tableId = new TestTableId(schemaName, tableName);
String[] changingDataSql =
new String[] {
@@ -207,8 +210,6 @@ class PostgresScanFetchTaskTest extends PostgresTestBase {
@Test
void testSnapshotScanSkipBackfillWithPreHighWatermark() throws Exception {
- customDatabase.createAndInitialize();
-
TestTableId tableId = new TestTableId(schemaName, tableName);
String[] changingDataSql =
new String[] {
@@ -244,8 +245,6 @@ class PostgresScanFetchTaskTest extends PostgresTestBase {
@Test
void testMultipleSplitsWithBackfill() throws Exception {
- customDatabase.createAndInitialize();
-
TestTableId tableId = new TestTableId(schemaName, tableName);
PostgresSourceConfigFactory sourceConfigFactory =
getMockPostgresSourceConfigFactory(
@@ -295,7 +294,6 @@ class PostgresScanFetchTaskTest extends PostgresTestBase {
@Test
void testSnapshotFetchSize() throws Exception {
- customDatabase.createAndInitialize();
PostgresSourceConfigFactory sourceConfigFactory =
getMockPostgresSourceConfigFactory(
customDatabase, schemaName, tableName, null, 10, true);
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java
index c8ead39f8..10ba00328 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java
@@ -61,6 +61,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
+import static
io.debezium.connector.postgresql.connection.ReplicationConnection.Builder.DEFAULT_PUBLICATION_NAME;
import static org.apache.flink.core.io.InputStatus.END_OF_INPUT;
import static org.apache.flink.core.io.InputStatus.MORE_AVAILABLE;
import static org.assertj.core.api.Assertions.assertThat;
@@ -81,6 +82,7 @@ public class PostgresSourceReaderTest extends
PostgresTestBase {
@BeforeEach
public void before() throws SQLException {
customDatabase.createAndInitialize();
+ customDatabase.createPublication(DEFAULT_PUBLICATION_NAME);
this.slotName = getSlotName();
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/testutils/UniqueDatabase.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/testutils/UniqueDatabase.java
index fb46b6d70..23c736031 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/testutils/UniqueDatabase.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/testutils/UniqueDatabase.java
@@ -165,6 +165,15 @@ public class UniqueDatabase {
}
}
+ /** Creates a publication for tests that bypass the source enumerator. */
+ public void createPublication(String publicationName) throws SQLException {
+ try (Connection connection =
PostgresTestBase.getJdbcConnection(container, databaseName);
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format("CREATE PUBLICATION %s FOR ALL TABLES",
publicationName));
+ }
+ }
+
/** Drop the database if it is existing. */
public void dropDatabase() {
try {
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/resources/ddl/customer.sql
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/resources/ddl/customer.sql
index 3626589aa..5a0db448c 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/resources/ddl/customer.sql
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/resources/ddl/customer.sql
@@ -81,6 +81,37 @@ VALUES (101,'user_1','Shanghai','123567891234'),
(1019,'user_20','Shanghai','123567891234'),
(2000,'user_21','Shanghai','123567891234');
+CREATE TABLE customers_2 (
+ "Id" INTEGER NOT NULL PRIMARY KEY,
+ "Name" VARCHAR(255) NOT NULL DEFAULT 'flink',
+ address VARCHAR(1024),
+ phone_number VARCHAR(512)
+);
+ALTER TABLE customers_2 REPLICA IDENTITY FULL;
+
+INSERT INTO customers_2
+VALUES (101,'user_1','Shanghai','123567891234'),
+ (102,'user_2','Shanghai','123567891234'),
+ (103,'user_3','Shanghai','123567891234'),
+ (109,'user_4','Shanghai','123567891234'),
+ (110,'user_5','Shanghai','123567891234'),
+ (111,'user_6','Shanghai','123567891234'),
+ (118,'user_7','Shanghai','123567891234'),
+ (121,'user_8','Shanghai','123567891234'),
+ (123,'user_9','Shanghai','123567891234'),
+ (1009,'user_10','Shanghai','123567891234'),
+ (1010,'user_11','Shanghai','123567891234'),
+ (1011,'user_12','Shanghai','123567891234'),
+ (1012,'user_13','Shanghai','123567891234'),
+ (1013,'user_14','Shanghai','123567891234'),
+ (1014,'user_15','Shanghai','123567891234'),
+ (1015,'user_16','Shanghai','123567891234'),
+ (1016,'user_17','Shanghai','123567891234'),
+ (1017,'user_18','Shanghai','123567891234'),
+ (1018,'user_19','Shanghai','123567891234'),
+ (1019,'user_20','Shanghai','123567891234'),
+ (2000,'user_21','Shanghai','123567891234');
+
CREATE TABLE customers_no_pk (
"Id" INTEGER NOT NULL,
"Name" VARCHAR(255) NOT NULL DEFAULT 'flink',
