Skip to content

[FLINK-39198] Fix data loss during restoration from checkpoint in TIMESTAMP startup mode. #4297

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
danzhewuju wants to merge 2 commits into
base: master
Choose a base branch
from
+93 −11
Open

[FLINK-39198] Fix data loss during restoration from checkpoint in TIMESTAMP startup mode. #4297

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0c3dd4c
[FLINK-39198] Fix data loss during restoration from checkpoint in TIM…
danzhewuju Mar 3, 2026
532fdba
[FLINK-39198] Add test for restoring from early checkpoint with TIMES…
danzhewuju Jul 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 17 additions & 8 deletions ...rc/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.apache.flink.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffsetKind;
import org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo;
import org.apache.flink.cdc.connectors.mysql.source.split.MySqlBinlogSplit;
import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSplit;
Expand Down Expand Up @@ -391,21 +392,29 @@ private void configureFilter() {
private Predicate<Event> createEventFilter() {
// If the startup mode is set as TIMESTAMP, we need to apply a filter on event to drop
// events earlier than the specified timestamp.

// NOTE: Here we take user's configuration (statefulTaskContext.getSourceConfig())
// as the ground truth. This might be fragile if user changes the config and recover
// the job from savepoint / checkpoint, as there might be conflict between user's config
// and the state in savepoint / checkpoint. But as we don't promise compatibility of
// checkpoint after changing the config, this is acceptable for now.
StartupOptions startupOptions = statefulTaskContext.getSourceConfig().getStartupOptions();
if (startupOptions.startupMode.equals(StartupMode.TIMESTAMP)) {
if (startupOptions.binlogOffset == null) {
// A TIMESTAMP-startup job keeps StartupMode.TIMESTAMP in the source config even after
// restoring from a checkpoint / savepoint. However, the restored split's starting
// offset already points to a concrete resume position (SPECIFIC / GTID) rather than the
// original submission-time timestamp. So we take the split's starting offset as the
// ground truth: only apply the timestamp filter when the offset kind is still
// TIMESTAMP, otherwise valid row events after recovery would be incorrectly dropped.
BinlogOffset startingOffset = currentBinlogSplit.getStartingOffset();
if (startingOffset == null) {
throw new NullPointerException(
"The startup option was set to TIMESTAMP "
+ "but unable to find starting binlog offset. Please check if the timestamp is specified in "
+ "configuration. ");
}
long startTimestampSec = startupOptions.binlogOffset.getTimestampSec();
if (startingOffset.getOffsetKind() != BinlogOffsetKind.TIMESTAMP) {
Comment thread
lvyanquan marked this conversation as resolved.
LOG.info(
"The starting offset kind is {}, no need to apply timestamp filter",
startingOffset.getOffsetKind());
return event -> true;
}

long startTimestampSec = startingOffset.getTimestampSec();
// We only skip data change event, as other kinds of events are necessary for updating
// some internal state inside MySqlStreamingChangeEventSource
LOG.info(
Expand Down
79 changes: 76 additions & 3 deletions ...est/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffsetKind;
import org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo;
import org.apache.flink.cdc.connectors.mysql.source.split.MySqlBinlogSplit;
import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSnapshotSplit;
Expand Down Expand Up @@ -1131,15 +1132,87 @@ void testRestoreFromCheckpointWithTimestampStartingOffset() throws Exception {
// Restore binlog reader from checkpoint
binlogReader.submitSplit(checkpointSplit);

// We mock a WRITE_ROWS event with timestamp = 1, which should be dropped by filter
// Restored checkpoint offsets are concrete positions, so we should not drop this event
// based on the original TIMESTAMP startup setting.
EventHeaderV4 header = new EventHeaderV4();
header.setEventType(EventType.WRITE_ROWS);
header.setTimestamp(1L);
Event event = new Event(header, new WriteRowsEventData());

// Check if the filter works
// Check that the restored reader keeps the event.
Predicate<Event> eventFilter = binlogReader.getBinlogSplitReadTask().getEventFilter();
assertThat(eventFilter.test(event)).isFalse();
Comment thread
danzhewuju marked this conversation as resolved.
assertThat(eventFilter.test(event)).isTrue();
}

@Test
void testRestoreFromEarlyCheckpointWithTimestampStartingOffset() throws Exception {
// This covers the "early checkpoint / savepoint" case: the checkpoint is taken before the
// binlog split has processed any record, so the restored split's starting offset is still
// BinlogOffsetKind.TIMESTAMP with the original submission-time timestamp (T1). The job is
// then resubmitted with a config that carries a different, later TIMESTAMP option (T2 >
// T1).
// The event filter must anchor on the restored split timestamp T1, not the newly submitted
// config timestamp T2.

// Preparations
inventoryDatabase8.createAndInitialize();
MySqlSourceConfig connectionConfig =
getConfig(MYSQL8_CONTAINER, inventoryDatabase8, new String[] {"products"});
binaryLogClient = DebeziumUtils.createBinaryClient(connectionConfig.getDbzConfiguration());
mySqlConnection = DebeziumUtils.createMySqlConnection(connectionConfig);

// T1 is the original submission-time timestamp captured in the checkpoint, while T2 is a
// different (later) timestamp coming from the resubmitted config. T2 > T1 on purpose so we
// can tell which one the filter uses.
long restoredTimestampMs = 15_000L;
long resubmittedTimestampMs = 30_000L;

// The resubmitted job config still runs in TIMESTAMP mode, but with a different timestamp.
MySqlSourceConfig sourceConfig =
getConfig(
MYSQL8_CONTAINER,
inventoryDatabase8,
StartupOptions.timestamp(resubmittedTimestampMs),
new String[] {"products"});

// The restored split still carries a TIMESTAMP starting offset (T1), which is exactly the
// state an early checkpoint / savepoint would hold.
BinlogSplitReader binlogReader = createBinlogReader(sourceConfig);
MySqlBinlogSplit checkpointSplit =
createBinlogSplit(
getConfig(
MYSQL8_CONTAINER,
inventoryDatabase8,
StartupOptions.timestamp(restoredTimestampMs),
new String[] {"products"}));
assertThat(checkpointSplit.getStartingOffset().getOffsetKind())
.isEqualTo(BinlogOffsetKind.TIMESTAMP);
assertThat(checkpointSplit.getStartingOffset().getTimestampSec())
.isEqualTo(restoredTimestampMs / 1000);

// Restore binlog reader from the early checkpoint
binlogReader.submitSplit(checkpointSplit);

Predicate<Event> eventFilter = binlogReader.getBinlogSplitReadTask().getEventFilter();

// An event between T1 and T2 must be kept, because the filter should use the restored split
// timestamp T1 (>= 15000ms) rather than the resubmitted config timestamp T2 (>= 30000ms).
Event eventBetweenT1AndT2 =
mockWriteRowsEvent(restoredTimestampMs + Duration.ofSeconds(5).toMillis());
assertThat(eventFilter.test(eventBetweenT1AndT2)).isTrue();

// An event before T1 must still be dropped, proving the timestamp filter is actually active
// on T1 and did not degrade into a pass-through filter.
Event eventBeforeT1 =
mockWriteRowsEvent(restoredTimestampMs - Duration.ofSeconds(5).toMillis());
assertThat(eventFilter.test(eventBeforeT1)).isFalse();
}

private static Event mockWriteRowsEvent(long timestampMs) {
EventHeaderV4 header = new EventHeaderV4();
header.setEventType(EventType.WRITE_ROWS);
header.setTimestamp(timestampMs);
return new Event(header, new WriteRowsEventData());
}

@Test
Expand Down
Loading