lvyanquan commented on code in PR #4297:
URL: https://github.com/apache/flink-cdc/pull/4297#discussion_r3570692196
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java:
##########
@@ -391,21 +392,26 @@ 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) {
Review Comment:
minor: Add one more log here.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java:
##########
@@ -1131,15 +1131,16 @@ 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();
Review Comment:
Could we add one more test for the early checkpoint/savepoint case?
Scenario:
1. The original job starts with TIMESTAMP mode and timestamp T1.
2. A checkpoint/savepoint is taken before the binlog split processes any
record, so the restored split starting offset is still
`BinlogOffsetKind.TIMESTAMP` with timestamp T1.
3. The job is submitted again from that state, but the source config now
contains a different TIMESTAMP startup option, e.g. T2 > T1.
4. The event filter should use the restored split timestamp T1, not the
newly submitted config timestamp T2.
--
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]
