[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