[SPARK-58292][CORE] Recreate the netty worker EventLoopGroup when a worker event loop thread dies

[ChuckLin2025]

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)

[ChuckLin2025]

Hi @cloud-fan @Ngone51 could you help me review this PR ?

[cloud-fan]

0 blocking, 0 non-blocking, 0 nits.
The recovery path, lifecycle handling, configuration contract, concurrency guard, and focused tests are coherent; I found no actionable issues.

Verification

Reviewed all six changed files and traced pooled-client reuse, failed channel registration, cause-chain matching, synchronized group replacement, superseded-group shutdown, and both configuration branches. Mechanical link, text, and contract scanners completed their full manifests. No tests were run as part of this review.

[ChuckLin2025]

Do we need a unique name here ?

[Ngone51]

Shall we always print a log for this error even if recreation is disabled? I think it'd be good for visibility.

[ChuckLin2025]

My feeling is that for other case, we should rely on outer exception handler. It's a common pattern that only log the error on the outermost exception handler. But the outermost exception have the more stace trace information. The infrastruction layer logs the exception and then re-throw which is not common and usually redudant.

Also we don't need to care about the !recreateWorkerGroupOnDeadEventLoop case, because it's default on.