From 47ff606d5f30bcf7d3906200cff5f40b5fb5f347 Mon Sep 17 00:00:00 2001 From: tchivs Date: Sat, 5 Sep 2026 11:21:41 +0800 Subject: [PATCH] [FLINK-40560][postgres] Drop the replication slot when a snapshot-only source finishes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FLINK-38277 made a snapshot-only Postgres source drop its replication slot once the stream split finishes, but gated it on streamSplit.getStartingOffset().isAtOrAfter(streamSplit.getEndingOffset()) which does not answer whether the split finished. HybridSplitAssigner#createStreamSplit seeds those two offsets with the lowest and highest high watermark of the finished snapshot splits, so they already differ whenever there is more than one split or any concurrent write. Afterwards only IncrementalSourceRecordEmitter#updateStreamSplitState advances the starting offset, and only for data-change records and heartbeats. A split that reaches its ending offset without emitting such a record — a captured publication with no traffic emits none — therefore left its slot behind on every run, pinning WAL until an operator dropped it by hand. IncrementalSourceReader#onSplitFinished already documents that a stream split finishes for exactly two reasons: the enumerator suspended it so newly added tables can be snapshotted, or it reached its ending offset. Only the second means the bounded read is over, and the reader context already distinguishes them, so gate on !isStreamSplitReaderSuspended() instead. That also removes the offset dereference and with it a latent NPE when the stopping offset is null. PostgresDialect#removeSlot swallows failures and returns false, so a failed cleanup was only visible as an INFO line reading "false". Log it at WARN with the manual pg_drop_replication_slot hint instead. The reader keeps its own reference to IncrementalSourceReaderContext because the base class holds its copy privately; that avoids widening the shared base class API. Two tests in PostgresSourceReaderTest cover both finish reasons: a split whose starting offset is behind its ending offset must release the slot, and a split suspended via StreamSplitUpdateRequestEvent must keep it, since resuming needs the slot's position. Reverting the guard turns the first one red. --- .../source/reader/PostgresSourceReader.java | 41 ++++++-- .../postgres/source/PostgresSourceITCase.java | 64 +++++++++++++ .../reader/PostgresSourceReaderTest.java | 95 +++++++++++++++++++ 3 files changed, 194 insertions(+), 6 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/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 30f77ed2548..5f0bb040572 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 @@ -65,6 +65,13 @@ public class PostgresSourceReader extends IncrementalSourceReaderWithCommit { private final PriorityQueue minHeap; private final int lsnCommitCheckpointsDelay; + /** + * Kept locally because {@link IncrementalSourceReader} holds its copy privately, and {@link + * #onSplitFinished} needs it to tell a suspended stream split apart from one that reached its + * ending offset. + */ + private final IncrementalSourceReaderContext incrementalSourceReaderContext; + public PostgresSourceReader( FutureCompletingBlockingQueue elementQueue, Supplier supplier, @@ -86,6 +93,7 @@ public PostgresSourceReader( this.lsnCommitCheckpointsDelay = ((PostgresSourceConfig) sourceConfig).getLsnCommitCheckpointsDelay(); this.minHeap = new PriorityQueue<>(); + this.incrementalSourceReaderContext = incrementalSourceReaderContext; } @Override @@ -157,14 +165,35 @@ protected void onSplitFinished(Map finishedSplitIds) { for (Object splitState : finishedSplitIds.values()) { SourceSplitBase sourceSplit = ((SourceSplitState) splitState).toSourceSplit(); if (sourceSplit.isStreamSplit()) { - StreamSplit streamSplit = sourceSplit.asStreamSplit(); + // A stream split finishes for one of two reasons, see + // IncrementalSourceReader#onSplitFinished: it was suspended by the enumerator so + // that newly added tables can be snapshotted, or it reached its ending offset. + // Only the latter means the bounded snapshot-only read is over and the slot is no + // longer needed. + // + // Comparing the split's starting and ending offsets cannot answer that question. + // For snapshot-only, HybridSplitAssigner#createStreamSplit seeds them with the + // lowest and highest high watermark of the finished snapshot splits, and afterwards + // only IncrementalSourceRecordEmitter#updateStreamSplitState advances the starting + // offset, for data-change records and heartbeats. A split that reaches its ending + // offset without emitting such a record — a captured publication with no traffic + // emits none — therefore left the slot behind on every run. if (this.sourceConfig.getStartupOptions().isSnapshotOnly() - && streamSplit - .getStartingOffset() - .isAtOrAfter(streamSplit.getEndingOffset())) { + && !incrementalSourceReaderContext.isStreamSplitReaderSuspended()) { PostgresDialect dialect = (PostgresDialect) this.dialect; - boolean removed = dialect.removeSlot(dialect.getSlotName()); - LOG.info("Remove slot '{}' result is {}.", dialect.getSlotName(), removed); + String slotName = dialect.getSlotName(); + boolean removed = dialect.removeSlot(slotName); + if (removed) { + LOG.info("Removed replication slot '{}'.", slotName); + } else { + // removeSlot swallows the failure, so without this the only trace of a + // slot that keeps pinning WAL would be an INFO line reading "false". + LOG.warn( + "Failed to remove replication slot '{}'. It will keep retaining WAL " + + "until it is dropped, e.g. with pg_drop_replication_slot('{}').", + slotName, + slotName); + } } } } 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 c4d3d91f995..d4f50c720c7 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 @@ -60,6 +60,7 @@ import java.io.IOException; import java.sql.SQLException; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -381,6 +382,52 @@ void testSnapshotOnlyModeWithDMLPostHighWaterMark() throws Exception { assertEqualsInAnyOrder(expectedRecords, records); } + @Test + void testSnapshotOnlyModeReleasesReplicationSlot() throws Exception { + ResolvedSchema customersSchema = + new ResolvedSchema( + Arrays.asList( + physical("Id", BIGINT().notNull()), + physical("Name", STRING()), + physical("address", STRING()), + physical("phone_number", STRING())), + new ArrayList<>(), + UniqueConstraint.primaryKey("pk", Collections.singletonList("id"))); + TestTable table = new TestTable(customersSchema); + PostgresSourceBuilder.PostgresIncrementalSource source = + PostgresSourceBuilder.PostgresIncrementalSource.builder() + .hostname(customDatabase.getHost()) + .port(customDatabase.getDatabasePort()) + .username(customDatabase.getUsername()) + .password(customDatabase.getPassword()) + .database(customDatabase.getDatabaseName()) + .decodingPluginName("pgoutput") + .slotName(slotName) + .tableList(new TestTableId("customer", "Customers").toString()) + .startupOptions(StartupOptions.snapshot()) + // Multiple splits have different high watermarks. No captured DML or + // heartbeat advances the stream split's starting offset to its end. + .splitSize(2) + .heartbeatInterval(Duration.ZERO) + .deserializer(table.getDeserializer()) + .build(); + + try (StreamExecutionEnvironment env = + StreamExecutionEnvironment.getExecutionEnvironment()) { + env.enableCheckpointing(1000); + env.setParallelism(1); + try (CloseableIterator iterator = + env.fromSource(source, WatermarkStrategy.noWatermarks(), "Snapshot Only Source") + .executeAndCollect()) { + // Request more than the 21 rows to wait for the bounded source to finish. + assertThat(fetchRowData(iterator, 22, table::stringify)).hasSize(21); + } + try (PostgresConnection connection = getConnection()) { + assertThat(getSlotCount(connection)).isZero(); + } + } + } + @Test void testSnapshotOnlyModeWithDMLPreHighWaterMark() throws Exception { // The data num is 21, set fetchSize = 22 to test the job is bounded @@ -1363,6 +1410,23 @@ private static long getCountOfTable(JdbcConnection jdbc, TableId tableId) throws }); } + private long getSlotCount(JdbcConnection jdbc) throws SQLException { + final String query = + String.format( + "SELECT count(*) FROM pg_replication_slots WHERE slot_name = '%s'", + slotName); + return jdbc.queryAndMap( + query, + rs -> { + if (!rs.next()) { + throw new SQLException( + String.format( + "No result returned after running query [%s]", query)); + } + return rs.getLong(1); + }); + } + private String getConfirmedFlushLsn(JdbcConnection jdbc) throws SQLException { final String query = String.format( 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 c8ead39f86d..c6520cf160e 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 @@ -22,15 +22,20 @@ import org.apache.flink.api.connector.source.ReaderOutput; import org.apache.flink.api.connector.source.SourceOutput; import org.apache.flink.cdc.connectors.base.options.StartupOptions; +import org.apache.flink.cdc.connectors.base.source.meta.events.StreamSplitUpdateRequestEvent; +import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset; 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.SourceSplitState; import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit; +import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplitState; import org.apache.flink.cdc.connectors.postgres.PostgresTestBase; import org.apache.flink.cdc.connectors.postgres.source.MockPostgresDialect; import org.apache.flink.cdc.connectors.postgres.source.PostgresDialect; import org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder; import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig; import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory; +import org.apache.flink.cdc.connectors.postgres.source.offset.PostgresOffset; import org.apache.flink.cdc.connectors.postgres.source.offset.PostgresOffsetFactory; import org.apache.flink.cdc.connectors.postgres.testutils.RecordsFormatter; import org.apache.flink.cdc.connectors.postgres.testutils.UniqueDatabase; @@ -58,6 +63,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; @@ -189,6 +195,95 @@ void testNotifyCheckpointWindowSizeDefault() throws Exception { assertThat(completedCheckpointIds).containsExactly(101L); } + /** + * A snapshot-only stream split that reached its ending offset must release the replication slot + * even though its starting offset is behind that ending offset. + * + *

