eskabetxe opened a new pull request, #4521:
URL: https://github.com/apache/flink-cdc/pull/4521
**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
--
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]
