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 efb1d3a00 [FLINK-40560][postgres] Drop the replication slot when a
snapshot-only source finishes (#4523)
efb1d3a00 is described below
commit efb1d3a000dffda3d5fd5970764fc34221564751
Author: Lanny Boarts <[email protected]>
AuthorDate: Fri Oct 16 12:18:34 2026 +0800
[FLINK-40560][postgres] Drop the replication slot when a snapshot-only
source finishes (#4523)
---
.../source/reader/IncrementalSourceReader.java | 2 +-
.../source/reader/PostgresSourceReader.java | 33 ++++++--
.../postgres/source/PostgresSourceITCase.java | 64 +++++++++++++++
.../source/reader/PostgresSourceReaderTest.java | 95 ++++++++++++++++++++++
4 files changed, 187 insertions(+), 7 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java
index 3f7da9525..b629d1fd6 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java
@@ -119,7 +119,7 @@ public class IncrementalSourceReader<T, C extends
SourceConfig>
protected final C sourceConfig;
protected final DataSourceDialect<C> dialect;
- private final IncrementalSourceReaderContext
incrementalSourceReaderContext;
+ protected final IncrementalSourceReaderContext
incrementalSourceReaderContext;
public IncrementalSourceReader(
FutureCompletingBlockingQueue<RecordsWithSplitIds<SourceRecords>>
elementQueue,
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 30f77ed25..12b47f4d2 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
@@ -157,14 +157,35 @@ public class PostgresSourceReader extends
IncrementalSourceReaderWithCommit {
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 95c15d775..dd896f087 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 org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.sql.SQLException;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -455,6 +456,52 @@ class PostgresSourceITCase extends PostgresTestBase {
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<RowData> source =
+
PostgresSourceBuilder.PostgresIncrementalSource.<RowData>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<RowData> 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
@@ -1504,6 +1551,23 @@ class PostgresSourceITCase extends PostgresTestBase {
});
}
+ 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 10ba00328..7ef016b97 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.common.typeinfo.TypeInformation;
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.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -191,6 +197,95 @@ public class PostgresSourceReaderTest extends
PostgresTestBase {
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.
+ *
+ * <p>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<String, SourceSplitState> 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<String, SourceSplitState> finished = new HashMap<>();
+ finished.put(split.splitId(), new StreamSplitState(split));
+ return finished;
+ }
+
+ private static Offset lsnOffset(long lsn) {
+ final Map<String, String> 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 {
