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 {