Copilot commented on code in PR #4566:
URL: https://github.com/apache/flink-cdc/pull/4566#discussion_r4256173540
##########
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 SourceSplitBase getSplit() {
}
public void commitCurrentOffset(@Nullable Offset offsetToCommit) {
+ 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 && 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)) {
+ // 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) {
Review Comment:
The `streamSplitReadTask.offsetContext != null` guard is now redundant in
this method (the offset context is no longer referenced after removing the
live-offset fallback). This adds an unnecessary coupling and may skip a valid
checkpoint LSN commit if `offsetContext` is temporarily null. Consider
simplifying the condition (or add a brief comment justifying why
`offsetContext` must be non-null to commit).
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java:
##########
@@ -189,6 +199,209 @@ void testNotifyCheckpointWindowSizeDefault() throws
Exception {
assertThat(completedCheckpointIds).containsExactly(101L);
}
+ /**
+ * 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, watch the slot for a while
+ Lsn confirmedFlushLsn = watchConfirmedFlushLsn(checkpointLsn,
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);
+ // give the slot time to receive what was committed
+ Thread.sleep(15_000L);
Review Comment:
Using fixed `Thread.sleep(...)` in container-backed integration tests is
typically slow and can be flaky under load. Prefer waiting on an explicit
condition (e.g., reuse `watchConfirmedFlushLsn(...)` with an expected bound, or
poll until the slot reaches/doesn't exceed a target) so the test proceeds as
soon as the system is ready and fails faster when it isn't.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java:
##########
@@ -390,6 +603,19 @@ private List<String> consumeSnapshotRecords(
return formatter.format(output.getResults());
}
+ private List<String> pollStreamRecords(
+ PostgresSourceReader sourceReader, DataType recordType, long
timeoutMillis)
+ throws Exception {
+ final SimpleReaderOutput output = new SimpleReaderOutput();
+ long deadline = System.currentTimeMillis() + timeoutMillis;
+ while (System.currentTimeMillis() < deadline) {
+ if (sourceReader.pollNext(output) != MORE_AVAILABLE) {
+ Thread.sleep(100L);
+ }
+ }
+ return new RecordsFormatter(recordType).format(output.getResults());
+ }
Review Comment:
`pollStreamRecords` always waits until `timeoutMillis` elapses, even if
records have already been produced. This makes tests slower and can amplify CI
flakiness. Consider returning early once at least one record is collected (or
accept an `expectedCount` and stop when reached), similar to the existing
`consumeStreamRecords` behavior.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java:
##########
@@ -189,6 +199,209 @@ void testNotifyCheckpointWindowSizeDefault() throws
Exception {
assertThat(completedCheckpointIds).containsExactly(101L);
}
+ /**
+ * 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, watch the slot for a while
+ Lsn confirmedFlushLsn = watchConfirmedFlushLsn(checkpointLsn,
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);
+ // give the slot time to receive what was committed
+ Thread.sleep(15_000L);
+ }
+
+ // restore from checkpoint 4, the latest completed one
+ try (PostgresSourceReader reader = createStreamReader(3)) {
+ reader.start();
+ reader.addSplits(checkpointState);
+ Thread.sleep(1000L);
+ insertCustomer(3003);
+
+ List<String> records = new ArrayList<>();
+ long deadline = System.currentTimeMillis() + 30_000L;
+ while (!records.contains(customerRecord(3003))
+ && System.currentTimeMillis() < deadline) {
+ records.addAll(pollStreamRecords(reader, CUSTOMER_TYPE,
1_000L));
+ }
+
assertThat(records).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)));
+ Thread.sleep(1000L);
+
+ insertCustomer(3001);
+ assertThat(pollStreamRecords(reader, CUSTOMER_TYPE, 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, 5_000L);
+ reader.notifyCheckpointComplete(checkpointId);
+
+ Lsn confirmedFlushLsn = watchConfirmedFlushLsn(checkpointLsn,
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)));
+ Thread.sleep(1000L);
+
+ insertCustomer(3001);
+ assertThat(pollStreamRecords(reader, CUSTOMER_TYPE, 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, 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 beyond the limit, or the last
one seen. */
+ private Lsn watchConfirmedFlushLsn(Lsn limit, long timeoutMillis) throws
Exception {
+ long deadline = System.currentTimeMillis() + timeoutMillis;
+ Lsn lsn = null;
+ while (System.currentTimeMillis() < deadline) {
+ try (Connection conn =
+ getJdbcConnection(
+ POSTGRES_CONTAINER,
customDatabase.getDatabaseName());
+ Statement stmt = conn.createStatement();
+ ResultSet rs =
+ stmt.executeQuery(
+ String.format(
+ "SELECT confirmed_flush_lsn FROM
pg_replication_slots"
+ + " WHERE slot_name =
'%s'",
+ slotName))) {
+ if (rs.next() && rs.getString(1) != null) {
+ lsn = Lsn.valueOf(rs.getString(1));
+ if (lsn.compareTo(limit) > 0) {
+ return lsn;
+ }
+ }
+ }
Review Comment:
`watchConfirmedFlushLsn` opens a new JDBC connection on every polling
iteration, which can be expensive and cause avoidable test slowness. Consider
opening a single connection/statement outside the loop and reusing it, and
switching to a `PreparedStatement` (even in tests) to avoid string-formatted
SQL.
--
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]
