DanielLeens opened a new pull request, #12141:
URL: https://github.com/apache/seatunnel/pull/12141
### Purpose of this pull request
Add E2E regression coverage for the literal mechanism behind
apache/seatunnel#11473 ("[Bug] [Zeta] stop-job with savepoint hangs in
DOING_SAVEPOINT"), fixed by #11489: a savepoint requested while the source is
busy emitting a very large split must still get its barrier injected and
complete, through the real multi-node path.
### The gap
`#11489` fixed the hang on two ends: `FakeSourceReader#pollNext` now emits
at most `MAX_ROWS_PER_POLL` rows per call and requeues the remainder of the
split, so the checkpoint lock that `SourceFlowLifeCycle#triggerBarrier` needs
is released between batches; and `JobMaster#savePoint` now leaves
`DOING_SAVEPOINT` deterministically when a savepoint cannot complete.
Existing coverage of that fix:
- `SavePointBusySourceTest` (seatunnel-engine-server): reproduces the busy
split, but only inside a single in-process server and by calling
`CoordinatorService#savePoint` directly. The barrier never crosses a node
boundary and no client is involved.
- `SavepointPreconditionRecoveryIT` (#12035): E2E, but it exercises a
different branch of the same fix, the `TASK_NOT_ALL_READY_WHEN_SAVEPOINT`
precondition rejection that restores the job to RUNNING. No busy source and no
starved barrier are involved there.
The #11473 mechanism itself, a barrier that has to be injected into a source
mid-split, with the trigger dispatched over the network by
`CheckpointBarrierTriggerOperation` and the request coming from a real client,
had no E2E coverage.
### What the new test does
`SavepointBusySourceBarrierIT` (connector-seatunnel-e2e-base, same harness
as `SavepointPreconditionRecoveryIT`):
1. Starts a split master/worker cluster and submits the reproduction job
from the issue verbatim (`stream_fake_busy_split_to_console_savepoint.conf`:
one `FakeSource` split of 100,000,000 rows, `split.read-interval = 10`,
`checkpoint.interval = 10000`, `checkpoint.timeout = 60000`, Console sink with
`log.print.data = false`).
2. Waits until `SourceReceivedCount` is positive and still growing between
two samples, i.e. the reader is provably mid-split, then fires a real
`SeaTunnelClient#savePointJob` from a separate thread.
3. Requires the request to complete within 120 s (a `TimeoutException` is
reported as the #11473 hang; an `ExecutionException` is the #11489 failure
exit, which is equally a regression here), then requires `SAVEPOINT_DONE`.
4. Mid-split proof: the `SourceReceivedCount` recorded in the finished-job
history must be at least the last live sample and strictly below the split
size. Pre-fix the barrier could only pass after the whole split drained (or
never, once the checkpoint expired).
5. Restores the job from that savepoint via `restoreExecutionContext`
(`RestoreMode.SAVEPOINT`, which only accepts a savepoint-type checkpoint),
waits for the restored source to emit again, and takes a second
stop-with-savepoint against the restored, still-busy source.
The bound of 120 s deliberately exceeds the job's `checkpoint.timeout` (60
s), so a starved barrier surfaces as a checkpoint expiry and a failed request
instead of a silent hang past the bound.
### Why it is not flaky
- The busy condition is not timing-guessed: at the observed generation rates
(tens of thousands of rows per second) the 100M-row split takes tens of
minutes, so every request lands mid-split, and the test only fires after
metrics show the counter growing.
- All waits are bounded Awaitility polls; metric reads that fail transiently
during deployment/teardown are logged and retried instead of aborting the poll.
- Before the restore, the test waits for the coordinator to release the
finished job id; otherwise `CoordinatorService#submitJob` treats the
still-registered id as already running and silently skips the restore.
### Verification
Local run (macOS, JDK 8, full `-am` install of the upstream modules from
this branch, then `verify -pl connector-seatunnel-e2e-base -DskipUT=true
-DskipIT=false -Dit.test=SavepointBusySourceBarrierIT`), 4 consecutive passes:
| Run | Result | Elapsed |
|---|---|---|
| 1 | Tests run: 1, Failures: 0, Errors: 0 | 17.9 s |
| 2 (INFO logging) | Tests run: 1, Failures: 0, Errors: 0 | 18.0 s |
| 3 | Tests run: 1, Failures: 0, Errors: 0 | 18.2 s |
| 4 (INFO logging) | Tests run: 1, Failures: 0, Errors: 0 | 18.7 s |
Trace from run 4 (checkpoint coordinator / job master at INFO):
```
turned from state SCHEDULED to RUNNING
FakeSourceReader - wait split!
turned from state RUNNING to DOING_SAVEPOINT
save point checkpoint is created, ... checkpoint id: 1
turned from state DOING_SAVEPOINT to SAVEPOINT_DONE
stop-with-savepoint (initial run) ... completed in 737 ms
SourceReceivedCount was 16187 right before the savepoint request and 77824
when the pipeline finished (split size 100000000)
Restore checkpoint, ... checkpointId=1, checkpointType=SAVEPOINT_TYPE
turned from state SCHEDULED to RUNNING
turned from state RUNNING to DOING_SAVEPOINT
save point checkpoint is created, ... checkpoint id: 2
turned from state DOING_SAVEPOINT to SAVEPOINT_DONE
stop-with-savepoint (restored run) ... completed in 236 ms
(restored): SourceReceivedCount was 109138 right before the second savepoint
request and 118784 when the pipeline finished
```
Both barriers landed on a 4096-row batch boundary (77824 = 19 x 4096, 118784
= 29 x 4096) with the split far from drained, which is exactly the lock release
added by #11489. No `timeout checkpoint` / `CHECKPOINT_EXPIRED` events, no
transient metric failures.
`./mvnw spotless:check -pl
seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base` passes.
### Does this PR introduce any user-facing change?
No. Test-only: one new IT class and one new test job config under
`connector-seatunnel-e2e-base/src/test`.
### How was this patch tested?
See Verification above; CI runs the new IT as part of the engine E2E module.
### Check list
* [x] Code changed are covered with tests, or it does not need tests for
reason: test-only change
* [ ] If any new Jar binary package adding in your PR, please add License
Notice according [New License
Guide](https://github.com/apache/seatunnel/blob/dev/docs/en/contribution/new-license.md)
* [ ] If necessary, please update the documentation to describe the new
feature. https://github.com/apache/seatunnel/tree/dev/docs
* [x] If you are contributing the connector code, please check that the
following files are updated:
1. Update change log that in connector document. For more details you can
refer to
[connector-v2](https://github.com/apache/seatunnel/tree/dev/docs/en/connector-v2)
2. Update
[plugin-mapping.properties](https://github.com/apache/seatunnel/blob/dev/plugin-mapping.properties)
and add new connector information in it
3. Update the pom file of
[seatunnel-dist](https://github.com/apache/seatunnel/blob/dev/seatunnel-dist/pom.xml)
* [ ] Update the
[`release-note`](https://github.com/apache/seatunnel/blob/dev/release-note.md).
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
