[
https://issues.apache.org/jira/browse/FLINK-40538?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40538:
-----------------------------------
Labels: pull-request-available (was: )
> [Postgres] Connector never starts streaming on a database with no publication
> traffic
> -------------------------------------------------------------------------------------
>
> Key: FLINK-40538
> URL: https://issues.apache.org/jira/browse/FLINK-40538
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Affects Versions: cdc-3.6.0
> Reporter: João Boto
> Priority: Major
> Labels: pull-request-available
> Fix For: cdc-3.6.0
>
>
> 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).
> h3. What happened
> A Postgres CDC source pointed at a database that receives no writes to its
> publication
> never leaves Debezium's WAL position search. It never begins streaming, and:
> - no change events are ever produced (including once writes eventually start,
> until the
> first of them arrives);
> - `heartbeat.action.query` never executes, 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's position for as long as the job runs.
> The last point is the damaging one: the job reports RUNNING and healthy while
> WAL grows
> without bound on the source database.
> h3. How to reproduce
> 1. Create a Postgres CDC source over a schema whose tables receive no writes.
> 2. Start the job. Other databases in the cluster may be active; the captured
> publication must be quiet.
> 3. Observe the job is RUNNING, all checkpoints complete, and `numRecordsOut`
> of the
> source is 0.
> 4. Observe `confirmed_flush_lsn` in `pg_replication_slots` stays fixed while
> `pg_current_wal_lsn()` advances.
> 5. Write any row to a captured table — the connector immediately starts
> streaming and
> the slot begins advancing.
> h3. Logs
> At DEBUG for `io.debezium`, TRACE for
> `io.debezium.connector.postgresql.connection`:
> INFO WalPositionLocator - Looking for WAL restart position for last commit
> LSN 'null'
> and last change LSN 'LSN\{0/518EB00}'
> INFO PostgresStreamingChangeEventSource - Searching for WAL resume position
> TRACE PostgresReplicationConnection - Streaming requested from LSN
> LSN\{0/518EB00}, received LSN LSN\{0/5192A60}
> TRACE PostgresReplicationConnection - Streaming requested from LSN
> LSN\{0/518EB00}, received LSN LSN\{0/5192AF0}
> ... repeats indefinitely, requested LSN never changes ...
> `"WAL resume position '{}' discovered"` is never logged.
> h3. Cause
> `PostgresSourceFetchTaskContext.loadStartingOffsetState` always returns a
> non-null
> `PostgresOffsetContext`, built from the stream split's starting offset. In
> `PostgresStreamingChangeEventSource#execute` that makes
> ```java
> boolean hasStartLsnStoredInContext = offsetContext != null;
> always true, so the WAL position search branch is taken unconditionally. Plain
> Debezium takes the other branch on a fresh start and logs
> "WAL position will not be searched".
> searchWalPosition then loops until a message is decoded, and — unlike the main
> streaming loop — dispatches no heartbeat while waiting:
> // searchWalPosition
> if (receivedMessage) \{ noMessageIterations = 0; }
> else {
> noMessageIterations++; // no dispatchHeartbeatEvent
> ...
> }
> // main streaming loop
> else {
> if (offsetContext.hasCompletelyProcessedPosition()) {
> dispatcher.dispatchHeartbeatEvent(partition, offsetContext);
> }
> ...
> }
> That is a deadlock: the search waits for publication traffic, and the only
> mechanism
> that would generate traffic on an idle database — heartbeat.action.query,
> driven by
> the heartbeat dispatch — runs only from the main loop, which the search
> precedes.
> Possible fixes
> 1. Do not force the search when the stored offset has nothing processed yet,
> so
> Debezium takes the new WalPositionLocator() path and starts from the slot's
> confirmed position.
> 2. Write one WAL record when the stream task starts, so the search terminates
> deterministically. We are running this as a local patch: a single
> pg_logical_emit_message(false, '<internal prefix>', ...) in
> PostgresStreamFetchTask#execute before delegating to
> StreamSplitReadTask#execute. The prefix matches no configured logical
> message, so
> the record is filtered before the sink.
> 3. Upstream Debezium: dispatch heartbeats in searchWalPosition as the main
> loop does.
> This would fix it without a change here, but only for versions carrying
> that fix.
> h3. Are you willing to submit a PR?
> Yes.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
