[FLINK-40560][postgres] Drop the replication slot when a snapshot-only source finishes - #4523
Open
tchivs wants to merge 1 commit into
Open
[FLINK-40560][postgres] Drop the replication slot when a snapshot-only source finishes#4523tchivs wants to merge 1 commit into
tchivs wants to merge 1 commit into
Conversation
…y source finishes
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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What is the purpose of this pull request?
Fix FLINK-40560: a PostgreSQL snapshot-only source can finish successfully while leaving its replication slot behind, retaining WAL after the job is done.
The stream split starts at the minimum snapshot high watermark and stops at the maximum. Its starting offset is subsequently advanced by emitted data-change records and heartbeats, not by every record consumed from WAL. Therefore, comparing the starting and ending offsets is not a reliable test of whether a bounded split has finished.
Brief change log
close(), where it could destroy state needed for recovery.PostgresDialect.removeSlotreturns false.Verifying this change
This change added tests and can be verified as follows:
PostgresSourceReaderTest: all 8 tests pass, including both new lifecycle cases.PostgresSourceITCase#testSnapshotOnly*: all 3 tests pass, including the new bounded-job regression. The regression reads all 21 fixture rows through end-of-input and then asserts that the replication slot is absent.expected: 0L, but was: 1Lfor the remaining slot count. Restoring the fix makes it pass.mvn -pl flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc \ com.diffplug.spotless:spotless-maven-plugin:apply mvn -pl flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc \ '-Dtest=PostgresSourceReaderTest,PostgresSourceITCase#testSnapshotOnly*' clean verifyIssue ownership: I have posted a claim comment on FLINK-40560. Self-assignment returned HTTP 403, so assignment to
tchivsstill requires a committer.Documentation
Was generative AI tooling used to co-author this PR?
Generated-by: OpenAI ChatGPT via Oh My Pi