diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java index b1e6d1dfc86..4050e2b7617 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java @@ -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; @@ -391,21 +392,29 @@ private void configureFilter() { private Predicate 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) { + 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( diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java index e7fd2c4cbb9..22b85c11fbd 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java @@ -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; @@ -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 eventFilter = binlogReader.getBinlogSplitReadTask().getEventFilter(); - assertThat(eventFilter.test(event)).isFalse(); + 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 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