Vivek1106-04 opened a new pull request, #12165:
URL: https://github.com/apache/seatunnel/pull/12165
### Purpose of this pull request
Closes #12122.
Every `CheckpointCoordinator` built its own two-thread scheduled pool
(`CheckpointCoordinator.java:227-238`), and `CheckpointManager` builds one
coordinator per entry of `checkpointPlanMap`
(`CheckpointManager.java:119-168`). Timer threads therefore grew with the
number of active checkpoint-enabled pipelines: a member running 500 of them
held up to 1000 threads that spend nearly all their time asleep.
The pool was only ever used as a timer. There are two call sites — re-arming
the periodic trigger in `scheduleTriggerPendingCheckpoint()` and arming the
checkpoint-timeout watchdog stored via
`pendingCheckpoint.setCheckpointTimeOutFuture(...)` — while the heavy barrier
work already runs on the injected `executorService` through `thenApplyAsync`.
That is what makes a shared timer safe here.
This PR introduces a member-wide scheduler and gives each coordinator a
per-pipeline lease.
**What changed**
- New `checkpoint/scheduler/SharedCheckpointScheduler`: one instance per
member, owned by `SeaTunnelEngineContext` (built once at
`SeaTunnelServer.java:155`) and closed with it. It is deliberately not a static
singleton, because several members run in one JVM during unit and E2E tests.
- Timing and execution are split. Two fixed timer threads
(`checkpoint-timer-%d`, with `setRemoveOnCancelPolicy(true)`) do nothing but
hand the task to an elastic dispatch pool (`checkpoint-dispatcher-%d`), so a
checkpoint body that blocks on an RPC delays neither its own pipeline's next
timer nor any other pipeline's.
- New `PipelineCheckpointScheduler`: the lease, identified by `(jobId,
pipelineId)`. It holds only its own outstanding tasks, so `cancelAll()` is
scoped to one pipeline and a coordinator no longer shuts down a pool it does
not own. The lease stays usable afterwards, so a coordinator restored after a
master-failover reset schedules again without rebuilding any pool.
- New `PipelineCheckpointTask`: the `ScheduledFuture` handed back to
callers. Because the task crosses two pools, cancelling it has to work both
before the timer fires and after the body has been handed to the dispatch pool;
the body re-checks its own cancellation flag at run time rather than relying on
the timer future, which is already complete by then.
- `CheckpointCoordinator` now leases its timer from `CheckpointManager`
instead of constructing a pool, and `cleanPendingCheckpoint()` cancels this
pipeline's tasks instead of calling `shutdownNow()` and building a replacement.
**Two deliberate choices worth reviewing**
Checkpoint work is *not* dispatched onto the coordinator executor.
`createCoordinatorExecutor()` (`CoordinatorService.java:287-298`) is a bounded
`ThreadPoolExecutor` over a `SynchronousQueue` with a rejection handler, so a
saturated coordinator pool could silently drop a checkpoint trigger or a
timeout watchdog. Keeping a separate dispatch pool also leaves this change
independent of the executor-bounding work in #12117 and #12121.
The timer pool size is a constant rather than a new engine config option.
Adding an `Option` would create a user-facing config contract for something
that should not need tuning, and would require matching `docs/en` and `docs/zh`
updates. Happy to add one if reviewers prefer it.
**Invariants preserved**
The single-pending-checkpoint guard is untouched. The whole trigger path
stays inside `synchronized (lock)`, and the `pendingCounter.get() > 0` guard
still short-circuits a redundant trigger. Coordinators previously already had
two threads that could enter that path concurrently, so sharing the timer adds
no new race: extra dispatch threads queue on the same lock and bail through the
same guard.
Note that the issue text cites this guard as `:802`, which matches the
commit it was written against; on current `dev` line 802 is the re-schedule
call and the guard sits at `:791`.
### Does this PR introduce _any_ user-facing change?
No.
No config option is added, removed, or renamed, and no default changes. The
only externally observable difference is thread naming in stack dumps and
profiles: the per-pipeline `checkpoint-coordinator-<pipelineId>/<jobId>`
threads are replaced by shared `checkpoint-timer-%d` and
`checkpoint-dispatcher-%d` threads. Job and pipeline context is still available
on those threads, because each scheduled body is wrapped with
`MDCTracer.tracing(...)` at schedule time, which is what the previous
`MDCTracer.tracing(scheduler)` wrapper did. No documented behavior depends on
the old thread names.
### How was this patch tested?
New `SharedCheckpointSchedulerTest` covers the lease contract:
- a scheduled body runs on the dispatch pool, not on a timer thread;
- `cancelAll()` on one pipeline's lease leaves a sibling pipeline's timer
running;
- a task cancelled before its timer fires never reaches the dispatch pool;
- a task cancelled after its timer fired still does not run its body, and
cancel is idempotent;
- a lease stays usable after `cancelAll()`, which is the master-failover
reset path;
- finished tasks remove themselves rather than accumulating on a long-lived
pipeline;
- scheduling after the shared scheduler is closed returns an
already-cancelled future and runs nothing.
The acceptance criterion from the issue is asserted directly: 500 leases
each scheduling a task keep the timer pool at 2 threads.
`CheckpointCoordinatorTest` needed one change. Nine tests build coordinators
on `Mockito.mock(CheckpointManager.class)`; a mocked manager returns `null` for
the new `leaseCheckpointScheduler(...)`, so those coordinators would NPE on
cancel. They now go through a helper that hands out real leases from a
test-scoped shared scheduler.
Local results on JDK 11:
- `SharedCheckpointSchedulerTest` — 8 tests, 0 failures
- `CheckpointCoordinatorTest` — 14 tests, 0 failures
- `*Checkpoint*Test` plus `CoordinatorServiceTest` — 83 tests, 0 failures, 3
skipped
- full `seatunnel-engine-server` unit test suite — 513 tests, 0 failures, 0
errors, 5 skipped
- `./mvnw spotless:apply` applied; the only formatting changes were
whitespace in the files this PR already touches
### Rebase note
#12149 is open on the same constructor and the same
`cleanPendingCheckpoint()` lines, and #12096 also touches
`cleanPendingCheckpoint()`. This change is complementary rather than a
duplicate: #12149 fixes the lifecycle of the per-coordinator pool, while this
removes the per-coordinator pool. I kept the diff narrow specifically so that
rebasing is cheap, and I am happy to rebase onto either of those once they
land, or to have this sequenced behind them — whichever reviewers prefer.
### Check list
* [x] 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/developer/new-license.md)
— no new dependencies
* [x] If necessary, please update the documentation to describe the new
feature. https://github.com/apache/seatunnel/tree/dev — no user-facing change,
so no doc update
* [x] If necessary, please update `incompatible-changes.md` to describe the
incompatibility caused by this PR — no incompatible change
* [x] If you are contributing the connector code, please check that the
following files are updated — not a connector change
--
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]
