li3zhi4 opened a new pull request, #11677:
URL: https://github.com/apache/seatunnel/pull/11677
## Why
After a CDC job (e.g. MySQL CDC → a JDBC/Doris sink) fails and the engine
restarts it from the last successful checkpoint, the source can silently stop
producing data: the checkpoint completes, the job stays `RUNNING`, but no
records flow. The reader-side restored splits are handed back to the enumerator
via `RestoredSplitOperation` → `IncrementalSourceEnumerator.addSplitsBack()`,
which only calls `splitAssigner.addSplits(splits)`. When the restored split
arrives **after** the reader already sent its split request, `assignSplits()`
is never re-invoked — the split stays in the assigner, the waiting reader never
receives it, and the job idles forever on empty checkpoints.
This is a different root cause from #10574 (Kafka source restore-offset
logic, fixed by #10612): that one is in the Kafka source; this one is in the
CDC enumerator's restored-split assignment.
## What changed
### `IncrementalSourceEnumerator.java`
`addSplitsBack()` becomes synchronous: after adding the restored splits to
the assigner, if the enumerator is already running, immediately run
`assignSplits()` so the splits are dispatched to the currently waiting reader
and consumption resumes from the checkpointed offset.
```java
@Override
public synchronized void addSplitsBack(List<SourceSplitBase> splits, int
subtaskId) {
LOG.debug("Incremental Source Enumerator adds splits back: {}", splits);
splitAssigner.addSplits(splits);
if (running) {
assignSplits();
}
}
```
Non-goals: no checkpoint-semantics change, no sink stream-load/transaction
change, no change to the transient sink error handling.
## File changes
| File | Change |
|---|---|
|
`connector-cdc/connector-cdc-base/.../source/enumerator/IncrementalSourceEnumerator.java`
| +3 (synchronized + `if (running) assignSplits()`) |
|
`connector-cdc/connector-cdc-base/src/test/.../enumerator/IncrementalSourceEnumeratorTest.java`
| new test `shouldAssignRestoredSplitsToWaitingReader` |
|
`seatunnel-e2e/.../connector-cdc-mysql-e2e/.../mysql/AbstractMysqlCDCITBase.java`
| new e2e `testMysqlCdcContinuesAfterSinkFailure` |
|
`seatunnel-e2e/.../connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_sink_failure_recovery.conf`
| new e2e conf |
## Verification
- **Unit test**:
`IncrementalSourceEnumeratorTest#shouldAssignRestoredSplitsToWaitingReader` —
reader already waiting, restored split must be dispatched immediately.
- **Reproduced on upstream dev (2026-08-06)**: the ported test **fails on
the original code** (`expected: <[SnapshotSplit(tableId=null, ...)]> but was:
<[]>` — restored split never assigned) and **passes after the fix** (`Tests
run: 1, Failures: 0, Errors: 0`, `BUILD SUCCESS`).
- **New official E2E (Zeta + Testcontainers)**:
`MysqlCDCIT#testMysqlCdcContinuesAfterSinkFailure` — completes
snapshot/checkpoint, asserts RUNNING + source/sink consistency, injects a JDBC
sink failure via a temporary MySQL trigger, waits for Zeta auto-restart,
removes the trigger, writes new binlog data, and asserts consistency again
(validated locally on 2.3.13, 1/1 passed, ~95 s; will run on dev CI).
## Checklist
- [x] Code change complete and verified locally on 2.3.13
- [x] Ported onto `dev` + regression test reproduced the race before the fix
and passes after it
- [ ] `mvn spotless:check` passes
- [ ] Related unit tests pass
- [x] Issue(s) referenced: #11676
- [ ] Docs updated if needed (none needed — behavior fix only)
## Note for upstream rebase
`IncrementalSourceEnumerator` on `dev` lives at
`connector-cdc/connector-cdc-base/.../source/enumerator/` — same package
layout; the patch ports as-is. The enumerator API (`addSplitsBack`,
`assignSplits`, `running`) is unchanged on dev.
--
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]
