You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2025/09/19 02:15:06 UTC
(flink-cdc) branch master updated: [Flink-38277] [postgres] Drop slot when startup.mode=snapshot and skip backfill.(#4113)
This is an automated email from the ASF dual-hosted git repository.
kunni 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 99427e8ae [Flink-38277] [postgres] Drop slot when startup.mode=snapshot and skip backfill.(#4113)
99427e8ae is described below
commit 99427e8ae2a9cbea00d6983694bda155a433fdef
Author: ouyangwulin <mr...@gmail.com>
AuthorDate: Fri Sep 19 10:15:01 2025 +0800
[Flink-38277] [postgres] Drop slot when startup.mode=snapshot and skip backfill.(#4113)
---
.../source/PostgresPipelineITCaseTest.java | 48 ++++++++++++++++++++++
.../postgres/source/PostgresDialect.java | 8 ++++
.../source/fetch/PostgresScanFetchTask.java | 2 +-
.../source/reader/PostgresSourceReader.java | 22 ++++++++++
.../postgres/testutils/UniqueDatabase.java | 21 ++++++++++
5 files changed, 100 insertions(+), 1 deletion(-)
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/PostgresPipelineITCaseTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresPipelineITCaseTest.java
index a57e1fed5..cdaae9eaa 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresPipelineITCaseTest.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresPipelineITCaseTest.java
@@ -132,6 +132,7 @@ public class PostgresPipelineITCaseTest extends PostgresTestBase {
List<Event> actual = fetchResultsExcept(events, expectedSnapshot.size(), createTableEvent);
assertThat(actual.subList(0, expectedSnapshot.size()))
.containsExactlyInAnyOrder(expectedSnapshot.toArray(new Event[0]));
+ assertThat(inventoryDatabase.checkSlot(slotName)).isEqualTo(slotName);
}
@ParameterizedTest(name = "unboundedChunkFirst: {0}")
@@ -308,6 +309,53 @@ public class PostgresPipelineITCaseTest extends PostgresTestBase {
}
}
+ @Test
+ public void testSnapshotOnlyMode() throws Exception {
+ inventoryDatabase.createAndInitialize();
+ PostgresSourceConfigFactory configFactory =
+ (PostgresSourceConfigFactory)
+ new PostgresSourceConfigFactory()
+ .hostname(POSTGRES_CONTAINER.getHost())
+ .port(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT))
+ .username(TEST_USER)
+ .password(TEST_PASSWORD)
+ .databaseList(inventoryDatabase.getDatabaseName())
+ .tableList("inventory.products")
+ .startupOptions(StartupOptions.snapshot())
+ .skipSnapshotBackfill(false)
+ .serverTimeZone("UTC");
+ configFactory.database(inventoryDatabase.getDatabaseName());
+ configFactory.slotName(slotName);
+ configFactory.decodingPluginName("pgoutput");
+
+ FlinkSourceProvider sourceProvider =
+ (FlinkSourceProvider)
+ new PostgresDataSource(configFactory).getEventSourceProvider();
+ CloseableIterator<Event> events =
+ env.fromSource(
+ sourceProvider.getSource(),
+ WatermarkStrategy.noWatermarks(),
+ PostgresDataSourceFactory.IDENTIFIER,
+ new EventTypeInfo())
+ .executeAndCollect();
+
+ TableId tableId = TableId.tableId("inventory", "products");
+ CreateTableEvent createTableEvent = getProductsCreateTableEvent(tableId);
+
+ // generate snapshot data
+ List<Event> expectedSnapshot = getSnapshotExpected(tableId);
+
+ // In this configuration, several subtasks might emit their corresponding CreateTableEvent
+ // to downstream. Since it is not possible to predict how many CreateTableEvents should we
+ // expect, we simply filter them out from expected sets, and assert there's at least one.
+ List<Event> actual = fetchResultsExcept(events, expectedSnapshot.size(), createTableEvent);
+ assertThat(actual.subList(0, expectedSnapshot.size()))
+ .containsExactlyInAnyOrder(expectedSnapshot.toArray(new Event[0]));
+ Thread.sleep(10000);
+ assertThat(inventoryDatabase.checkSlot(slotName))
+ .isEqualTo(String.format("Replication slot \"%s\" does not exist", slotName));
+ }
+
private static <T> List<T> fetchResultsExcept(Iterator<T> iter, int size, T sideEvent) {
List<T> result = new ArrayList<>(size);
List<T> sideResults = new ArrayList<>();
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java
index 1c99f651c..474324789 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java
@@ -262,4 +262,12 @@ public class PostgresDialect implements JdbcDataSourceDialect {
public String getPluginName() {
return sourceConfig.getDbzProperties().getProperty(PLUGIN_NAME.name());
}
+
+ public boolean removeSlot(String slotName) {
+ try (PostgresConnection jdbc = (PostgresConnection) openJdbcConnection(sourceConfig)) {
+ return jdbc.dropReplicationSlot(slotName);
+ } catch (Exception e) {
+ return false;
+ }
+ }
}
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/PostgresScanFetchTask.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTask.java
index ee868d0b8..f31db75bd 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTask.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTask.java
@@ -202,7 +202,7 @@ public class PostgresScanFetchTask extends AbstractScanFetchTask {
try {
replicationConnection.close(true);
} catch (Throwable t) {
- LOG.info("here exception occurs");
+ LOG.error("Unexpected error while dropping replication slot", t);
throw new FlinkRuntimeException(t);
}
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReader.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReader.java
index 0bad047cf..ed954d77f 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReader.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReader.java
@@ -25,9 +25,11 @@ import org.apache.flink.cdc.connectors.base.source.meta.events.LatestFinishedSpl
import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitSerializer;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitState;
import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit;
import org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceReaderContext;
import org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceReaderWithCommit;
+import org.apache.flink.cdc.connectors.postgres.source.PostgresDialect;
import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig;
import org.apache.flink.cdc.connectors.postgres.source.events.OffsetCommitAckEvent;
import org.apache.flink.cdc.connectors.postgres.source.events.OffsetCommitEvent;
@@ -39,6 +41,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
+import java.util.Map;
import java.util.PriorityQueue;
import java.util.function.Supplier;
@@ -133,6 +136,25 @@ public class PostgresSourceReader extends IncrementalSourceReaderWithCommit {
}
}
+ @Override
+ protected void onSplitFinished(Map finishedSplitIds) {
+ super.onSplitFinished(finishedSplitIds);
+ for (Object splitState : finishedSplitIds.values()) {
+ SourceSplitBase sourceSplit = ((SourceSplitState) splitState).toSourceSplit();
+ if (sourceSplit.isStreamSplit()) {
+ StreamSplit streamSplit = sourceSplit.asStreamSplit();
+ if (this.sourceConfig.getStartupOptions().isSnapshotOnly()
+ && streamSplit
+ .getStartingOffset()
+ .isAtOrAfter(streamSplit.getEndingOffset())) {
+ PostgresDialect dialect = (PostgresDialect) this.dialect;
+ boolean removed = dialect.removeSlot(dialect.getSlotName());
+ LOG.info("Remove slot '{}' result is {}.", dialect.getSlotName(), removed);
+ }
+ }
+ }
+ }
+
private boolean isCommitOffset() {
// Only if scan newly added table is enable, offset commit is controlled by isCommitOffset.
return !sourceConfig.isScanNewlyAddedTableEnabled() || isCommitOffset;
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 beccaf5bc..142b71298 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
@@ -26,6 +26,8 @@ import java.net.URL;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
@@ -183,6 +185,25 @@ public class UniqueDatabase {
}
}
+ /** Drop slot from database. */
+ public String checkSlot(String slotName) {
+ String sql =
+ String.format(
+ "SELECT slot_name from pg_replication_slots where slot_name = '%s'",
+ slotName);
+ try (Connection connection = PostgresTestBase.getJdbcConnection(container, databaseName);
+ PreparedStatement statement = connection.prepareStatement(sql)) {
+ try (ResultSet rs = statement.executeQuery()) {
+ if (rs.next()) {
+ return rs.getString("slot_name");
+ }
+ return String.format("Replication slot \"%s\" does not exist", slotName);
+ }
+ } catch (Exception exception) {
+ return exception.getMessage();
+ }
+ }
+
private String convertSQL(final String sql) {
return sql.replace("$DBNAME$", schemaName);
}