[FLINK-40538][postgres] Skip WAL position search on idle publication …

[eskabetxe]

What is the purpose of this pull request?

Fixes FLINK-40538: a Postgres CDC source pointed at a database whose captured publication receives no writes never leaves Debezium's WAL-position search, so it never starts streaming.

While stuck in the search, the job reports RUNNING and healthy and all checkpoints complete, but:

• no change events are produced (and none will be, even once writes eventually start, until the first write arrives);
• heartbeat.action.query never runs, so heartbeat.interval.ms has no effect;
• the replication slot's confirmed_flush_lsn never advances, so PostgreSQL retains every WAL segment from the slot position for the life of the job — WAL grows without bound on the source database.

Root cause. PostgresSourceFetchTaskContext.loadStartingOffsetState always returns a non-null PostgresOffsetContext (built from the stream split's starting offset), so in the forked PostgresStreamingChangeEventSource#execute the WAL-search branch is entered unconditionally (walPosition.searchingEnabled() is always true). searchWalPosition then loops until a message is decoded but — unlike the main streaming loop — dispatches no heartbeat while waiting. On an idle publication that is a deadlock: the search waits for publication traffic, and the only thing that would generate traffic (heartbeat.action.query, driven from the main loop) runs only after the search returns.

Fix. Guard the search with offsetContext.hasCompletelyProcessedPosition(), mirroring the fix Debezium shipped in 2.7 (searchingEnabled() && effectiveOffset.hasCompletelyProcessedPosition()). On a fresh start nothing has been completely processed, so the search is skipped; streaming still begins from the stored LSN via startStreaming(lsn, walPosition), so no events are missed. On a resumed offset (e.g. after a checkpoint) the search still runs, preserving exact-resume behavior.

Brief change log

• PostgresStreamingChangeEventSource#execute (forked copy under io.debezium.connector.postgresql): change the WAL-search condition from if (walPosition.searchingEnabled()) to if (walPosition.searchingEnabled() && offsetContext.hasCompletelyProcessedPosition()), with a comment explaining the idle-publication deadlock and the Debezium 2.7 parallel.
• Add PostgresStreamingChangeEventSourceTest pinning the decision boundary for both a fresh-start offset (search skipped) and a resumed offset (search runs).

---

Verifying this change

This change added tests and can be verified as follows:

• Added unit tests in flink-connector-postgres-cdc → io.debezium.connector.postgresql.PostgresStrea a fresh-start offset yields searchingEnabled()== true but hasCompletelyProcessedPosition() == false (so the search is now correctly skipped), while a resumed offset
  yields both true (search still runs).
• Manually reproduced against a Postgres instance with a quiet captured publication: before the fix the job stays RUNNING with numRecordsOut == 0 and a fixed confirmed_flush_lsn while pg_current_wal_lsn() advances; after the fix the connector
  streams immediately and the slot advances.

Documentation

• Does this pull request introduce a new feature
• If yes, how is the feature documented? not applicable

[loserwang1024]

Skipping the WAL-position search does not appear to resolve the idle-publication issue, because processMessages() still guards heartbeat dispatch with the same condition:

if (offsetContext.hasCompletelyProcessedPosition()) {
    dispatcher.dispatchHeartbeatEvent(partition, offsetContext);
}

On a fresh start, the offset contains a starting LSN but no lsn_proc, so hasCompletelyProcessedPosition() returns false. Even after this change allows execution to reach the main streaming loop, heartbeat dispatch is still skipped.

This creates a circular dependency: processing a message is required to enable heartbeats, but on an idle publication, heartbeat.action.query may be the only mechanism that would generate that message. The query is invoked through heartbeat dispatch; it does not run independently on a timer.

As a result, this change moves the wait from the WAL-position search into the main streaming loop, while the heartbeat query remains blocked until an external write produces a message that advances the processed position.

[eskabetxe]

Good catch, you're right — the 2.7 search guard only works because upstream had already removed the hasCompletelyProcessedPosition() guard around heartbeat dispatch in 2.4 (DBZ-6635, "Send heartbeats also before processing first event"). Since this fork is based on Debezium 1.9.8, that guard was still in place, so skipping the search would indeed just have moved the stall into processMessages().

Pushed a follow-up that completes the backport:

• processMessages() now dispatches heartbeats unconditionally when no message is received (DBZ-6635), so heartbeat.action.query can generate WAL on a fresh start;
• searchWalPosition() also dispatches heartbeats while waiting, mirroring 2.7, so the search terminates on an idle publication when resuming.

Added a unit test that runs execute() with a fresh-start offset and a stream that never yields a message, asserting heartbeat dispatch happens (and the search is skipped); it fails without the fix. PTAL, thanks!