[Fix][Connector-V2][MySQL CDC] Use checkpoint offset for timestamp startup restore

[goutamadwant]

Purpose of this pull request

Fixes #10899.

For startup.mode = timestamp, the initial incremental split stores a timestamp-only binlog offset. That offset should be resolved to a concrete MySQL binlog position before the reader starts.

After a checkpoint restore, the split already contains the concrete binlog offset saved by the reader. Reusing the configured timestamp at that point can move the recovery anchor back to the original startup timestamp, so this change uses the restored offset directly and skips the timestamp event filter for restored binlog offsets.

Does this PR introduce any user-facing change?

Yes. MySQL CDC jobs configured with startup.mode = timestamp can recover from checkpoints using the saved binlog offset, instead of resolving the original startup timestamp again.

How was this patch tested?

Added unit coverage for the restore decisions and checkpoint-state path:

• MySqlSourceFetchTaskContext only resolves the configured timestamp for timestamp-only bootstrap offsets.
• MySqlBinlogFetchTask only applies the timestamp filter for timestamp-only bootstrap offsets.
• IncrementalSplitState.toSourceSplit() preserves a checkpointed binlog offset after a timestamp bootstrap, so restore does not fall back to the original timestamp.

Verified with:

JAVA_HOME=$(/usr/libexec/java_home -v 11) PATH="$JAVA_HOME/bin:$PATH" ./mvnw -pl seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql -DfailIfNoTests=false package

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If necessary, please update incompatible-changes.md to describe the incompatibility caused by this PR.
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[DanielLeens]

Thanks for working on this. I reviewed the full current head locally and traced the restore path from split startup offset selection into the binlog fetch task.

What this PR solves

• User pain: with startup.mode = TIMESTAMP, a job restored from checkpoint can incorrectly re-resolve the startup position from the configured timestamp instead of honoring the restored binlog offset.
• Fix approach: distinguish a pure timestamp bootstrap offset from a real restored binlog offset, and only apply timestamp resolution/filtering to the former.
• One-line summary: the source-level fix looks correct to me on the latest head, and I did not find a new code blocker.

Runtime path I checked

startup / restore
  -> MySqlSourceFetchTaskContext.getInitOffset() [302-315]
      -> only resolve timestamp when the split startup offset is still a pure timestamp placeholder
      -> otherwise reuse the restored split startup offset directly

binlog read
  -> MySqlBinlogFetchTask.execute() [73-101]
      -> shouldFilterByTimestamp(...) [143-147]
      -> only uses TimestampFilterMySqlStreamingChangeEventSource for the timestamp-bootstrap case

Re-review result

• BinlogOffset.isTimestampOffset() (BinlogOffset.java:114-118) cleanly separates a timestamp-only bootstrap offset from a restored binlog/file-position offset.
• That same guard is now used consistently in both startup-offset resolution and timestamp filtering.
• I do not see a reopened GTID/file-position correctness issue from this change on the current head.

Tests / CI

• The new unit coverage around timestamp startup offset handling looks good to me.
• I did not see a flaky-test pattern in the added tests.
• The current GitHub Build failure does not look like a source regression from this diff. The check-run output is still the fork workflow detection failure, so it is not giving us a usable validation signal yet.

Conclusion: can merge after fixes

1. Blocking items

• No new source-level blocker from my side on the latest head.
• Please get the current GitHub Build into a valid state first (enable/re-run the fork workflow or refresh from latest dev and push again) so the branch has a real CI signal before merge.

2. Suggested non-blocking follow-up

• None from my side on the code path itself.

From the code-review side this looks good now. The remaining blocker is the missing CI signal, not a logic bug in the latest restore-path fix.

[goutamadwant]

Thanks for your review @DanielLeens.. I synced the branch with latest dev and pushed again. The Build check is running now.

[DanielLeens]

Thanks for working on this. I reviewed the full current head locally and traced the restore path from split startup offset selection into the binlog fetch task.

What this PR solves

