ChuckLin2025 opened a new pull request, #57462:
URL: https://github.com/apache/spark/pull/57462
### What changes were proposed in this pull request?
A netty worker event-loop thread that dies (a `Throwable` escaping `run()`
at the `runIo()`/select level; per-task exceptions are swallowed by
`safeExecute`) is driven to `ST_TERMINATED` by
`SingleThreadEventExecutor.doStartThread()`'s finally block. Once that happens
the thread is:
- **never replaced** in the fixed-size `MultithreadEventExecutorGroup`
(`children` is final, there is no repopulation),
- still **handed out** by the round-robin `EventExecutorChooser`, which has
no liveness check, and
- **not restartable** (`startThread()` only starts from
`ST_NOT_STARTED`/`ST_SUSPENDED`, never from `ST_TERMINATED`).
So the dead loop permanently poisons any channel pinned to it, which
surfaces in `TransportClientFactory` as two failure modes:
1. **New connections (~1/N fail):** a fresh channel bound to the dead loop
fails registration with `RejectedExecutionException("event executor
terminated")` (caught by `AbstractChannel.AbstractUnsafe.register`), so
`createClient` throws `IOException`. Each connect round-robins across the N
worker threads, so roughly 1 in N attempts binds to the dead loop and fails.
2. **Reused cached client (worse — silent hang):** a pooled
`TransportClient` pinned to the dead loop still has an open socket, so
`isActive()` was true and `createClient` kept returning it.
`writeAndFlush().addListener()` then submits to the dead loop; netty's
`safeExecute` **swallows** the `RejectedExecutionException` (only logs "Failed
to submit a listener notification task. Event loop shut down?"), the
callback/listener is **orphaned**, and the fetch (broadcast/RDD/RPC) **hangs
forever**.
This PR makes the client network stack self-heal in-process, all within
`common/network-common`:
- **`TransportClient.isActive()`** returns `false` when
`channel.eventLoop().isShuttingDown()` is true, so a poisoned pooled client is
no longer treated as active and is not reused — `createClient` creates a new
one instead.
- **`TransportClientFactory.createClient`**, when a connect fails and the
cause chain contains a `RejectedExecutionException` whose message is exactly
`"event executor terminated"` (the terminated-loop rejection only — the
queue-full default handler throws with no message), replaces `workerGroup` with
a fresh group and rethrows, so the existing `IOException` retry path (e.g.
`RetryingBlockTransferor`) reconnects onto a fresh, all-live group.
- `recreateWorkerGroup` is `synchronized` and **identity-guarded**
(`workerGroup != connectGroup` → no-op), so N concurrent callers that all hit
the same dead group swap it exactly once. `workerGroup` is `volatile`.
- The superseded group is **not shut down eagerly** — its still-live
threads may be serving already-open channels. It is retained via a
`WeakReference` and shut down best-effort in `close()`; its threads are daemon,
so a not-yet-collected group cannot block JVM shutdown.
- Gated by a new config `spark.network.recreateWorkerGroupOnDeadEventLoop`,
default `true`.
### Why are the changes needed?
Without this, a single dead netty worker thread degrades the client network
stack for the lifetime of the JVM: new connections fail ~1/N of the time, and —
worse — a reused pooled client submits to the dead loop where the rejection is
swallowed, orphaning the callback so the fetch hangs forever. Only a fresh JVM
fully clears the poison. Recreating the worker group on the terminated-loop
rejection lets the existing retry path recover in-process instead.
### Does this PR introduce _any_ user-facing change?
No. This is an internal reliability fix. It adds an internal-style network
config `spark.network.recreateWorkerGroupOnDeadEventLoop` (default `true`);
when disabled, the previous behavior is preserved. When no event loop dies,
behavior is unchanged.
### How was this patch tested?
New unit tests in `common/network-common`:
- `TransportClientSuite.isActiveFalseWhenEventLoopIsShuttingDown` — a client
whose event loop reports `isShuttingDown()` is not active even though the
channel still reports open/active.
- `TransportClientFactorySuite.recreatesWorkerGroupWhenEventLoopIsDead` —
shutting down the factory's worker group makes the next `createClient` fail
with the terminated-loop rejection; the factory swaps in a fresh live group and
a subsequent connection succeeds.
- `TransportClientFactorySuite.doesNotRecreateWorkerGroupWhenDisabled` —
negative control with the config off: the connect still fails and the worker
group is left unchanged.
`network-common/testOnly TransportClientSuite TransportClientFactorySuite`
passes (12 tests). `core` compiles; `network-common` checkstyle (main + test)
and `core` scalastyle report no issues.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic)
--
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]