That combination is the normal case rather than an edge case: {@code + * HybridSplitAssigner#createStreamSplit} seeds the two offsets with the lowest and highest high + * watermark of the finished snapshot splits, so they differ as soon as there is more than one + * split or any concurrent write, and only emitted data-change or heartbeat records move the + * starting offset afterwards. + */ + @Test + void testSnapshotOnlyRemovesSlotWhenStartingOffsetIsBehindEndingOffset() throws Exception { + final PostgresSourceReader reader = createSnapshotOnlyReader(); + createSlot(); + assertThat(customDatabase.checkSlot(slotName)).isEqualTo(slotName); + + reader.onSplitFinished(finishedStreamSplit()); + + // checkSlot returns a human-readable "does not exist" sentence that itself embeds the slot + // name, so compare for equality rather than containment. + assertThat(customDatabase.checkSlot(slotName)).isNotEqualTo(slotName); + } + + /** + * A suspended stream split is the other reason {@code onSplitFinished} fires — the enumerator + * paused streaming so newly added tables can be snapshotted, and the split will resume. The + * slot has to survive that, otherwise resuming loses its position. + */ + @Test + void testSnapshotOnlyKeepsSlotWhenStreamSplitIsMerelySuspended() throws Exception { + final PostgresSourceReader reader = createSnapshotOnlyReader(); + createSlot(); + // StreamSplitUpdateRequestEvent is what the enumerator sends to pause streaming; note the + // reader context flag is the authority here, not StreamSplit#isSuspended. + reader.handleSourceEvents(new StreamSplitUpdateRequestEvent()); + + reader.onSplitFinished(finishedStreamSplit()); + + assertThat(customDatabase.checkSlot(slotName)).isEqualTo(slotName); + } + + private Map finishedStreamSplit() { + // Deliberately startingOffset < endingOffset: that is what the assigner produces for a + // multi-split snapshot, and what the previous guard treated as "not finished". + final Offset startingOffset = lsnOffset(1000L); + final Offset endingOffset = lsnOffset(2000L); + final StreamSplit split = + new StreamSplit( + StreamSplit.STREAM_SPLIT_ID, + startingOffset, + endingOffset, + new ArrayList<>(), + new HashMap<>(), + 0, + false, + true); + final Map finished = new HashMap<>(); + finished.put(split.splitId(), new StreamSplitState(split)); + return finished; + } + + private static Offset lsnOffset(long lsn) { + final Map offset = new HashMap<>(); + offset.put("lsn", Long.toString(lsn)); + return PostgresOffset.of(offset); + } + + private void createSlot() throws Exception { + try (Connection connection = + getJdbcConnection(POSTGRES_CONTAINER, customDatabase.getDatabaseName()); + Statement statement = connection.createStatement()) { + statement.execute( + String.format( + "SELECT pg_create_logical_replication_slot('%s', 'pgoutput')", + slotName)); + } + } + + private PostgresSourceReader createSnapshotOnlyReader() throws Exception { + final PostgresOffsetFactory offsetFactory = new PostgresOffsetFactory(); + final PostgresSourceConfigFactory configFactory = createConfigFactory(); + configFactory.startupOptions(StartupOptions.snapshot()); + PostgresDialect dialect = new PostgresDialect(configFactory.create(0)); + final PostgresSourceBuilder.PostgresIncrementalSource source = + new PostgresSourceBuilder.PostgresIncrementalSource<>( + configFactory, new ForwardDeserializeSchema(), offsetFactory, dialect); + return source.createReader(new TestingReaderContext()); + } + @ParameterizedTest @ValueSource(booleans = {true, false}) void testMultipleSnapshotSplit(boolean skipBackFill) throws Exception {