[FLINK-39198] Fix data loss during restoration from checkpoint in TIMESTAMP startup mode.#4297
[FLINK-39198] Fix data loss during restoration from checkpoint in TIMESTAMP startup mode.#4297danzhewuju wants to merge 2 commits into
Conversation
|
This pull request has been automatically marked as stale because it has not had recent activity for 120 days. It will be closed in 60 days if no further activity occurs. |
haruki-830
left a comment
haruki-830
left a comment
There was a problem hiding this comment.
+1. It might be helpful to add a brief code comment explaining the submission-time-based timestamp scenario for future readers.
danzhewuju
force-pushed
the
fix-loss-data-restore-from-checkpoint
branch
from
09f67bb to
64a3e20
Compare
Thank you. I’ve added a comment to clarify the recovery scenario: a TIMESTAMP-startup job may restore using a concrete, SPECIFIC offset; therefore, we should honor the restored split offset rather than reapplying the original submission-time timestamp filter. |
…ESTAMP startup mode. Signed-off-by: danzhewuju <1668926294@qq.com>
danzhewuju
force-pushed
the
fix-loss-data-restore-from-checkpoint
branch
from
64a3e20 to
0c3dd4c
Compare
lvyanquan
left a comment
lvyanquan
left a comment
There was a problem hiding this comment.
The production code looks good to me now.
Left some minor comments.
| + "configuration. "); | ||
| } | ||
| long startTimestampSec = startupOptions.binlogOffset.getTimestampSec(); | ||
| if (startingOffset.getOffsetKind() != BinlogOffsetKind.TIMESTAMP) { |
There was a problem hiding this comment.
minor: Add one more log here.
|
CC @ruanhang1993. |
…AMP starting offset Signed-off-by: danzhewuju <1668926294@qq.com>
3 participants
Fix: https://issues.apache.org/jira/browse/FLINK-39198
What is the purpose of the change
Flink CDC creates an eventFilter inside the MySqlBinlogSplitReadTask. This filter is produced through the following method. In timestamp startOptions mode, the eventFilter will exclude events whose timestamps are earlier than the user-defined timestamp. However, if the job resumes from a savepoint or checkpoint and the system automatically sets the timestamp parameter, this approach can result in data loss.
In this scenario, the user configures a startOption with timestamp mode (which may be the default behavior, tied to the Flink job’s submission time) and restarts the Flink CDC job from a savepoint. If the savepoint contains MySQL binlog information including a ts_sec value, and that ts_sec is earlier than the user-defined timestamp, data loss may occur.
Brief change log
Prevent filtering out binlog events when resuming from savepoints/checkpoints with older ts_sec
Verifying this change
Test result:
Adding split(s) to reader: [MySqlBinlogSplit{splitId='binlog-split', offset={transaction_id=null, ts_sec=1772531889, file=.577482, pos=883008898, kind=SPECIFIC, gtids=, row=1, event=3, server_id=5212, timestamp_strictly=false}, endOffset={ts_sec=0, file=, pos=-9223372036854775808, kind=NON_STOPPING, row=0, event=0, timestamp_strictly=false}, isSuspended=false}]
Creating an event filter that drops row mutation events occurring before the destination timestamp in seconds 1772531889