• User pain: with startup.mode = TIMESTAMP, a job restored from checkpoint can incorrectly re-resolve the startup position from the configured timestamp instead of honoring the restored binlog offset.
• Fix approach: distinguish a pure timestamp bootstrap offset from a real restored binlog offset, and only apply timestamp resolution/filtering to the former.
• One-line summary: the source-level fix looks correct to me on the latest head, and I did not find a new code blocker.

Runtime path I checked

startup / restore
  -> MySqlSourceFetchTaskContext.getInitOffset() [302-315]
      -> only resolve timestamp when the split startup offset is still a pure timestamp placeholder
      -> otherwise reuse the restored split startup offset directly

binlog read
  -> MySqlBinlogFetchTask.execute() [73-101]
      -> shouldFilterByTimestamp(...) [143-147]
      -> only uses TimestampFilterMySqlStreamingChangeEventSource for the timestamp-bootstrap case

Re-review result

• BinlogOffset.isTimestampOffset() (BinlogOffset.java:114-118) cleanly separates a timestamp-only bootstrap offset from a restored binlog/file-position offset.
• That same guard is now used consistently in both startup-offset resolution and timestamp filtering.
• I do not see a reopened GTID/file-position correctness issue from this change on the current head.

Tests / CI

• The new unit coverage around timestamp startup offset handling looks good to me.
• I did not see a flaky-test pattern in the added tests.
• The current GitHub Build failure does not look like a source regression from this diff. The check-run output is still not giving a clean, usable CI signal for the latest head.

Conclusion: can merge after fixes

1. Blocking items

• No new source-level blocker from my side on the latest head.
• Please get the current GitHub Build into a valid state first (enable/re-run the fork workflow or refresh from latest dev and push again) so the branch has a real CI signal before merge.

2. Suggested non-blocking follow-up

• None from my side on the code path itself.

From the code-review side this looks good now. The remaining blocker is the missing CI signal, not a logic bug in the latest restore-path fix.

[DanielLeens]

Thanks for the update. I re-reviewed the latest head from scratch and traced the full restore path again, from split startup-offset selection into the binlog reader startup behavior.

What this PR fixes

• User pain: with startup.mode = timestamp, a checkpoint restore can accidentally fall back to the original configured timestamp instead of reusing the concrete binlog offset that was already persisted by the reader.
• Fix approach: the patch now distinguishes a timestamp-only bootstrap offset from a real restored binlog/file-position offset, and it uses the same guard in both startup-offset resolution and timestamp-filter activation.
• In one sentence: this revision makes the restore path honor checkpoint state instead of rewinding to the original timestamp anchor.

Simple example:

• Before this fix, a job that had already advanced to mysql-bin.000021:4096 could still re-run the original timestamp lookup after restore.
• On the current head, that restored split offset is reused directly, and the binlog reader no longer attaches the timestamp filter in that restore case.

Full runtime chain I checked

timestamp bootstrap
  -> MySqlSourceFetchTaskContext.getInitOffset(...) [302-315]
      -> shouldResolveTimestampStartupOffset(...) [318-323] = true
      -> findBinlogOffsetBytimestamp(...)
  -> MySqlBinlogFetchTask.execute(...) [75-100]
      -> shouldFilterByTimestamp(...) [143-146] = true
      -> TimestampFilterMySqlStreamingChangeEventSource

checkpoint restore
  -> IncrementalSplitState.toSourceSplit() restores a concrete binlog offset
  -> MySqlSourceFetchTaskContext.getInitOffset(...) [302-315]
      -> shouldResolveTimestampStartupOffset(...) = false
      -> reuse splitStartupOffset directly
  -> MySqlBinlogFetchTask.execute(...) [75-100]
      -> shouldFilterByTimestamp(...) = false
      -> plain MySqlStreamingChangeEventSource

Re-review result

• I do not see a reopened source-level bug on the latest head.
• BinlogOffset.isTimestampOffset() (BinlogOffset.java:114-117) cleanly distinguishes the timestamp-only bootstrap placeholder from a real restored binlog offset.
• That same distinction is now applied consistently in both MySqlSourceFetchTaskContext.java:302-323 and MySqlBinlogFetchTask.java:75-76,143-146, which is exactly what this restore-path fix needed.
• The new tests cover both key semantics:
  • MySqlTimestampStartupOffsetTest.java:34-68
  • MySqlBinlogFetchTaskTest.java:28-38

