dev-donghwan commented on code in PR #4566:
URL: https://github.com/apache/flink-cdc/pull/4566#discussion_r4256385158
##########
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:
Removed in e6e5596. `offsetContext` is set right before the streaming
starts, and `commitOffset` already ignores the request while the replication
stream is not open, so the check did not guard anything.
##########
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:
Replaced in e6e5596. The tests now wait until the replication slot is active
after adding the stream split, and until `confirmed_flush_lsn` reaches the
checkpoint offset before restoring. The newly-added-tables test still watches
the slot for up to 15s, since the slot is already at the offset of checkpoint
1, which equals that of checkpoint 7, and it has to see whether the slot passes
it.
##########
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:
Done in e6e5596, it now reuses one connection and a `PreparedStatement`.
##########
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:
Done in e6e5596, `pollStreamRecords` now takes an expected count and returns
as soon as it is reached.
--
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]
