Vivek1106-04 commented on PR #12165:
URL: https://github.com/apache/seatunnel/pull/12165#issuecomment-5776671351
@DanielLeens Both items from your last review are pushed, on `da0cbbf52`,
rebased onto `dev` at `0a5c2cc2f`.
## 1. The watchdog IMap hop, in this PR
The watchdog body was doing the whole expiry inline on a dispatch thread:
`handleCoordinatorError` -> `updateStatus`, which is a
`RetryUtils.retryWithException`
loop over `runningJobStateIMap.get` / `set`, and then
`cleanPendingCheckpoint`, which
takes the coordinator `lock` - the same lock `startSavepoint` can be
holding. So the
expiry path could occupy a member-wide dispatch thread for as long as an
IMap retry
sequence plus a lock wait, which is exactly the shape of `31a454f35`.
Now the dispatch thread keeps only the cheap part:
```java
if (pendingCheckpoints.get(pendingCheckpoint.getCheckpointId()) != null
&& !pendingCheckpoint.isFullyAcknowledged()) {
LOG.info("timeout checkpoint: {}", pendingCheckpoint.getInfo());
expireCheckpoint();
}
```
and `expireCheckpoint()` hands the rest to `executorService`.
One decision I want to put in front of you rather than bury. That executor
is bounded
over a `SynchronousQueue` with an abort handler, which is the reason this PR
does not
schedule checkpoint work there in the first place. For the expiry the
failure mode of a
rejection is worse than the coupling: a dropped expiry leaves the checkpoint
pending
with nothing left to fail it, so the job hangs instead of failing. The
rejection
therefore falls back to running inline on the dispatch thread, with a WARN:
```java
try {
executorService.execute(
() ->
handleCoordinatorError(CheckpointCloseReason.CHECKPOINT_EXPIRED, null));
} catch (RejectedExecutionException e) {
LOG.warn("Coordinator executor rejected the expiry of a checkpoint of
pipeline {}, "
+ "handling it on the checkpoint dispatch thread", pipelineId,
e);
handleCoordinatorError(CheckpointCloseReason.CHECKPOINT_EXPIRED, null);
}
```
That restores the old behaviour for one checkpoint in the case where the
coordinator pool
is already saturated, and removes it in every other case. If you would
rather have the
rejection surface as an error than as a silent fallback, say so and I will
change it.
### Coverage
`CheckpointCoordinatorTest#testCheckpointExpiryHandlingNeverRunsOnTheDispatchThread`
arms a 10 ms timeout on a real lease from the shared scheduler and records
the thread
that reaches `CheckpointManager#handleCheckpointError`.
Red first, with the fix reverted and everything else identical:
```
AssertionFailedError: checkpoint expiry handling must not run on a shared
dispatch
thread, but ran on checkpoint-dispatcher-0 ==> expected: <false> but was:
<true>
```
Green with it, the expiry landing on the coordinator executor instead.
`SharedCheckpointSchedulerTest`
is 10/10 on the same head.
## 2. The corrected docs
The class Javadoc on `SharedCheckpointScheduler` claimed a dispatched body
is always
short-lived because the barrier work runs on `thenApplyAsync`. It now says
what is
actually true after the hop above - atomics, timestamp comparisons, logging
and a map
put in steady state - and then names the savepoint case explicitly:
`startSavepoint`
holds the coordinator lock across its sleep-poll,
`tryTriggerPendingCheckpoint` takes
that same lock on a dispatch thread, and
`CheckpointManager.triggerSavePoints` fans out
over every coordinator of a job at once, so with one shared pool enough
concurrently
savepointing pipelines can delay unrelated pipelines' timers. It states that
moving the
wait off the lock is tracked separately rather than fixed here.
STIP-40 (#12339) carries the same correction, as a marked **Correction**
block rather
than a silent edit, with the #12418 saturation figures attached. I also
amended
"Alternatives Considered" item 1: dispatching the *scheduling* onto the
coordinator
executor stays rejected, but the STIP now says it is the right destination
for an
individual piece of work where a rejection can be caught rather than lost,
which is what
the expiry hop does. And I added the new test to the test plan.
## 3. CI
Rebased onto current `dev`, which matters for two of the red legs. On the
previous head
every failure reproduced on `dev`:
| Leg | Test | Status |
|---|---|---|
| `engine-v2-it` (8 and 11) |
`SplitClusterFaultToleranceIT#testStreamJobCancelResolvesWhenWorkerCrashesBeforeCancelAck`
| #12353, a `dev`-level race; @SEZ9 measured 13 failures in 17 real executions
across three code bases including plain `dev`. Fix is #12311, open and
unreviewed since 09-15 |
| `all-connectors-it-2` (8 and 11) |
`OpengaussCDCIT#testAddFieldWithRestore` | #12344, closed 09-20; the fix
`ea3df1667` (#12346) is in `dev` and was not in my previous head. The rebase
picks it up |
| `rocketmq-connector-it` (11) | `RocketMqIT`, 14 failures and 29 errors |
`dev` has `36c0d1b8a`, `5f7f5c1db` and `b70652be8` since my previous head. Also
picked up by the rebase |
| `all-connectors-it-7` (8 and 11) |
`PostgresCDCIT#testPostgresCdcSnapshotOnlyAndCommittedOffsetStartupModes` |
#11847 / #12382, open CDC offset bugs |
| `all-connectors-it-1` (11) | `NebulaGraphIT` at `startUp` | #12345, JDK-11
leg only, elevated since 09-15 |
| `transform-v2-it-part-1` (11) |
`TestFilterRowKindIT#testFilterRowKindMultiTable[3]` | #12116 |
| `all-connectors-it-4` (8) | none - `org.testcontainers:azure:1.21.4`
download reset | infrastructure |
Nothing there traces to this PR. I expect the Opengauss and RocketMQ legs to
clear on
this run; `SplitClusterFaultToleranceIT` will not until #12311 lands, since
it fails at
that rate on `dev` itself.
## 4. One thing I would like settled before I call this ready
Your review here treats "stop sleeping while holding `lock`" as a tracked
follow-up with
corrected docs, which is what I have implemented. On #12339, about eight
hours earlier,
@SEZ9 restated the gate as still requiring the regression that fills
dispatch capacity
with savepoint-contended coordinators and shows an unrelated pipeline still
receiving its
trigger and watchdog work on time. I do not think those two readings have
been
reconciled, and I would rather ask than pick the convenient one.
If the regression is still required for this PR, I will write it against the
current
design - the honest outcome is that it fails, since the coupling is real and
unfixed -
and then implement `tryLock` plus reschedule inside the dispatch body, which
keeps the
blast radius inside this PR's scheduler rather than changing `dev`'s
savepoint path. Say
which and I will start.
--
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]
