danzhewuju commented on code in PR #4297:
URL: https://github.com/apache/flink-cdc/pull/4297#discussion_r3559424159
##########
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:
You're right, this can indeed be made smaller. I'll limit the scope of
changes to the TIMESTAMP branch in createEventFilter(), and use the restored
split's getStartingOffset() method instead of the source configuration, which
also avoids that edge case. I've already pushed the latest version~ Thanks!
--
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]
