From bde1eb1491e900f61d5e6fa5950179ee1ed1c9ed Mon Sep 17 00:00:00 2001 From: Joao Boto Date: Wed, 2 Sep 2026 14:29:25 +0200 Subject: [PATCH] [FLINK-40538][postgres] Skip WAL position search on idle publication for a fresh start PostgresSourceFetchTaskContext.loadStartingOffsetState always returns a non-null PostgresOffsetContext built from the stream split's starting offset, so in the forked PostgresStreamingChangeEventSource#execute the WAL position search branch was taken unconditionally. On an idle publication that search loops forever waiting for a decoded message: no WAL is produced, and the heartbeat action query that would generate some only runs from the main streaming loop, which the search precedes. The job stays RUNNING while the slot's confirmed_flush_lsn never advances and WAL grows without bound. Guard the search with offsetContext.hasCompletelyProcessedPosition(), mirroring the fix Debezium shipped in 2.7. On a fresh start (nothing processed yet) the search is skipped; streaming still starts from the stored LSN, so no events are missed. On a resumed offset the search still runs. Co-Authored-By: Claude Opus 4.8 --- .../PostgresStreamingChangeEventSource.java | 10 +- ...ostgresStreamingChangeEventSourceTest.java | 102 ++++++++++++++++++ 2 files changed, 111 insertions(+), 1 deletion(-) create mode 100644 flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSourceTest.java diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSource.java index e309a3b29a4..2ccf610a7b4 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSource.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSource.java @@ -187,7 +187,15 @@ public void execute( this.lastCompletelyProcessedLsn = replicationStream.get().startLsn(); - if (walPosition.searchingEnabled()) { + // Only search for the WAL resume position when the stored offset has actually + // processed a position. On a fresh start (nothing processed yet) the search loop + // would block forever waiting for a decoded message: on an idle publication no WAL + // is produced, and the heartbeat action query that would generate some only runs + // from the main streaming loop, which this search precedes. Skipping the search here + // still starts streaming from the stored LSN, so no events are missed. This mirrors + // the fix Debezium shipped in 2.7, which added the hasCompletelyProcessedPosition() + // guard to searchingEnabled(). + if (walPosition.searchingEnabled() && offsetContext.hasCompletelyProcessedPosition()) { searchWalPosition(context, stream, walPosition); try { if (!isInPreSnapshotCatchUpStreaming(offsetContext)) { diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSourceTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSourceTest.java new file mode 100644 index 00000000000..993e199ec0d --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSourceTest.java @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.debezium.connector.postgresql; + +import org.apache.flink.cdc.connectors.postgres.testutils.TestHelper; + +import io.debezium.connector.postgresql.connection.Lsn; +import io.debezium.connector.postgresql.connection.WalPositionLocator; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit test for the WAL-position-search guard in {@link PostgresStreamingChangeEventSource}. + * + *

On an idle publication the search loop blocks forever, because it waits for a decoded WAL + * message while the only mechanism that would produce one on a quiet database (the heartbeat action + * query) runs from the main streaming loop that the search precedes. The fix — mirroring Debezium + * 2.7 — only enters the search when the stored offset has actually processed a position, i.e. when + * {@code searchingEnabled() && offsetContext.hasCompletelyProcessedPosition()}. This test pins that + * decision boundary for both a fresh start and a resumed offset. + */ +class PostgresStreamingChangeEventSourceTest { + + private PostgresConnectorConfig connectorConfig; + private PostgresOffsetContext.Loader offsetLoader; + + @BeforeEach + public void beforeEach() { + this.connectorConfig = new PostgresConnectorConfig(TestHelper.defaultConfig().build()); + this.offsetLoader = new PostgresOffsetContext.Loader(this.connectorConfig); + } + + /** + * Builds the {@link WalPositionLocator} the same way {@code execute} does for a stored offset. + */ + private static WalPositionLocator walPositionFor(PostgresOffsetContext offsetContext) { + Lsn lsn = + offsetContext.lastCompletelyProcessedLsn() != null + ? offsetContext.lastCompletelyProcessedLsn() + : offsetContext.lsn(); + return new WalPositionLocator(offsetContext.lastCommitLsn(), lsn); + } + + @Test + void shouldNotSearchWalPositionOnFreshStart() { + // A fresh stream split start: the starting offset carries an LSN (the low watermark) but + // nothing has been completely processed yet. + final Map offsetValues = new HashMap<>(); + offsetValues.put(SourceInfo.LSN_KEY, 12345L); + offsetValues.put(SourceInfo.TIMESTAMP_USEC_KEY, 67890L); + + final PostgresOffsetContext offsetContext = offsetLoader.load(offsetValues); + + // searchingEnabled() alone is true, so the pre-fix condition would enter the search loop + // and + // stall forever on an idle publication... + assertThat(walPositionFor(offsetContext).searchingEnabled()).isTrue(); + // ...but the added guard is false on a fresh start, so the search is correctly skipped. + assertThat(offsetContext.hasCompletelyProcessedPosition()) + .as( + "WAL search must be skipped on a fresh start so an idle publication cannot stall it") + .isFalse(); + } + + @Test + void shouldSearchWalPositionWhenResumingFromProcessedOffset() { + // A resumed offset (e.g. after a checkpoint): a position has already been processed, so the + // search is still required to locate the exact resume point among already-seen LSNs. + final Map offsetValues = new HashMap<>(); + offsetValues.put(SourceInfo.LSN_KEY, 12345L); + offsetValues.put(SourceInfo.TIMESTAMP_USEC_KEY, 67890L); + offsetValues.put(PostgresOffsetContext.LAST_COMPLETELY_PROCESSED_LSN_KEY, 12345L); + + final PostgresOffsetContext offsetContext = offsetLoader.load(offsetValues); + + // Both the pre-fix condition and the added guard are true, so the search still runs. + assertThat(walPositionFor(offsetContext).searchingEnabled()).isTrue(); + assertThat(offsetContext.hasCompletelyProcessedPosition()) + .as("WAL search must still run when resuming from an already-processed position") + .isTrue(); + } +}