andygrove opened a new issue, #2158: URL: https://github.com/apache/datafusion-ballista/issues/2158
## Overview The Spice AI fork of Ballista ([spiceai/datafusion-ballista](https://github.com/spiceai/datafusion-ballista)) has accumulated a set of scheduler and executor reliability fixes that are directly relevant to distributed execution stability, most notably the executor heartbeat/hang failures tracked in #2144. Phillip LeBlanc (Spice) noted on #2144 that they have fixes still to be upstreamed. This epic tracks investigating each change, validating it against our current `main` (the fork sits on a diverged base, so nothing is directly cherry-pickable), and porting the ones we want. Several already live on `upstream/*`-prefixed branches in the fork, indicating Spice intends to contribute them. ## Priority: fixes for the #2144 hang These two together are the core of the q9/q18 hang (executors go idle, then get heartbeat-timed-out fleet-wide): - [ ] **Keep heartbeating while all task slots are busy.** Under pull-based scheduling `poll_work` is the executor's only heartbeat, but the poll loop blocked on `available_task_slots.acquire().await` before polling. When every slot holds a task running longer than `executor_timeout`, the executor stops polling, the scheduler marks the healthy-but-busy executor dead and resets its tasks — a `reset -> re-dispatch -> reset` livelock. Fix caps the slot wait and polls anyway with zero free slots. (Fork: `fix(executor): keep heartbeating while all task slots are busy`) - [ ] **Ignore stale `TaskStatus` updates for reset partitions.** `RunningStage::update_task_info` unwrapped `task_infos[partition_id]`, which is `None` after a task was reset. A late in-flight status update from a lost executor panics the scheduler event-loop worker, closes the event channel, and wedges the whole scheduler ("Fail to send event due to channel closed") — explains the synchronized fleet-wide executor removal. Fix treats `None` as a stale update and ignores it. (Fork: `fix(scheduler): ignore stale TaskStatus updates for reset partitions`) ## Stuck-job revival - [ ] **Periodic reconciliation sweep for pull-based stage revival.** Pull-based scheduling resolves downstream stages only via the event-driven path (no periodic revive sweep like push mode), so a single lost/raced revival wedges a job forever (Running, no available tasks). Adds a background sweep (~10s) that re-runs `update_job` on running jobs; idempotent no-op in the common case. (Fork: `fix(scheduler): periodic reconciliation sweep for pull-based stage revival`) ## Shuffle-fetch reliability - [ ] **Pool shuffle-fetch clients per peer.** `fetch_partition_remote` opened a fresh gRPC + TLS connection per partition fetch; a distributed shuffle issues thousands of concurrent handshakes that abort under CPU load ("connection reset by peer"). Cache one multiplexed `BallistaClient` per `(host, port, tls)` peer. (Fork: `fix(shuffle): pool shuffle-fetch clients per peer instead of dialing per fetch`) - [ ] **Treat a missing partition file as an empty partition.** The writer creates partition files lazily, so a 0-row partition has no file; the reader failed the stage with "No such file or directory" -> `FetchPartitionError` -> unrecoverable resubmit loop. Map a missing file to an empty stream across flight server, client, and local fetch. (Fork: `fix(shuffle): treat a missing partition file as an empty partition`) ## Scheduler responsiveness (already on the fork's `upstream/*` branches) - [ ] **`poll_now_notify` + `on_work_available` callback.** Wake an idle executor immediately via `Arc<Notify>` instead of waiting for the next poll interval. (Fork: `upstream/poll-now-notify-on-work-available`) - [ ] **External semaphore for executor task concurrency.** Let the caller supply the task-slot semaphore so busy-state is observable and shareable across poll loops to different schedulers. (Fork: `upstream/external-semaphore-busy-state`) - [ ] **Broadcast channel for job state events.** `tokio::broadcast` of job state transitions, subscribable via `subscribe_job_updates()`. (Fork: `upstream/job-state-broadcast-channel`) ## Cross-scheduler recovery & status correctness - [ ] **Serialize execution graphs for cross-scheduler recovery.** Persist a job's graph to shared storage so another scheduler can `recover_job` and resume it. (Fork: `feat(scheduler): serialize execution graphs for cross-scheduler recovery`) - [ ] **Persist terminal job status before removing it from the active cache.** `succeed_job` evicted the graph before the `save_job` write landed, so a concurrent `get_job_status` returned stale `Running` and the client's poll timed out on an already-succeeded job. (Fork: `fix(cluster): persist terminal job status before removing it from the active cache`) ## Investigate further (lower priority / experimental) - [ ] In-memory shuffle, broadcast-join enablement, and "increase task parallelism to total available slots" (Peasee WIP branches) — evaluate whether any are worth pursuing independently. - [ ] Diagnostics from `phillip/diag-pollwork-latency` (heartbeat send-gap vs call-duration, scheduler handler latency) — useful to port temporarily while validating the fixes above. Their notes report that CPU starvation was ruled out as the cause, which is consistent with the slot-acquire block being the real culprit. ## Notes - The fork's `main` is behind apache `main`; the fixes live on feature branches on a diverged base, so each needs to be re-derived against current upstream rather than cherry-picked. - Each ported change should land as its own focused PR with the reliability rationale preserved from the fork commit messages. - Credit to Phillip LeBlanc and the Spice AI team for the original fixes. Related: #2144 -- 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]
