tchivs opened a new issue, #4522:
URL: https://github.com/apache/flink-cdc/issues/4522
## Description
After #4113, a `snapshot-only` Postgres source is supposed to drop its
replication slot in
`PostgresSourceReader.onSplitFinished`. In practice the drop never happens
unless the whole
snapshot is effectively instantaneous, so every bounded snapshot run leaks a
replication slot
that keeps pinning WAL until an operator removes it by hand.
## Root cause
`PostgresSourceReader.onSplitFinished` guards the drop with
(`flink-connector-postgres-cdc`, release-3.6.0, lines 161-168):
```java
if (this.sourceConfig.getStartupOptions().isSnapshotOnly()
&&
streamSplit.getStartingOffset().isAtOrAfter(streamSplit.getEndingOffset())) {
boolean removed = dialect.removeSlot(dialect.getSlotName());
LOG.info("Remove slot '{}' result is {}.", dialect.getSlotName(),
removed);
}
```
But `HybridSplitAssigner.createStreamSplit()` builds that split as:
```java
// minOffset / maxOffset are the lowest / highest high-watermark among
finished snapshot splits
Offset stoppingOffset = offsetFactory.createNoStoppingOffset();
if (sourceConfig.getStartupOptions().isSnapshotOnly()) {
stoppingOffset = maxOffset;
}
return new StreamSplit(
STREAM_SPLIT_ID,
minOffset == null ? offsetFactory.createInitialOffset() : minOffset,
stoppingOffset,
...);
```
So `startingOffset == min(high watermarks)` and `endingOffset == max(high
watermarks)`.
The guard therefore reduces to `min >= max`, which can only hold when every
snapshot split
carries the exact same high watermark — i.e. when no WAL is produced between
the first and the
last split finishing. With more than one split and any concurrent write
activity, the condition
is false by construction and the slot is never dropped.
## How to reproduce
1. Configure a Postgres incremental source with `scan.startup.mode=snapshot`.
2. Capture enough tables/partitions that the snapshot takes more than a few
seconds
(any workload with concurrent writes elsewhere in the cluster is enough).
3. Let the job run to `FINISHED`, then inspect `pg_replication_slots`.
## Observed
Job reaches `FINISHED` normally, the stream split completes, and no `Remove
slot` line is
logged. The offsets in our run show the guard cannot hold:
```
StreamSplitReadTask finished for StreamSplit{splitId='stream-split',
offset=Offset{lsn=LSN{2A/385B31F8}, ...}, <- startingOffset (min
watermark)
endOffset=Offset{lsn=LSN{2A/3D86A000}, ...}, <- endingOffset (max
watermark)
isSnapshotCompleted=true}
at Offset{lsn=LSN{2A/3D8E96A8}, ...}
```
`pg_replication_slots` still lists the slot as `active = false` afterwards.
On our (idle)
development database a single ~17 minute run retained 85-135 MB of WAL; the
slot keeps
accumulating until it is dropped manually.
## Expected
In `snapshot-only` mode the slot should be dropped once the stream split has
finished, since
the split is bounded and the reader has already reached its stopping offset.
## Suggested fix
Key the drop off "the bounded stream split finished" rather than
`startingOffset.isAtOrAfter(endingOffset)`. `onSplitFinished` is only
invoked for a split that
has completed, so for `isSnapshotOnly()` the extra offset comparison does
not add safety — it
only suppresses the cleanup in every non-trivial case.
## Environment
- Flink CDC 3.6.0 (`flink-connector-postgres-cdc`)
- Flink 2.2.1
- PostgreSQL 18.4, `pgoutput`
## Workaround
We register a `JobStatusHook` that drops the slot on `FINISHED` / `FAILED` /
`CANCELED`.
Note for anyone doing the same: any `Throwable` escaping a `JobStatusHook`
is routed to
Flink's `FatalExitExceptionHandler` and kills the JobManager process, so
such a hook has to
contain every `Throwable` itself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