Compatibility

• Fully compatible from my review.
• No API, config, default-value, protocol, or serialization break was introduced here.

Performance / side effects

• The new guard checks are lightweight.
• On the restore path, this actually avoids an unnecessary timestamp-to-binlog lookup when a concrete checkpointed offset is already available.
• I did not find a new concurrency, retry, or resource-release issue on the current head.

Error handling / logging

• Error handling is still consistent with the existing code path: timestamp-resolution failures are still surfaced instead of being silently swallowed.
• I did not find a new logging or exception-handling problem in the changed MySQL CDC path.

Test stability assessment

• Rating: stable
• Basis:
  • the newly added tests are pure guard/state tests with no timing, network, port, or external-environment dependency;
  • I did not see a flaky-test anti-pattern in the added coverage.

CI status I checked

The current Build is still red, but the failures I traced are outside this PR's changed MySQL CDC files:

1. unit-test (8, ubuntu-latest)

• failing test: SeaTunnelEngineClusterRoleTest.testWorkerIsFirstMemberThenGetJobDetailStatus
• module: seatunnel-engine-client
• evidence: the latest log ends with There are test failures in seatunnel-engine-client, with the error in SeaTunnelEngineClusterRoleTest.java:459/471

2. updated-modules-integration-test-part-1 (8, ubuntu-latest)

• failure: replication slot "seatunnel" already exists
• evidence: the latest log shows PostgreSQL CDC integration failure with org.postgresql.util.PSQLException: ERROR: replication slot "seatunnel" already exists

3. kafka-connector-it (11, ubuntu-latest)

• failure: Kafka JAAS / auth setup
• evidence: the latest log shows No JAAS configuration section named 'Client' and then Failed to construct kafka producer

So from the code-review side I still do not see a source regression caused by this diff, but the branch does not have a clean CI signal yet.

Conclusion: can merge after fixes

1. Blocking items

• Please get the current Build back to a usable green state before merge. The source change itself looks fine to me, but the branch still needs a valid CI signal.

2. Suggested follow-up

• No additional non-blocking source issue from my side on the current MySQL CDC restore path.

From the source-review perspective this looks good now. The remaining blocker is the still-red CI, not a logic problem in the changed MySQL CDC code path.

[goutamadwant]

@DanielLeens @davidzollo I checked the latest CI failure.

The failed jobs look unrelated to this MySQL CDC change:

• unit-test (8, ubuntu-latest) failed in seatunnel-engine-server at JobStateEventTest.testNotEndJobStateEvent
• paimon-connector-it (11, ubuntu-latest) failed in PaimonSinkCDCIT.testSinkWithIncompatibleSchema

I do not see these failures touching the changed MySQL CDC restore path or the new timestamp offset tests.

This looks similar to the broad CI failures showing up on a few recent PRs. Could someone rerun the failed jobs when possible

[DanielLeens]

Thanks for the update. I re-reviewed the latest head 557759d743cc545fd7fecc86057e7fef5ac3cfe1 from scratch and retraced the restore path from split startup-offset resolution into binlog reader startup on the current code.

What this PR solves

• User pain: with startup.mode = timestamp, a job restored from checkpoint can accidentally fall back to the original configured timestamp instead of reusing the concrete binlog offset already persisted by the reader.
• Fix approach: distinguish a pure timestamp bootstrap placeholder from a real restored binlog offset, and use the same guard in both startup-offset resolution and timestamp-filter activation.
• One-line summary: the current head makes checkpoint restore honor the saved binlog position instead of rewinding to the configured timestamp anchor.

Simple example:

• Before this fix, a job that had already advanced to mysql-bin.000021:4096 could still re-run the timestamp lookup after restore.
• On the current head, that restored split offset is reused directly, and the binlog reader no longer attaches the timestamp filter in that restore case.

Full runtime chain I checked

timestamp bootstrap
  -> MySqlSourceFetchTaskContext.getInitOffset(...)
      -> shouldResolveTimestampStartupOffset(startupMode, splitStartupOffset)
      -> resolve from configured timestamp only when the split still carries a timestamp-only bootstrap offset

