Skip to content

[FLINK-40560][postgres] Drop the replication slot when a snapshot-only source finishes #4523

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tchivs wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[FLINK-40560][postgres] Drop the replication slot when a snapshot-only source finishes #4523

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 35 additions & 6 deletions ...ain/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,13 @@ public class PostgresSourceReader extends IncrementalSourceReaderWithCommit {
private final PriorityQueue<Long> 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,
Expand All @@ -86,6 +93,7 @@ public PostgresSourceReader(
this.lsnCommitCheckpointsDelay =
((PostgresSourceConfig) sourceConfig).getLsnCommitCheckpointsDelay();
this.minHeap = new PriorityQueue<>();
this.incrementalSourceReaderContext = incrementalSourceReaderContext;
}

@Override
Expand Down Expand Up @@ -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);
}
}
}
}
Expand Down
64 changes: 64 additions & 0 deletions ...c/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<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
Expand Down Expand Up @@ -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(
Expand Down
95 changes: 95 additions & 0 deletions ...java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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.
*
* <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 {
Expand Down