danzhewuju commented on code in PR #4297:
URL: https://github.com/apache/flink-cdc/pull/4297#discussion_r3575692184
##########
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:
done
--
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]
