lvyanquan commented on code in PR #4297:
URL: https://github.com/apache/flink-cdc/pull/4297#discussion_r3557925779
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java:
##########
@@ -421,6 +424,41 @@ private Predicate<Event> createEventFilter() {
return event -> true;
}
+ private Predicate<Event> createEventFilterWithTimestamp(
+ long startTimestampSec, BinlogOffsetKind offsetKind) {
+ StartupOptions startupOptions =
statefulTaskContext.getSourceConfig().getStartupOptions();
+ // If startup mode is not TIMESTAMP, no filtering needed
+ if (!startupOptions.startupMode.equals(StartupMode.TIMESTAMP)) {
+ return event -> true;
+ }
+ if (startupOptions.binlogOffset == 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. ");
+ }
+ // A TIMESTAMP-startup job still keeps StartupMode.TIMESTAMP in the
source config after
+ // recovery, but the restored split may already point to a concrete
checkpoint/savepoint
+ // position (SPECIFIC) instead of the original submission-time
timestamp. In that case we
+ // must follow the restored split offset metadata rather than
re-applying the original
+ // startup timestamp, otherwise valid row events after recovery may be
dropped.
+ if (offsetKind == BinlogOffsetKind.TIMESTAMP) {
+ return createEventFilter();
+ }
+ // If the offset kind is SPECIFIC, that means the job is restored from
a savepoint which has
+ // a specific offset.
+ LOG.info(
+ "Creating an event filter that drops row mutation events
occurring before the destination"
+ + " timestamp in seconds {}",
+ startTimestampSec);
+ return event -> {
+ if (!EventType.isRowMutation(getEventType(event))) {
+ return true;
+ }
+ return event.getHeader().getTimestamp() >= startTimestampSec *
1000;
+ };
+ }
Review Comment:
Thanks for working on this. I think the fix can be smaller and also avoid an
edge case.
In the original code, `createEventFilter()` uses
`startupOptions.binlogOffset.getTimestampSec()` from the current source config.
However, after restoring from checkpoint/savepoint, the source config still
keeps `startupMode = TIMESTAMP`, while the restored split may already contain
the actual resume offset in `currentBinlogSplit.getStartingOffset()`.
So I think we can fix this by changing only the TIMESTAMP branch in
`createEventFilter()`:
```java
StartupOptions startupOptions =
statefulTaskContext.getSourceConfig().getStartupOptions();
if (startupOptions.startupMode.equals(StartupMode.TIMESTAMP)) {
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. ");
}
if (startingOffset.getOffsetKind() != BinlogOffsetKind.TIMESTAMP) {
return event -> true;
}
long startTimestampSec = startingOffset.getTimestampSec();
...
}
--
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]
