This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 3a0b76814 [FLINK-40853][postgres] Skip LSN commit for checkpoints
without a stream offset (#4566)
3a0b76814 is described below
commit 3a0b768145064cb64e2b9715af04330e8684b18e
Author: κΉλν <[email protected]>
AuthorDate: Wed Oct 21 12:03:54 2026 +0900
[FLINK-40853][postgres] Skip LSN commit for checkpoints without a stream
offset (#4566)
---
.../source/fetch/PostgresStreamFetchTask.java | 50 ++--
.../source/reader/PostgresSourceReaderTest.java | 272 ++++++++++++++++++++-
2 files changed, 294 insertions(+), 28 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresStreamFetchTask.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresStreamFetchTask.java
index a3d28a12c..893db82d1 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresStreamFetchTask.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresStreamFetchTask.java
@@ -124,33 +124,29 @@ public class PostgresStreamFetchTask implements
FetchTask<SourceSplitBase> {
}
public void commitCurrentOffset(@Nullable Offset offsetToCommit) {
- if (streamSplitReadTask != null && streamSplitReadTask.offsetContext
!= null) {
- PostgresOffsetContext postgresOffsetContext =
streamSplitReadTask.offsetContext;
-
- // only extracting and storing the lsn of the last commit
- Long commitLsn =
- (Long)
- postgresOffsetContext
- .getOffset()
-
.get(PostgresOffsetContext.LAST_COMMIT_LSN_KEY);
-
- if (offsetToCommit != null) {
- // We should commit the checkpoint's LSN instead of
postgresOffsetContext's LSN to
- // the slot.
- // If the checkpoint succeeds and a table UPDATE message
arrives before the
- // notifyCheckpoint is called, which is represented as a
BEGIN/UPDATE/COMMIT WAL
- // event sequence. The LSN of postgresOffsetContext will be
updated to the LSN of
- // the COMMIT event. Committing the COMMIT LSN to the slot is
incorrect because if a
- // failover occurs after the successful commission, Flink will
recover from that
- // checkpoint and consume WAL starting from the slot LSN that
is the LSN of COMMIT
- // event, rather than from the checkpoint's LSN. Therefore,
UPDATE messages cannot
- // be consumed, resulting in data loss.
- commitLsn = ((PostgresOffset)
offsetToCommit).getLsn().asLong();
- }
-
- if (commitLsn != null
- && (lastCommitLsn == null
- ||
Lsn.valueOf(commitLsn).compareTo(Lsn.valueOf(lastCommitLsn)) > 0)) {
+ if (offsetToCommit == null) {
+ // No stream offset was recorded for the checkpoint, e.g. it was
taken before the
+ // stream split was assigned, or while newly added tables were
being snapshotted. The
+ // LSN of postgresOffsetContext may already be beyond the offset
of the latest
+ // completed checkpoint, so committing it could lose data for the
reason below.
+ LOG.debug("Skip committing offset as the checkpoint has no stream
offset.");
+ return;
+ }
+ if (streamSplitReadTask != null) {
+ // We should commit the checkpoint's LSN instead of
postgresOffsetContext's LSN to
+ // the slot.
+ // If the checkpoint succeeds and a table UPDATE message arrives
before the
+ // notifyCheckpoint is called, which is represented as a
BEGIN/UPDATE/COMMIT WAL
+ // event sequence. The LSN of postgresOffsetContext will be
updated to the LSN of
+ // the COMMIT event. Committing the COMMIT LSN to the slot is
incorrect because if a
+ // failover occurs after the successful commission, Flink will
recover from that
+ // checkpoint and consume WAL starting from the slot LSN that is
the LSN of COMMIT
+ // event, rather than from the checkpoint's LSN. Therefore, UPDATE
messages cannot
+ // be consumed, resulting in data loss.
+ long commitLsn = ((PostgresOffset)
offsetToCommit).getLsn().asLong();
+
+ if (lastCommitLsn == null
+ ||
Lsn.valueOf(commitLsn).compareTo(Lsn.valueOf(lastCommitLsn)) > 0) {
lastCommitLsn = commitLsn;
Map<String, Object> offsets = new HashMap<>();
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java
index 7ef016b97..2852e6c2d 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java
@@ -35,6 +35,7 @@ import
org.apache.flink.cdc.connectors.postgres.source.PostgresDialect;
import org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder;
import
org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig;
import
org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory;
+import
org.apache.flink.cdc.connectors.postgres.source.events.OffsetCommitEvent;
import org.apache.flink.cdc.connectors.postgres.source.offset.PostgresOffset;
import
org.apache.flink.cdc.connectors.postgres.source.offset.PostgresOffsetFactory;
import org.apache.flink.cdc.connectors.postgres.testutils.RecordsFormatter;
@@ -48,6 +49,7 @@ import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.Collector;
+import io.debezium.connector.postgresql.connection.Lsn;
import io.debezium.relational.TableId;
import io.debezium.relational.history.TableChanges;
import org.apache.kafka.connect.source.SourceRecord;
@@ -58,6 +60,8 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
@@ -66,6 +70,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.function.Predicate;
import static
io.debezium.connector.postgresql.connection.ReplicationConnection.Builder.DEFAULT_PUBLICATION_NAME;
import static org.apache.flink.core.io.InputStatus.END_OF_INPUT;
@@ -76,6 +81,12 @@ import static org.assertj.core.api.Assertions.assertThat;
public class PostgresSourceReaderTest extends PostgresTestBase {
private static final String DB_NAME_PREFIX = "postgres";
private static final String SCHEMA_NAME = "customer";
+ private static final DataType CUSTOMER_TYPE =
+ DataTypes.ROW(
+ DataTypes.FIELD("Id", DataTypes.BIGINT()),
+ DataTypes.FIELD("Name", DataTypes.STRING()),
+ DataTypes.FIELD("address", DataTypes.STRING()),
+ DataTypes.FIELD("phone_number", DataTypes.STRING()));
private String slotName;
private final UniqueDatabase customDatabase =
new UniqueDatabase(
@@ -286,6 +297,235 @@ public class PostgresSourceReaderTest extends
PostgresTestBase {
return source.createReader(new TestingReaderContext());
}
+ /**
+ * Checkpoints taken before the stream split is assigned (e.g. during the
snapshot phase) have
+ * no stream offset. With a commit delay, such a checkpoint is still
committed after the stream
+ * phase has started. The committed LSN must not go beyond the offset of
the latest completed
+ * checkpoint, otherwise the changes in between are lost when restoring
from it.
+ */
+ @Test
+ void testLsnCommitDoesNotPassLatestCompletedCheckpoint() throws Exception {
+ PostgresSourceConfig sourceConfig = createConfigFactory().create(0);
+ try (PostgresDialect dialect = new PostgresDialect(sourceConfig);
+ PostgresSourceReader reader = createStreamReader(3)) {
+ Lsn checkpointLsn =
+
streamOffsetLsn(completeCheckpointsAcrossStreamStart(reader, dialect));
+
+ // checkpoint 4 is committed by now, wait until the slot reaches it
+ Lsn confirmedFlushLsn =
+ waitForConfirmedFlushLsn(lsn ->
lsn.compareTo(checkpointLsn) >= 0, 15_000L);
+ assertThat(confirmedFlushLsn)
+ .as(
+ "confirmed_flush_lsn %s must not pass the offset
%s of the latest"
+ + " completed checkpoint",
+ confirmedFlushLsn, checkpointLsn)
+ .isLessThanOrEqualTo(checkpointLsn);
+ }
+ }
+
+ /** Same as above, but restores from the latest completed checkpoint and
checks the data. */
+ @Test
+ void testNoDataLossWhenRestoringAfterLsnCommit() throws Exception {
+ PostgresSourceConfig sourceConfig = createConfigFactory().create(0);
+ List<SourceSplitBase> checkpointState;
+ try (PostgresDialect dialect = new PostgresDialect(sourceConfig);
+ PostgresSourceReader reader = createStreamReader(3)) {
+ checkpointState = completeCheckpointsAcrossStreamStart(reader,
dialect);
+ // wait until the slot receives what was committed
+ Lsn checkpointLsn = streamOffsetLsn(checkpointState);
+ waitForConfirmedFlushLsn(lsn -> lsn.compareTo(checkpointLsn) >= 0,
15_000L);
+ }
+
+ // restore from checkpoint 4, the latest completed one
+ try (PostgresSourceReader reader = createStreamReader(3)) {
+ reader.start();
+ reader.addSplits(checkpointState);
+ waitForSlotActive(30_000L);
+ insertCustomer(3003);
+
+ assertThat(pollStreamRecords(reader, CUSTOMER_TYPE, 2, 30_000L))
+ .contains(customerRecord(3003))
+ .contains(customerRecord(3002));
+ }
+ }
+
+ /**
+ * With scan.newly-added-table.enabled, the enumerator turns offset commit
off while newly added
+ * tables are snapshotted, and the reader drops the stream offsets of the
checkpoints taken
+ * meanwhile. Once offset commit is turned on again, those checkpoints are
still committed
+ * because of the commit delay. The committed LSN must not go beyond the
offset of the latest
+ * completed checkpoint.
+ */
+ @Test
+ void
testLsnCommitDoesNotPassLatestCompletedCheckpointAfterNewlyAddedTables() throws
Exception {
+ PostgresSourceConfig sourceConfig = createConfigFactory().create(0);
+ try (PostgresDialect dialect = new PostgresDialect(sourceConfig);
+ PostgresSourceReader reader = createStreamReader(3, true)) {
+ reader.start();
+ reader.handleSourceEvents(new OffsetCommitEvent(true));
+
reader.addSplits(Collections.singletonList(createStreamSplit(dialect)));
+ waitForSlotActive(30_000L);
+
+ insertCustomer(3001);
+ assertThat(pollStreamRecords(reader, CUSTOMER_TYPE, 1, 5_000L))
+ .containsExactly(customerRecord(3001));
+ long checkpointId = 1L;
+ for (; checkpointId <= 3L; checkpointId++) {
+ reader.snapshotState(checkpointId);
+ reader.notifyCheckpointComplete(checkpointId);
+ }
+
+ // newly added tables are being snapshotted
+ reader.handleSourceEvents(new OffsetCommitEvent(false));
+ for (; checkpointId <= 6L; checkpointId++) {
+ reader.snapshotState(checkpointId);
+ reader.notifyCheckpointComplete(checkpointId);
+ }
+
+ // the snapshot of newly added tables has finished
+ reader.handleSourceEvents(new OffsetCommitEvent(true));
+ Lsn checkpointLsn =
streamOffsetLsn(reader.snapshotState(checkpointId));
+
+ // a change after the latest checkpoint, which must be replayed
when restoring from it
+ insertCustomer(3002);
+ pollStreamRecords(reader, CUSTOMER_TYPE, 1, 5_000L);
+ reader.notifyCheckpointComplete(checkpointId);
+
+ // the slot may already be at the offset of checkpoint 1, which
equals that of
+ // checkpoint
+ // 7, so watch it for a while in case it passes the offset
+ Lsn confirmedFlushLsn =
+ waitForConfirmedFlushLsn(lsn ->
lsn.compareTo(checkpointLsn) > 0, 15_000L);
+ assertThat(confirmedFlushLsn)
+ .as(
+ "confirmed_flush_lsn %s must not pass the offset
%s of the latest"
+ + " completed checkpoint",
+ confirmedFlushLsn, checkpointLsn)
+ .isLessThanOrEqualTo(checkpointLsn);
+ }
+ }
+
+ /**
+ * Takes checkpoints 1 to 3 before the stream split is assigned, then
checkpoint 4 in the stream
+ * phase after change 3001, then reads change 3002, and completes
checkpoints 1 to 7. Returns
+ * the state of checkpoint 4.
+ */
+ private List<SourceSplitBase> completeCheckpointsAcrossStreamStart(
+ PostgresSourceReader reader, PostgresDialect dialect) throws
Exception {
+ reader.start();
+
+ // checkpoints taken before the stream split is assigned
+ reader.snapshotState(1L);
+ reader.snapshotState(2L);
+ reader.snapshotState(3L);
+
+
reader.addSplits(Collections.singletonList(createStreamSplit(dialect)));
+ waitForSlotActive(30_000L);
+
+ insertCustomer(3001);
+ assertThat(pollStreamRecords(reader, CUSTOMER_TYPE, 1, 5_000L))
+ .containsExactly(customerRecord(3001));
+ List<SourceSplitBase> checkpointState = reader.snapshotState(4L);
+
+ // a change after checkpoint 4, which must be replayed when restoring
from it. Keep
+ // polling so that the stream reader reads it and moves its offset
context forward.
+ insertCustomer(3002);
+ pollStreamRecords(reader, CUSTOMER_TYPE, 1, 5_000L);
+
+ for (long checkpointId = 1L; checkpointId <= 7L; checkpointId++) {
+ if (checkpointId > 4L) {
+ reader.snapshotState(checkpointId);
+ }
+ reader.notifyCheckpointComplete(checkpointId);
+ }
+ return checkpointState;
+ }
+
+ private StreamSplit createStreamSplit(PostgresDialect dialect) {
+ PostgresOffsetFactory offsetFactory = new PostgresOffsetFactory();
+ return new StreamSplit(
+ StreamSplit.STREAM_SPLIT_ID,
+ offsetFactory.createInitialOffset(),
+ offsetFactory.createNoStoppingOffset(),
+ Collections.emptyList(),
+
dialect.discoverDataCollectionSchemas(createConfigFactory().create(0)),
+ 0);
+ }
+
+ private static String customerRecord(int id) {
+ return String.format("+I[%d, user, Beijing, 123]", id);
+ }
+
+ private void insertCustomer(int id) throws SQLException {
+ try (Connection conn =
+ getJdbcConnection(POSTGRES_CONTAINER,
customDatabase.getDatabaseName());
+ Statement stmt = conn.createStatement()) {
+ stmt.execute(
+ String.format(
+ "INSERT INTO customer.\"Customers\" VALUES (%d,
'user', 'Beijing', '123')",
+ id));
+ }
+ }
+
+ private static Lsn streamOffsetLsn(List<SourceSplitBase> splits) {
+ return splits.stream()
+ .filter(SourceSplitBase::isStreamSplit)
+ .map(split -> ((PostgresOffset)
split.asStreamSplit().getStartingOffset()).getLsn())
+ .findFirst()
+ .orElseThrow(() -> new IllegalStateException("No stream split
in state"));
+ }
+
+ /**
+ * Returns the first confirmed_flush_lsn that matches the condition, or
the last one seen when
+ * the timeout elapses.
+ */
+ private Lsn waitForConfirmedFlushLsn(Predicate<Lsn> condition, long
timeoutMillis)
+ throws Exception {
+ long deadline = System.currentTimeMillis() + timeoutMillis;
+ Lsn lsn = null;
+ try (Connection conn =
+ getJdbcConnection(POSTGRES_CONTAINER,
customDatabase.getDatabaseName());
+ PreparedStatement stmt =
+ conn.prepareStatement(
+ "SELECT confirmed_flush_lsn FROM
pg_replication_slots"
+ + " WHERE slot_name = ?")) {
+ stmt.setString(1, slotName);
+ while (System.currentTimeMillis() < deadline) {
+ try (ResultSet rs = stmt.executeQuery()) {
+ if (rs.next() && rs.getString(1) != null) {
+ lsn = Lsn.valueOf(rs.getString(1));
+ if (condition.test(lsn)) {
+ return lsn;
+ }
+ }
+ }
+ Thread.sleep(200L);
+ }
+ }
+ return lsn;
+ }
+
+ /** Waits until the stream reader is connected to the replication slot. */
+ private void waitForSlotActive(long timeoutMillis) throws Exception {
+ long deadline = System.currentTimeMillis() + timeoutMillis;
+ try (Connection conn =
+ getJdbcConnection(POSTGRES_CONTAINER,
customDatabase.getDatabaseName());
+ PreparedStatement stmt =
+ conn.prepareStatement(
+ "SELECT active FROM pg_replication_slots WHERE
slot_name = ?")) {
+ stmt.setString(1, slotName);
+ while (System.currentTimeMillis() < deadline) {
+ try (ResultSet rs = stmt.executeQuery()) {
+ if (rs.next() && rs.getBoolean(1)) {
+ return;
+ }
+ }
+ Thread.sleep(200L);
+ }
+ }
+ throw new IllegalStateException("Replication slot " + slotName + " is
not active");
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
void testMultipleSnapshotSplit(boolean skipBackFill) throws Exception {
@@ -487,6 +727,24 @@ public class PostgresSourceReaderTest extends
PostgresTestBase {
return formatter.format(output.getResults());
}
+ /** Polls until at least {@code expectedCount} records are read or the
timeout elapses. */
+ private List<String> pollStreamRecords(
+ PostgresSourceReader sourceReader,
+ DataType recordType,
+ int expectedCount,
+ long timeoutMillis)
+ throws Exception {
+ final SimpleReaderOutput output = new SimpleReaderOutput();
+ long deadline = System.currentTimeMillis() + timeoutMillis;
+ while (output.getResults().size() < expectedCount
+ && System.currentTimeMillis() < deadline) {
+ if (sourceReader.pollNext(output) != MORE_AVAILABLE) {
+ Thread.sleep(100L);
+ }
+ }
+ return new RecordsFormatter(recordType).format(output.getResults());
+ }
+
private List<String> consumeStreamRecords(
PostgresSourceReader sourceReader, DataType recordType, int size)
throws Exception {
// Poll all the n records of the single split.
@@ -518,10 +776,22 @@ public class PostgresSourceReaderTest extends
PostgresTestBase {
}
private PostgresSourceReader createStreamReader() throws Exception {
+ return createStreamReader(1);
+ }
+
+ private PostgresSourceReader createStreamReader(final int
lsnCommitCheckpointsDelay)
+ throws Exception {
+ return createStreamReader(lsnCommitCheckpointsDelay, false);
+ }
+
+ private PostgresSourceReader createStreamReader(
+ final int lsnCommitCheckpointsDelay, boolean
scanNewlyAddedTableEnabled)
+ throws Exception {
final PostgresOffsetFactory offsetFactory = new
PostgresOffsetFactory();
final PostgresSourceConfigFactory configFactory =
createConfigFactory();
configFactory.startupOptions(StartupOptions.latest());
- configFactory.setLsnCommitCheckpointsDelay(1);
+ configFactory.setLsnCommitCheckpointsDelay(lsnCommitCheckpointsDelay);
+ configFactory.scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled);
PostgresDialect dialect = new PostgresDialect(configFactory.create(0));
final PostgresSourceBuilder.PostgresIncrementalSource<?> source =
new PostgresSourceBuilder.PostgresIncrementalSource<>(
