This is an automated email from the ASF dual-hosted git repository.
lvyanquan 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 5b5d1dbe3 [FLINK-34806][postgres] Expose
scan.newly-added-table.enabled option (#4436)
5b5d1dbe3 is described below
commit 5b5d1dbe3e32b815502f6e07ca42f9a6159bfe0e
Author: Mehmet Can Şakiroğlu <[email protected]>
AuthorDate: Wed Sep 23 12:37:17 2026 +0300
[FLINK-34806][postgres] Expose scan.newly-added-table.enabled option (#4436)
Signed-off-by: Mehmet Can Şakiroğlu <[email protected]>
Co-authored-by: lvyanquan <[email protected]>
---
.../connectors/pipeline-connectors/postgres.md | 12 +
.../connectors/pipeline-connectors/postgres.md | 13 +
.../factory/PostgresDataSourceFactory.java | 4 +
.../postgres/source/PostgresDataSourceOptions.java | 12 +
.../factory/PostgresDataSourceFactoryTest.java | 20 ++
.../PostgresPipelineNewlyAddedTableITCase.java | 400 +++++++++++++++++++++
.../postgres/testutils/UniqueDatabase.java | 38 +-
7 files changed, 483 insertions(+), 16 deletions(-)
diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/postgres.md
b/docs/content.zh/docs/connectors/pipeline-connectors/postgres.md
index 02fb272cf..4a231d4ec 100644
--- a/docs/content.zh/docs/connectors/pipeline-connectors/postgres.md
+++ b/docs/content.zh/docs/connectors/pipeline-connectors/postgres.md
@@ -283,6 +283,18 @@ pipeline:
默认值为 false。
</td>
</tr>
+ <tr>
+ <td>scan.newly-added-table.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ 是否扫描新增的表。默认值为 false。
+ 该选项仅在从 savepoint 或 checkpoint 恢复时生效，
+ 会启用 SnapshotSplitAssigner#captureNewlyAddedTables() 代码路径，
+ 以发现在 savepoint 时刻未被捕获但匹配 source <code>tables</code> 模式的新表。
+ </td>
+ </tr>
<tr>
<td>schema-change.enabled</td>
<td>optional</td>
diff --git a/docs/content/docs/connectors/pipeline-connectors/postgres.md
b/docs/content/docs/connectors/pipeline-connectors/postgres.md
index 1deb246f5..174261258 100644
--- a/docs/content/docs/connectors/pipeline-connectors/postgres.md
+++ b/docs/content/docs/connectors/pipeline-connectors/postgres.md
@@ -275,6 +275,19 @@ pipeline:
Defaults to false.
</td>
</tr>
+ <tr>
+ <td>scan.newly-added-table.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to scan the newly added tables or not. Defaults to false.
+ This option only takes effect when restoring from a savepoint or
checkpoint,
+ and enables the existing
SnapshotSplitAssigner#captureNewlyAddedTables() code path
+ to discover tables that match the source <code>tables</code> pattern
but were not part of
+ the captured set at savepoint time.
+ </td>
+ </tr>
<tr>
<td>schema-change.enabled</td>
<td>optional</td>
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java
index bb6332f01..5a7f5f932 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java
@@ -72,6 +72,7 @@ import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSource
import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED;
import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_LSN_COMMIT_CHECKPOINTS_DELAY;
+import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE;
import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_STARTUP_MODE;
import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCHEMA_CHANGE_ENABLED;
@@ -133,6 +134,7 @@ public class PostgresDataSourceFactory implements
DataSourceFactory {
int lsnCommitCheckpointsDelay =
config.get(SCAN_LSN_COMMIT_CHECKPOINTS_DELAY);
boolean tableIdIncludeDatabase = config.get(TABLE_ID_INCLUDE_DATABASE);
boolean includeSchemaChanges = config.get(SCHEMA_CHANGE_ENABLED);
+ boolean scanNewlyAddedTableEnabled =
config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED);
validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize,
1);
validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1);
@@ -175,6 +177,7 @@ public class PostgresDataSourceFactory implements
DataSourceFactory {
.assignUnboundedChunkFirst(isAssignUnboundedChunkFirst)
.includeDatabaseInTableId(tableIdIncludeDatabase)
.includeSchemaChanges(includeSchemaChanges)
+ .scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
.getConfigFactory();
List<TableId> tableIds =
PostgresSchemaUtils.listTables(configFactory.create(0), null);
@@ -266,6 +269,7 @@ public class PostgresDataSourceFactory implements
DataSourceFactory {
options.add(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED);
options.add(TABLE_ID_INCLUDE_DATABASE);
options.add(SCHEMA_CHANGE_ENABLED);
+ options.add(SCAN_NEWLY_ADDED_TABLE_ENABLED);
return options;
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java
index 95cc823d2..183d7cc89 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java
@@ -281,4 +281,16 @@ public class PostgresDataSourceOptions {
.defaultValue(false)
.withDescription(
"Whether to infer CDC column types when processing
pgoutput Relation messages.");
+
+ public static final ConfigOption<Boolean> SCAN_NEWLY_ADDED_TABLE_ENABLED =
+ ConfigOptions.key("scan.newly-added-table.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to scan the newly added tables or not.
Defaults to false. "
+ + "This option only takes effect when
restoring from a savepoint or checkpoint, "
+ + "and enables the existing
SnapshotSplitAssigner#captureNewlyAddedTables() code path "
+ + "to discover tables that match the
source `tables:` pattern but were not part of "
+ + "the captured set at savepoint time.
Mirrors the MySQL Pipeline connector option "
+ + "of the same name.");
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java
index c3cce43bc..f64a0935b 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java
@@ -48,6 +48,7 @@ import java.util.stream.Collectors;
import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.HOSTNAME;
import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.PASSWORD;
import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.PG_PORT;
+import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SLOT_NAME;
import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.TABLES;
import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.TABLES_EXCLUDE;
@@ -97,6 +98,25 @@ public class PostgresDataSourceFactoryTest extends
PostgresTestBase {
.isEqualTo(Arrays.asList("inventory.products"));
}
+ @Test
+ public void testScanNewlyAddedTableEnabled() {
+ Map<String, String> options = new HashMap<>();
+ options.put(HOSTNAME.key(), POSTGRES_CONTAINER.getHost());
+ options.put(
+ PG_PORT.key(),
String.valueOf(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT)));
+ options.put(USERNAME.key(), TEST_USER);
+ options.put(PASSWORD.key(), TEST_PASSWORD);
+ options.put(TABLES.key(), POSTGRES_CONTAINER.getDatabaseName() +
".inventory.prod\\.*");
+ options.put(SLOT_NAME.key(), slotName);
+ options.put(SCAN_NEWLY_ADDED_TABLE_ENABLED.key(), "true");
+
+ Factory.Context context = new
MockContext(Configuration.fromMap(options));
+ PostgresDataSourceFactory factory = new PostgresDataSourceFactory();
+ PostgresDataSource dataSource = (PostgresDataSource)
factory.createDataSource(context);
+
+
assertThat(dataSource.getPostgresSourceConfig().isScanNewlyAddedTableEnabled()).isTrue();
+ }
+
@Test
public void testNoMatchedTable() {
Map<String, String> options = new HashMap<>();
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresPipelineNewlyAddedTableITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresPipelineNewlyAddedTableITCase.java
new file mode 100644
index 000000000..20a1c57a8
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresPipelineNewlyAddedTableITCase.java
@@ -0,0 +1,400 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.factories.Factory;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.FlinkSourceProvider;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.connectors.postgres.PostgresTestBase;
+import
org.apache.flink.cdc.connectors.postgres.factory.PostgresDataSourceFactory;
+import org.apache.flink.cdc.connectors.postgres.testutils.UniqueDatabase;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.StateRecoveryOptions;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import
org.apache.flink.streaming.api.operators.collect.AbstractCollectResultBuffer;
+import
org.apache.flink.streaming.api.operators.collect.CheckpointedCollectResultBuffer;
+import org.apache.flink.streaming.api.operators.collect.CollectResultIterator;
+import
org.apache.flink.streaming.api.operators.collect.CollectResultIteratorAdapter;
+import org.apache.flink.streaming.api.operators.collect.CollectSinkOperator;
+import
org.apache.flink.streaming.api.operators.collect.CollectSinkOperatorFactory;
+import org.apache.flink.streaming.api.operators.collect.CollectStreamSink;
+import org.apache.flink.streaming.util.RestartStrategyUtils;
+
+import io.debezium.connector.postgresql.connection.PostgresConnection;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+import static java.lang.String.format;
+import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.DECODING_PLUGIN_NAME;
+import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.HOSTNAME;
+import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.PASSWORD;
+import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.PG_PORT;
+import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
+import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SERVER_TIME_ZONE;
+import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SLOT_NAME;
+import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.TABLES;
+import static
org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.USERNAME;
+import static org.assertj.core.api.Assertions.assertThat;
+import static
org.testcontainers.containers.PostgreSQLContainer.POSTGRESQL_PORT;
+
+/** IT tests to cover newly added tables during capture process in Postgres
pipeline mode. */
+class PostgresPipelineNewlyAddedTableITCase extends PostgresTestBase {
+
+ private static final String SCHEMA_NAME = "customer";
+
+ private final UniqueDatabase customDatabase =
+ new UniqueDatabase(
+ POSTGRES_CONTAINER, "postgres", SCHEMA_NAME, TEST_USER,
TEST_PASSWORD);
+
+ private String slotName;
+
+ @BeforeEach
+ void before() throws SQLException {
+ customDatabase.createAndInitialize();
+ this.slotName = getSlotName();
+ }
+
+ @AfterEach
+ void after() throws Exception {
+ // sleep 1000ms to wait until connections are closed.
+ Thread.sleep(1000L);
+ customDatabase.removeSlot(slotName);
+ }
+
+ private PostgresConnection getConnection() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("hostname", customDatabase.getHost());
+ properties.put("port",
String.valueOf(customDatabase.getDatabasePort()));
+ properties.put("user", customDatabase.getUsername());
+ properties.put("password", customDatabase.getPassword());
+ properties.put("dbname", customDatabase.getDatabaseName());
+ return createConnection(properties);
+ }
+
+ @Test
+ void testAddNewTableByPatternSingleParallelism() throws Exception {
+ // step 1: create the first table
+ initialAddressTables(getConnection(),
Collections.singletonList("address_hangzhou"));
+
+ Path savepointDir = Files.createTempDirectory("add-new-table-test");
+ final String savepointDirectory =
savepointDir.toAbsolutePath().toString();
+ String finishedSavePointPath = null;
+
+ // step 2: listen to the pattern in the first round with single
parallelism
+ // so that each table emits exactly one CreateTableEvent.
+ StreamExecutionEnvironment env =
getStreamExecutionEnvironment(finishedSavePointPath, 1);
+ List<String> tables = Collections.singletonList("address_\\.*");
+ FlinkSourceProvider sourceProvider = getFlinkSourceProvider(tables, 1);
+ DataStreamSource<Event> source =
+ env.fromSource(
+ sourceProvider.getSource(),
+ WatermarkStrategy.noWatermarks(),
+ PostgresDataSourceFactory.IDENTIFIER,
+ new EventTypeInfo());
+
+ TypeSerializer<Event> serializer =
+ source.getTransformation()
+ .getOutputType()
+
.createSerializer(env.getConfig().getSerializerConfig());
+ CheckpointedCollectResultBuffer<Event> resultBuffer =
+ new CheckpointedCollectResultBuffer<>(serializer);
+ String accumulatorName = "dataStreamCollect_" + UUID.randomUUID();
+ CollectResultIterator<Event> iterator =
+ addCollector(env, source, resultBuffer, serializer,
accumulatorName);
+ JobClient jobClient = env.executeAsync("beforeAddNewTable");
+ iterator.setJobClient(jobClient);
+
+ // step 3: assert fetched snapshot data of the first table
+ List<Event> actual = fetchResults(iterator, 4);
+ multiAssert(actual, Collections.singletonList("address_hangzhou"));
+
+ // step 4: create a newly added table matching the pattern
+ initialAddressTables(getConnection(),
Collections.singletonList("address_beijing"));
+
+ // sleep 1s to wait for the assign status to
INITIAL_ASSIGNING_FINISHED.
+ // Otherwise, the restart job won't read newly added tables, and this
test will be stuck.
+ Thread.sleep(1000L);
+
+ // step 5: trigger a savepoint and cancel the job
+ finishedSavePointPath = triggerSavepointWithRetry(jobClient,
savepointDirectory);
+ jobClient.cancel().get();
+ iterator.close();
+
+ // step 6: restore from savepoint with the same pattern
+ StreamExecutionEnvironment restoredEnv =
+ getStreamExecutionEnvironment(finishedSavePointPath, 1);
+ FlinkSourceProvider restoredSourceProvider =
getFlinkSourceProvider(tables, 1);
+ DataStreamSource<Event> restoreSource =
+ restoredEnv.fromSource(
+ restoredSourceProvider.getSource(),
+ WatermarkStrategy.noWatermarks(),
+ PostgresDataSourceFactory.IDENTIFIER,
+ new EventTypeInfo());
+ CollectResultIterator<Event> restoredIterator =
+ addCollector(restoredEnv, restoreSource, resultBuffer,
serializer, accumulatorName);
+ restoredEnv.executeAsync("AfterAddNewTable");
+
+ // step 7: assert the newly added table is captured after restore
+ List<Event> newlyTableEvent = fetchResults(restoredIterator, 4);
+ multiAssert(newlyTableEvent,
Collections.singletonList("address_beijing"));
+ }
+
+ private void multiAssert(List<Event> actualEvents, List<String>
listenTables) {
+ List<Event> expectedCreateTableEvents = new ArrayList<>();
+ List<Event> expectedDataChangeEvents = new ArrayList<>();
+ for (String table : listenTables) {
+
expectedCreateTableEvents.add(getCreateTableEvent(TableId.tableId(SCHEMA_NAME,
table)));
+ expectedDataChangeEvents.addAll(
+ getSnapshotExpected(TableId.tableId(SCHEMA_NAME, table)));
+ }
+
+ // compare create table events
+ List<Event> actualCreateTableEvents =
+ actualEvents.stream()
+ .filter(event -> event instanceof CreateTableEvent)
+ .collect(Collectors.toList());
+ assertThat(actualCreateTableEvents)
+
.containsExactlyInAnyOrder(expectedCreateTableEvents.toArray(new Event[0]));
+
+ // compare data change events
+ List<Event> actualDataChangeEvents =
+ actualEvents.stream()
+ .filter(event -> event instanceof DataChangeEvent)
+ .collect(Collectors.toList());
+ assertThat(actualDataChangeEvents)
+
.containsExactlyInAnyOrder(expectedDataChangeEvents.toArray(new Event[0]));
+ }
+
+ private CreateTableEvent getCreateTableEvent(TableId tableId) {
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.BIGINT().notNull())
+ .physicalColumn("country",
DataTypes.VARCHAR(255).notNull())
+ .physicalColumn("city",
DataTypes.VARCHAR(255).notNull())
+ .physicalColumn("detail_address",
DataTypes.VARCHAR(1024))
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+ return new CreateTableEvent(tableId, schema);
+ }
+
+ private List<Event> getSnapshotExpected(TableId tableId) {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.BIGINT().notNull(),
+ DataTypes.VARCHAR(255).notNull(),
+ DataTypes.VARCHAR(255).notNull(),
+ DataTypes.VARCHAR(1024)
+ },
+ new String[] {"id", "country", "city",
"detail_address"});
+ BinaryRecordDataGenerator generator = new
BinaryRecordDataGenerator(rowType);
+ String cityName = tableId.getTableName().split("_")[1];
+ return Arrays.asList(
+ DataChangeEvent.insertEvent(
+ tableId,
+ generator.generate(
+ new Object[] {
+ 416874195632735147L,
+ BinaryStringData.fromString("China"),
+ BinaryStringData.fromString(cityName),
+ BinaryStringData.fromString(cityName + "
West Town address 1")
+ })),
+ DataChangeEvent.insertEvent(
+ tableId,
+ generator.generate(
+ new Object[] {
+ 416927583791428523L,
+ BinaryStringData.fromString("China"),
+ BinaryStringData.fromString(cityName),
+ BinaryStringData.fromString(cityName + "
West Town address 2")
+ })),
+ DataChangeEvent.insertEvent(
+ tableId,
+ generator.generate(
+ new Object[] {
+ 417022095255614379L,
+ BinaryStringData.fromString("China"),
+ BinaryStringData.fromString(cityName),
+ BinaryStringData.fromString(cityName + "
West Town address 3")
+ })));
+ }
+
+ private String triggerSavepointWithRetry(JobClient jobClient, String
savepointDirectory)
+ throws Exception {
+ int retryCount = 0;
+ final int maxRetries = 600;
+ while (retryCount < maxRetries) {
+ try {
+ return jobClient
+ .triggerSavepoint(savepointDirectory,
SavepointFormatType.DEFAULT)
+ .get();
+ } catch (Exception e) {
+ retryCount++;
+ if (retryCount >= maxRetries) {
+ throw e;
+ }
+ Thread.sleep(100);
+ }
+ }
+ throw new Exception("Failed to trigger savepoint after " + maxRetries
+ " retries");
+ }
+
+ private void initialAddressTables(PostgresConnection connection,
List<String> addressTables)
+ throws SQLException {
+ try {
+ connection.setAutoCommit(false);
+ for (String tableName : addressTables) {
+ String tableId =
+ customDatabase.getDatabaseName() + '.' + SCHEMA_NAME +
'.' + tableName;
+ String cityName = tableName.split("_")[1];
+ connection.execute(
+ "CREATE TABLE "
+ + tableId
+ + "("
+ + " id BIGINT NOT NULL PRIMARY KEY,"
+ + " country VARCHAR(255) NOT NULL,"
+ + " city VARCHAR(255) NOT NULL,"
+ + " detail_address VARCHAR(1024)"
+ + ");");
+ connection.execute(
+ format(
+ "INSERT INTO %s "
+ + "VALUES (416874195632735147,
'China', '%s', '%s West Town address 1'),"
+ + " (416927583791428523,
'China', '%s', '%s West Town address 2'),"
+ + " (417022095255614379,
'China', '%s', '%s West Town address 3');",
+ tableId, cityName, cityName, cityName,
cityName, cityName,
+ cityName));
+ connection.execute(format("ALTER TABLE %s REPLICA IDENTITY
FULL", tableId));
+ }
+ connection.commit();
+ } finally {
+ connection.close();
+ }
+ }
+
+ private FlinkSourceProvider getFlinkSourceProvider(List<String> tables,
int parallelism) {
+ List<String> fullTableNames =
+ tables.stream()
+ .map(
+ table ->
+ customDatabase.getDatabaseName()
+ + "."
+ + SCHEMA_NAME
+ + "."
+ + table)
+ .collect(Collectors.toList());
+ Map<String, String> options = new HashMap<>();
+ options.put(HOSTNAME.key(), POSTGRES_CONTAINER.getHost());
+ options.put(
+ PG_PORT.key(),
String.valueOf(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT)));
+ options.put(USERNAME.key(), TEST_USER);
+ options.put(PASSWORD.key(), TEST_PASSWORD);
+ options.put(SERVER_TIME_ZONE.key(), "UTC");
+ options.put(TABLES.key(), String.join(",", fullTableNames));
+ options.put(SLOT_NAME.key(), slotName);
+ options.put(DECODING_PLUGIN_NAME.key(), "pgoutput");
+ options.put(SCAN_NEWLY_ADDED_TABLE_ENABLED.key(), "true");
+
+ Factory.Context context =
+ new FactoryHelper.DefaultContext(
+
org.apache.flink.cdc.common.configuration.Configuration.fromMap(options),
+ null,
+ this.getClass().getClassLoader());
+
+ PostgresDataSourceFactory factory = new PostgresDataSourceFactory();
+ PostgresDataSource dataSource = (PostgresDataSource)
factory.createDataSource(context);
+
+ return (FlinkSourceProvider) dataSource.getEventSourceProvider();
+ }
+
+ private <T> CollectResultIterator<T> addCollector(
+ StreamExecutionEnvironment env,
+ DataStreamSource<T> source,
+ AbstractCollectResultBuffer<T> buffer,
+ TypeSerializer<T> serializer,
+ String accumulatorName) {
+ CollectSinkOperatorFactory<T> sinkFactory =
+ new CollectSinkOperatorFactory<>(serializer, accumulatorName);
+ CollectStreamSink<T> sink = new CollectStreamSink<>(source,
sinkFactory);
+ String operatorUid = "Data stream collect sink";
+ sink.name(operatorUid).uid(operatorUid);
+ CollectSinkOperator<T> operator = (CollectSinkOperator<T>)
sinkFactory.getOperator();
+ env.addOperator(sink.getTransformation());
+ CollectResultIterator<T> iterator =
+ new CollectResultIteratorAdapter<>(
+ buffer, operatorUid, operator, accumulatorName, 0);
+ env.registerCollectIterator(iterator);
+ return iterator;
+ }
+
+ private StreamExecutionEnvironment getStreamExecutionEnvironment(
+ String finishedSavePointPath, int parallelism) {
+ Configuration configuration = new Configuration();
+ if (finishedSavePointPath != null) {
+ configuration.set(StateRecoveryOptions.SAVEPOINT_PATH,
finishedSavePointPath);
+ }
+ StreamExecutionEnvironment env =
+
StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+ env.setParallelism(parallelism);
+ env.enableCheckpointing(500L);
+ RestartStrategyUtils.configureFixedDelayRestartStrategy(env, 3, 1000L);
+ return env;
+ }
+
+ private List<Event> fetchResults(CollectResultIterator<Event> iterator,
int size)
+ throws Exception {
+ List<Event> result = new ArrayList<>(size);
+ while (size > 0 && iterator.hasNext()) {
+ Event event = iterator.next();
+ if (event instanceof CreateTableEvent || event instanceof
DataChangeEvent) {
+ result.add(event);
+ size--;
+ }
+ }
+ return result;
+ }
+}
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 142b71298..fb46b6d70 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
@@ -22,9 +22,11 @@ import
org.apache.flink.cdc.connectors.postgres.PostgresTestBase;
import org.assertj.core.api.Assertions;
import org.testcontainers.containers.PostgreSQLContainer;
+import java.io.BufferedReader;
+import java.io.InputStream;
+import java.io.InputStreamReader;
import java.net.URL;
-import java.nio.file.Files;
-import java.nio.file.Paths;
+import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
@@ -129,21 +131,25 @@ public class UniqueDatabase {
createDatabase(databaseName);
try (Connection connection =
PostgresTestBase.getJdbcConnection(container,
databaseName);
- Statement statement = connection.createStatement()) {
+ Statement statement = connection.createStatement();
+ InputStream inputStream =
+
UniqueDatabase.class.getClassLoader().getResourceAsStream(ddlFile);
+ BufferedReader reader =
+ new BufferedReader(
+ new InputStreamReader(inputStream,
StandardCharsets.UTF_8))) {
+ final String ddl =
+ reader.lines()
+ .map(String::trim)
+ .filter(x -> !x.startsWith("--") &&
!x.isEmpty())
+ .map(
+ x -> {
+ final Matcher m =
COMMENT_PATTERN.matcher(x);
+ return m.matches() ? m.group(1) :
x;
+ })
+ .map(this::convertSQL)
+ .collect(Collectors.joining("\n"));
final List<String> statements =
- Arrays.stream(
-
Files.readAllLines(Paths.get(ddlTestFile.toURI())).stream()
- .map(String::trim)
- .filter(x ->
!x.startsWith("--") && !x.isEmpty())
- .map(
- x -> {
- final Matcher m =
-
COMMENT_PATTERN.matcher(x);
- return m.matches()
? m.group(1) : x;
- })
- .map(this::convertSQL)
-
.collect(Collectors.joining("\n"))
- .split(";"))
+ Arrays.stream(ddl.split(";"))
.map(x -> x.replace("$$", ";"))
.collect(Collectors.toList());
for (String stmt : statements) {
