[jira] [Created] (FLINK-40560) [Postgres] snapshot-only replication slot is never dropped unless the snapshot is instantaneous

["tchivs (Jira)" <ji...@apache.org>]

tchivs created FLINK-40560:
------------------------------

             Summary: [Postgres] snapshot-only replication slot is never dropped unless the snapshot is instantaneous
                 Key: FLINK-40560
                 URL: https://issues.apache.org/jira/browse/FLINK-40560
             Project: Flink
          Issue Type: Bug
          Components: Flink CDC
    Affects Versions: cdc-3.6.0
            Reporter: tchivs


h3. Flink version / Flink CDC version

Flink 2.2.1, Flink CDC 3.6.0 (`flink-connector-postgres-cdc`), Debezium 1.9.8.Final,
PostgreSQL 18 (not version-specific), `pgoutput`.

h3. What happened

FLINK-38277 added slot cleanup for `snapshot-only` sources: when the stream split finishes,
`PostgresSourceReader.onSplitFinished` drops the replication slot. In practice the drop never
happens unless the whole snapshot is effectively instantaneous, so every bounded snapshot run
leaves its replication slot behind.

The job reaches FINISHED normally, no `Remove slot '{}' result is {}.` line is ever logged, and
the slot stays in `pg_replication_slots` with `active = false`, pinning WAL from its
`restart_lsn` until an operator drops it manually.

h3. How to reproduce

# Configure a Postgres incremental source with `scan.startup.mode=snapshot`.
# Capture enough tables/partitions that the snapshot takes more than a moment. Our case is six
  tables, four of them monthly-partitioned, expanding to ~380 snapshot splits over ~17 minutes.
  Any workload with concurrent writes elsewhere in the cluster is enough.
# Let the job run to FINISHED.
# Inspect `pg_replication_slots`: the slot is still there.

h3. Logs

The stream split completes and the reader reaches an offset past the stopping offset, yet the
drop is skipped:

{noformat}
StreamSplitReadTask finished for StreamSplit{splitId='stream-split',
  offset=Offset{lsn=LSN{2A/385B31F8}, ...},      <- startingOffset = min(high watermarks)
  endOffset=Offset{lsn=LSN{2A/3D86A000}, ...},   <- endingOffset   = max(high watermarks)
  isSnapshotCompleted=true}
at Offset{lsn=LSN{2A/3D8E96A8}, ...}
{noformat}

No `Remove slot` line follows. On an otherwise idle development database, a single ~17 minute run
retained 85-135 MB of WAL, and the slot keeps accumulating until it is dropped by hand.

h3. Cause

`PostgresSourceReader.onSplitFinished` guards the drop with an offset comparison:

{code: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);
}
{code}

But `HybridSplitAssigner.createStreamSplit()` fills those two offsets with the *lowest* and the
*highest* high-watermark observed across the finished snapshot splits:

{code: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,
        ...);
{code}

So the guard reduces to `min(high watermarks) >= max(high watermarks)`, which can only hold when
every snapshot split carries the exact same high watermark — that is, when no WAL at all is
produced between the first and the last split finishing. With more than one split, or any
concurrent write activity in the cluster, the condition is false by construction and the cleanup
never runs. This is also why it passes in tests over small tables.

h3. Expected

In `snapshot-only` mode the slot should be dropped once the bounded stream split has finished.

h3. Suggested fix

Key the drop off "the bounded stream split finished" instead of
`startingOffset.isAtOrAfter(endingOffset)`. `onSplitFinished` is only invoked for a split that
already completed, so for `isSnapshotOnly()` the offset comparison adds no safety — it only
suppresses the cleanup in every non-trivial case.

h3. Workaround

Registering a `JobStatusHook` that drops the slot on FINISHED / FAILED / CANCELED works. One
caveat for anyone doing the same: a `Throwable` escaping a `JobStatusHook` is routed to Flink's
`FatalExitExceptionHandler`, which terminates the JobManager process, so the hook has to contain
every `Throwable` itself. Its JDBC class graph also needs to be warmed up before the terminal
callback, because the user class loader is being torn down by then.

h3. Related

* FLINK-38277 — introduced the cleanup this issue reports as ineffective (fixed in cdc-3.5.0).
* FLINK-40538 — separate bug in the same area; on a publication with no traffic the stream split
  cannot reach its stopping offset at all.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)