checkpoint restore
  -> IncrementalSplitState restores startupOffset from reader state
  -> restored split now carries a real binlog file/position offset
  -> shouldResolveTimestampStartupOffset(...) returns false
  -> getInitOffset(...) reuses the restored startup offset directly

binlog reader startup
  -> MySqlBinlogFetchTask.execute(...)
      -> shouldFilterByTimestamp(startupMode, split.getStartupOffset())
      -> TimestampFilterMySqlStreamingChangeEventSource is used only for the bootstrap timestamp case
      -> normal MySqlStreamingChangeEventSource is used for restored binlog offsets

Findings

• BinlogOffset.isTimestampOffset() now cleanly separates a bootstrap timestamp placeholder from a restored binlog position.
• MySqlSourceFetchTaskContext.shouldResolveTimestampStartupOffset(...) only re-resolves from configured timestamp for the bootstrap case.
• MySqlBinlogFetchTask.shouldFilterByTimestamp(...) uses the same guard, so the timestamp filter is skipped on checkpoint restore.
• The added tests cover both sides of the behavior change and do not introduce flaky patterns.
• I did not find a new code blocker on the current head.

CI

I also checked the current failed Build on this head. The failing jobs are:

• unit-test (8, ubuntu-latest): JobStateEventTest.testNotEndJobStateEvent in seatunnel-engine-server
• paimon-connector-it (11, ubuntu-latest): PaimonSinkCDCIT.testSinkWithIncompatibleSchema

Those failures are outside the MySQL CDC restore path touched here, so I do not see evidence that this PR introduced them.

Merge conclusion: can merge

1. Blocking items

• No blocking code issue from my side on the latest revision.

2. Suggested follow-up

• Please rerun or resync once the unrelated CI failures are cleared so the branch has a green signal for merge.

Thanks for the focused fix here.

[DanielLeens]

Thanks for the update. I rechecked the latest head 1552d36fc5 against the current gate state again.

From Daniel's side, the earlier source-level approval on this branch still stands for me. The remaining gate is operational:

• base branch: dev
• compare status: diverged
• behind_by: 1
• current failing workflow/job: Pull Request Labeler / Build (cancelled)

Because this branch is behind the latest dev and the remaining blocker is now only the cancelled Build gate, the lowest-cost next step is to sync with the latest dev and rerun Build first. If the refreshed head still fails after the sync, please share that exact new failing check and I can help narrow it down on top of the updated revision.

[DanielLeens]

Thanks for the update. I re-reviewed the latest head for this PR after the new commit.

What this PR fixes

• PR: [Fix][Connector-V2][MySQL CDC] Use checkpoint offset for timestamp startup restore
• User-facing value: keeps the relevant SeaTunnel behavior/docs/tests aligned with the intended change.
• In one sentence: I do not see a reopened source-level blocker on the current head.

Runtime / review path checked

MySQL CDC timestamp startup restore
  -> startup mode chooses timestamp-based offset
  -> checkpointed offset is restored after restart
  -> source resumes from stored checkpoint state rather than recomputing the original timestamp offset

Current review findings

• Changed files in the current full diff: 5.
• Source-level conclusion: no new blocker from Daniel's side on the latest head.
• Compatibility: I did not find an API/config/default/protocol/serialization break in the reviewed path.
• Test stability: I did not see a new flaky-test pattern in the files touched by this PR during this static review.
• Base sync fact recorded: compare status ahead, behind_by=0. This is recorded as queue metadata; I am not turning it into a standalone sync request unless it is needed to interpret a failing CI signal.

Testing and local verification

• Fetched the PR head into a local review branch named seatunnel-review-10987.
• Checked the current full diff, changed-file list, and latest commit list against dev.
• I did not run local tests or a local build; this round is based on static diff and call-path review.

Merge conclusion

Conclusion: can merge after CI

Blocking items: none from Daniel's current source review.
Non-blocking suggestions: none from this round.

Current CI is PENDING; I am not approving until the required checks finish green.