andygrove commented on PR #2212:
URL:
https://github.com/apache/datafusion-ballista/pull/2212#issuecomment-5134333623
Thanks for picking this up — the diagnosis is right and the shape of the fix
(bounded wait, scoped to the `ExecutorLost` path, config-gated, re-check before
failing) is the shape I'd want.
*Disclosure: this review was produced with LLM assistance (Claude Code).
Code references were checked against the tree; the Scenario G timing analysis
below is derived from reading the code, not from an executed run.*
## First: there's already a regression test for this, and it's waiting on you
#2026 added an HA chaos harness whose whole purpose was to drive these
fault-tolerance paths on a real multi-process cluster, and #2029 is one of the
three bugs it surfaced. It ships a dedicated reproducer:
`chaos-testing/tests/ha.rs` → Scenario G,
`killing_every_executor_terminates_the_job`, `#[case::aqe_off]` /
`#[case::aqe_on]`, currently:
```rust
#[ignore = "reproduces #2029: killing every executor hangs the job instead
of failing it"]
```
`chaos-testing/README.md` (Finding 3) says explicitly: *"Un-ignore it as the
regression test when #2029 is fixed."* This PR is that fix, so please enable it
here rather than leaving the harness carrying a stale `#[ignore]`.
By my reading it should pass as-is, with no assertion changes:
- `ChaosRun::start(aqe, 2)` uses `executor_timeout_seconds: 5` /
`expire_dead_executor_interval_seconds: 1`, so both SIGKILLed executors are
reaped and `ExecutorLost` posted at ~6s.
- `chaos-testing/src/bin/chaos-scheduler.rs` builds `SchedulerConfig { ..,
..Default::default() }`, so it inherits the new 30s default →
`JobRunningFailed` at ~36s → client errors out, comfortably inside the
scenario's 120s budget.
Suggested follow-through:
1. Remove both `#[ignore]`s, and update the Scenario G doc comment (it
currently reads "Both cases reproduce #2029") plus the README table row and
Finding 3.
2. Plumb the new knob into the harness — `CHAOS_NO_EXECUTORS_GRACE_SECONDS`
in `chaos-scheduler.rs` and a builder on `TestCluster`, mirroring the existing
`executor_timeout_seconds` knob and its "defaults would make every
executor-kill scenario take three minutes" comment. With a ~1s grace, G costs
~7s instead of ~37s. That matters: `rust.yml` runs `cargo test` workspace-wide,
so an un-ignored G lands on every PR's critical path.
3. Tighten the assertion. The comment calls it "deliberately weak: this is a
hang detector". Now that termination is achievable, assert the outcome is an
`Err` whose message names executor loss — that's the contract this PR
advertises, and it's worth locking in.
4. Please double-check the `aqe_on` case specifically. AQE keeps its own
status/graph bookkeeping (`state/aqe/mod.rs:181,265`); the new guard sits on
the shared `ExecutorLost` path so it should apply, but that's the case I'd
least want to assume.
5. The branch may need a rebase — #2026 merged very recently, so
`chaos-testing/` may not be on the head branch at all yet.
Worth being explicit about what Scenario G does **not** cover, because it's
blind to finding 1 below: it kills two executors and asserts only that the job
terminates, so it passes even with duplicate failure events. It also never
exercises the recovery path (executor returns inside the grace window), which
is the actual new behaviour being introduced.
## Review findings
### 1. Duplicate grace timers → N `JobRunningFailed` per job
This misfires in exactly the scenario the PR targets. When the whole cluster
dies at once, `expire_dead_executors` (`scheduler_server/mod.rs:326`) returns
*all* stale heartbeats in a single pass and calls `remove_executor` for each.
Meanwhile `get_alive_executors` (`executor_manager.rs:466`) already filters on
`executor_timeout_seconds` staleness, so by the time the first `ExecutorLost`
is handled every other dying executor is already "not alive". Each of the N
`ExecutorLost` events therefore sees an empty set and arms its own grace timer,
and each timer posts `JobRunningFailed` for every running job.
Per job that means: `record_failed` N times (inflated failure metrics),
`abort_job` logging `"Fail to find job {job_id} in the cache, unable to cancel
tasks"` for every attempt after the first, and `clean_up_failed_job` running N
times with N delayed state cleanups queued. Nothing dedupes.
Cleanest fix is to stop deciding inside the spawned task and let the event
loop decide — which also removes finding 3:
```rust
// in the ExecutorLost arm
if self.state.executor_manager.get_alive_executors().is_empty()
&& !self.no_executor_check_pending.swap(true, Ordering::SeqCst)
{
let sender = event_sender.clone();
let grace =
Duration::from_secs(self.state.config.no_executors_grace_period_seconds);
let lost_at = timestamp_millis();
tokio::spawn(async move {
tokio::time::sleep(grace).await;
let _ = sender.post_event(QueryStageSchedulerEvent::NoExecutorsCheck
{ lost_at }).await;
});
}
```
…with a new arm that clears the flag and does the alive-check plus fail loop
inline. A plain `AtomicBool` on `QueryStageScheduler` is enough if a full event
variant feels heavy.
### 2. Jobs submitted during the grace window get a truncated grace and a
misleading message
A job reaches `Status::Running` as soon as planning succeeds
(`execution_graph.rs:365`) — executor availability doesn't enter into it. So
the comment's claim that "jobs merely queued waiting for their first executor
(autoscaling cold start) are never affected" only holds for jobs submitted
*before* the timer was armed. A job submitted at t=29s of a 30s window is
failed ~1s later, told that no executor re-registered within 30s. That's both a
wrong message and the opposite of the autoscaling tolerance the PR is aiming
for.
Capturing `lost_at` (as above) and skipping jobs whose `running.started_at >
lost_at` handles this precisely.
### 3. Snapshot-then-post race (low severity, but `grace = 0` is a
documented setting)
The task snapshots the running-job cache and then `await`s a `post_event`
per job. With `grace = 0`, `sleep(0)` merely yields, so a `JobFinished` already
queued behind the `ExecutorLost` can be processed after the snapshot but before
the `JobRunningFailed` lands. The job then takes `record_failed` and
`clean_up_failed_job` → `clean_up_job_data` despite having succeeded. Moving
the check into the event loop makes this impossible; otherwise re-verify
`job_info.status` is still `Running` immediately before posting.
### 4. Test coverage
- **Both new tests use `with_no_executors_grace_period_seconds(0)`**, so the
grace behaviour — the core of the new logic — is never exercised. The
highest-value missing test: grace > 0, lose the only executor, re-register one
inside the window, assert the job is *not* failed. `#[tokio::test(start_paused
= true)]` keeps it fast.
- Nothing asserts the failure reason reaches the client. Asserting the
`FailedJob` message mentions executor loss locks in the user-visible half of
this fix.
- Finding 1 is unit-testable and cheap: `TestMetricsCollector::job_events`
(`test_utils.rs:716`) lets you assert exactly one `MetricEvent::Failed` per job
after losing N executors at once.
- `test_running_job_survives_partial_executor_loss` leans on a fixed
`tokio::time::sleep(500ms)`; a paused clock is both faster and less of a timing
guess.
- "Wait until the job is actually running with tasks in flight" over-claims
— `Running` is set at planning time, so the condition can be satisfied before
any task is launched.
### 5. Nits
- `_ => timestamp_millis()` for `queued_at` is unreachable:
`get_running_job_cache` already filters to `Running` (`task_manager.rs:398`).
An `else { continue }` reads better. (`scheduler_server/mod.rs:222` does the
same thing, so this is cosmetic.)
- `for pair in ... { let (job_id, job_info) = pair; }` → `for (job_id,
job_info) in ...`.
- `fail_message` starts with `"Job {job_id} failed: "`, but the handler
already logs `"Job failed: [{job_id}]"` and other producers
(`execution_graph.rs:430`) don't prefix the id — dropping it reads better in
the client-facing error.
- Comment convention in this tree leans toward the full URL (`// See
https://github.com/apache/datafusion-ballista/issues/2029`) over a bare `#2029`.
- Naming: the neighbouring field is `executor_termination_grace_period`.
Something like `executor_loss_grace_period_seconds` sits closer to both the
trigger and existing naming, though `no_executors_...` is fine if you prefer it.
- The spawned timer isn't tied to scheduler shutdown, so a stopping
scheduler can hold a task for up to `grace`. Harmless, just worth knowing.
No security concerns, and the performance cost is negligible (one
`get_alive_executors()` snapshot per `ExecutorLost`).
**Overall:** worth landing. I'd want finding 1 fixed before merge since it
fires on the primary scenario, finding 2 is a small addition on the same
change, and I'd like either the grace-period recovery unit test or un-ignored
Scenario G (ideally both) so the new config isn't shipping untested.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
