Repository navigation
[SPARK-58292][CORE] Recreate the netty worker EventLoopGroup when a worker event loop thread dies - #57462
Closed
ChuckLin2025 wants to merge 4 commits into
Closed
[SPARK-58292][CORE] Recreate the netty worker EventLoopGroup when a worker event loop thread dies#57462ChuckLin2025 wants to merge 4 commits into
ChuckLin2025 wants to merge 4 commits into
Conversation
Contributor
Author
|
Hi @cloud-fan @Ngone51 could you help me review this PR ? |
cloud-fan
approved these changes
Jul 23, 2026
cloud-fan
left a comment
Contributor
There was a problem hiding this comment.
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
commented
Jul 24, 2026
Ngone51
approved these changes
Jul 24, 2026
Contributor
Author
|
Hi @cloud-fan @Ngone51 — a heads-up on the CI status on this PR. The failing checks are all downstream of one infra failure, not the code change. The This cascades into every I believe this is a fork-runner / buildx environment issue, not a problem with this PR:
All the checks that actually exercise this change are passing: Could you advise on how you'd like to proceed? A few options I see:
Thanks! |
…orker event loop thread dies When a netty worker event-loop thread dies, it is never replaced in the fixed-size MultithreadEventExecutorGroup, cannot be restarted, and keeps being handed out by the round-robin chooser. This poisons the TransportClientFactory worker group so a broadcast/RDD/RPC fetch can hang forever. - TransportClient.isActive() returns false when the event loop is shutting down, so a poisoned pooled client is no longer reused. - TransportClientFactory.createClient recreates the worker EventLoopGroup on a "event executor terminated" rejection (identity-guarded, done once) and rethrows so the existing IOException retry path binds onto a fresh group. - Gated by spark.network.recreateWorkerGroupOnDeadEventLoop (default true). Co-authored-by: Isaac
…nct thread-name prefix Address review feedback: tag the recreated worker EventLoopGroup with a "-client-recreated-<n>" thread-name prefix so the dead-event-loop recovery is self-identifying in thread dumps and logs. Netty's DefaultThreadFactory already appends an incrementing pool id so names would not collide, but the explicit suffix makes the recreation obvious, and the <n> counter distinguishes successive recreations if a loop dies more than once. Co-authored-by: Isaac
… new network config SparkConfigBindingPolicySuite enforces that every ConfigEntry declares a bindingPolicy. spark.network.recreateWorkerGroupOnDeadEventLoop is a transport-layer config read once at TransportClientFactory construction, not a session/SQL binding, so NOT_APPLICABLE is the correct policy. Co-authored-by: Isaac
ChuckLin2025
force-pushed
the
SPARK-58292-recreate-eventloop
branch
from
July 27, 2026 02:34
434304e to
161a8b9
Compare
Contributor
Author
Thanks @Ngone51. Addressed by sync and retry. |
cloud-fan
reviewed
Jul 27, 2026
cloud-fan
left a comment
Contributor
There was a problem hiding this comment.
0 addressed, 0 remaining, 2 new. (0 newly introduced, 2 late catches, 0 previously raised.)
1 blocking, 0 non-blocking, 1 nit.
The recovery mechanism is coherent, but its replacement path currently breaks the factory's closed lifecycle by allowing a later create attempt to install a fresh live worker group.
Design / architecture (1)
- common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java:341: Preserve the factory's closed lifecycle state when handling a terminated event loop.
close()shuts down the current group, so a latercreateClient()fails with the same terminated-executor signal and this path replaces it with a fresh live group. Add an explicit closed-state guard (and a regression assertion that repeated creates after close cannot recreate the group). -- see inline
Nits: 1 minor item (see inline comments).
Verification
Traced pooled-client reuse, connection registration failure, cause-chain matching, synchronized group replacement, and factory shutdown. The existing close-before-create path reaches the same terminated-executor rejection used as the recovery trigger, and there is no closed-state guard before replacement. All three mechanical scanners completed their manifests; no tests were run as part of this review.
| // terminated"), the worker group is permanently degraded: that dead loop is never replaced | ||
| // and keeps being handed out by the round-robin chooser. Replace the group so retries bind | ||
| // to fresh live threads, then rethrow so the caller (e.g. RetryingBlockTransferor) retries. | ||
| recreateWorkerGroupIfEventLoopDead(connectGroup, cf.cause()); |
Contributor
There was a problem hiding this comment.
Keep the factory closed on this path. After close() shuts down workerGroup, a later createClient() fails with this same terminated-executor cause and recreates a fresh live group, so the closed factory is resurrected and can leak threads. Please guard replacement with explicit closed state and extend closeFactoryBeforeCreateClient to assert the group is never replaced.
Contributor
Author
There was a problem hiding this comment.
Done. Add a closed signale to guard this.
…losed factory Address review feedback: after close() shuts down the worker group, a later createClient() would fail with the same "event executor terminated" cause and recreate a fresh live group, resurrecting the closed factory and leaking its threads (close() never reaps them again). Guard recreateWorkerGroup with a closed flag, set under the same lock in close() before the group is shut down, so a concurrent or later createClient() cannot recreate the group. Extend closeFactoryBeforeCreateClient to assert the worker group is never replaced. Also fix a wording nit in the recreation test comment. Co-authored-by: Isaac
cloud-fan
approved these changes
Jul 29, 2026
cloud-fan
left a comment
Contributor
There was a problem hiding this comment.
2 addressed, 0 remaining, 0 new.
0 blocking, 0 non-blocking, 0 nits.
The previous lifecycle defect and wording nit are addressed; the current recovery, concurrency, configuration, and test contracts are coherent.
Verification
Reviewed all six changed files and traced cached-client selection, failed channel registration, class-and-message cause matching, volatile publication, synchronized identity-guarded replacement, close/recreate ordering, superseded-group cleanup, and both configuration branches. The new close guard shares the recreation monitor, so either replacement completes before close and the new group is shut down, or close marks the factory first and later recreation returns. Mechanical contract, link, and text scanners covered their complete manifests. No tests were run as part of this review.
cloud-fan
pushed a commit
that referenced
this pull request
Jul 29, 2026
…orker event loop thread dies
### 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)
Closes #57462 from ChuckLin2025/SPARK-58292-recreate-eventloop.
Authored-by: ChuckLin2025 <lzequn@gmail.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>
(cherry picked from commit 2c8570a)
Signed-off-by: Wenchen Fan <wenchen@databricks.com>
Contributor
1 of 2 tasks
SteNicholas
added a commit
to apache/celeborn
that referenced
this pull request
Aug 11, 2026
…r event loop thread dies This ports SPARK-58292 (apache/spark#57462) to Celeborn's network client stack, and extends it to invalidate clients that are already poisoned. A netty event-loop thread that dies (e.g. an uncaught error) is never replaced within a fixed-size EventLoopGroup, and the round-robin chooser keeps handing it out. Any channel pinned to the dead loop can no longer send or complete anything: writes and listener notifications route to the dead thread and are silently dropped, so a request on such a client can hang forever, and new connections registered on it fail with "event executor terminated". This makes the client recover from that state. Not reusing a poisoned client: - TransportClient.isActive() returns false once the channel's event loop is shutting down, so the pool evicts the client instead of reusing it. Unblocking whoever already holds one: - TransportClient.sendRpc() fails the callback up front rather than writing into a dead loop. Unlike pushes and fetches, an outstanding RPC has no timeout checker to fall back on, so an orphaned callback there hangs its owner forever. - TransportClientFactory, once a connection failure has revealed a dead loop, synchronously fails the outstanding requests of every pooled client still pinned to one. Marking a client inactive does not help an owner that keeps it for the lifetime of a stream -- e.g. Flink's CelebornBufferStream, which sends credits on the client it captured rather than reacquiring one -- because the dead loop delivers neither the write listener nor channelInactive(). The client cannot be force-closed either, since close() is itself submitted to the dead loop. The sweep runs outside the connection-pool lock, as failing a request invokes its callback on the calling thread and callbacks re-enter the factory. Restoring the worker group: - TransportClientFactory replaces its worker group when a connection fails with the terminated-executor rejection, then reconnects inline on the fresh group. The reconnect is deliberately not charged to celeborn.<module>.io.maxRetries, which may be as low as 1 and sleeps io.retryWait between attempts, and it also covers callers with no retry wrapper such as createUnmanagedClient. - The superseded group is not shut down eagerly, since its still-live threads may serve already-open channels. Its channels are tracked and the group is shut down once they drain, with close() as the backstop. It cannot be left to the GC: a netty thread keeps its executor, and the executor its parent group, strongly reachable. - Adds celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop (default true). It gates the group recreation and its channel bookkeeping only; refusing to reuse or write to a dead loop is unconditional, since such a request cannot succeed either way. Also logs the failure cause in CelebornBufferStream, which previously printed e.getCause() (null for these failures) or dropped the throwable entirely, leaving the recovery undiagnosable in exactly the scenario this change addresses. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
SteNicholas
added a commit
to SteNicholas/celeborn
that referenced
this pull request
Aug 11, 2026
…r event loop thread dies Ports SPARK-58292 (apache/spark#57462) to Celeborn's network client stack, and extends it to invalidate clients that are already poisoned. A netty event-loop thread that dies (e.g. an uncaught error) is never replaced within a fixed-size EventLoopGroup, and the round-robin chooser keeps handing it out. Any channel pinned to the dead loop can no longer send or complete anything: writes and listener notifications are silently dropped, so a request on such a client can hang forever, and new connections registered on it fail with "event executor terminated". The client now recovers from that state: - TransportClient.isActive() returns false once the channel's event loop is shutting down, so the pool evicts the client instead of reusing it. - TransportClient.sendRpc() fails the callback up front rather than writing into a dead loop. Unlike pushes and fetches, an outstanding RPC has no timeout checker to fall back on. - TransportClientFactory, once a connection failure has revealed a dead loop, fails the outstanding requests of every pooled client still pinned to one. Marking a client inactive does not help an owner that keeps it for the lifetime of a stream -- e.g. Flink's CelebornBufferStream -- because the dead loop delivers neither the write listener nor channelInactive(). - TransportClientFactory replaces its worker group on the terminated-executor rejection and reconnects inline on the fresh group. The superseded group is retired once its channels drain, with close() as the backstop. - Adds celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop (default true). It gates the group recreation only; refusing to reuse or write to a dead loop is unconditional. Also logs the failure cause in CelebornBufferStream, which previously printed e.getCause() (null for these failures) or dropped the throwable entirely.
SteNicholas
added a commit
to SteNicholas/celeborn
that referenced
this pull request
Aug 11, 2026
…r event loop thread dies Ports SPARK-58292 (apache/spark#57462) to Celeborn's network client stack, and extends it to invalidate clients that are already poisoned. A netty event-loop thread that dies (e.g. an uncaught error) is never replaced within a fixed-size EventLoopGroup, and the round-robin chooser keeps handing it out. Any channel pinned to the dead loop can no longer send or complete anything: writes and listener notifications are silently dropped, so a request on such a client can hang forever, and new connections registered on it fail with "event executor terminated". The client now recovers from that state: - TransportClient.isActive() returns false once the channel's event loop is shutting down, so the pool evicts the client instead of reusing it. - TransportClient.sendRpc() fails the callback up front rather than writing into a dead loop. Unlike pushes and fetches, an outstanding RPC has no timeout checker to fall back on. - TransportClientFactory, once a connection failure has revealed a dead loop, fails the outstanding requests of every pooled client still pinned to one. Marking a client inactive does not help an owner that keeps it for the lifetime of a stream -- e.g. Flink's CelebornBufferStream -- because the dead loop delivers neither the write listener nor channelInactive(). - TransportClientFactory replaces its worker group on the terminated-executor rejection and reconnects inline on the fresh group. The superseded group is retired once its channels drain, with close() as the backstop. - Adds celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop (default true). It gates the group recreation only; refusing to reuse or write to a dead loop is unconditional. Also logs the failure cause in CelebornBufferStream, which previously printed e.getCause() (null for these failures) or dropped the throwable entirely.
SteNicholas
added a commit
to SteNicholas/celeborn
that referenced
this pull request
Aug 11, 2026
…r event loop thread dies Ports SPARK-58292 (apache/spark#57462) to Celeborn's network client stack, and extends it to invalidate clients that are already poisoned. A netty event-loop thread that dies (e.g. an uncaught error) is never replaced within a fixed-size EventLoopGroup, and the round-robin chooser keeps handing it out. Any channel pinned to the dead loop can no longer send or complete anything: writes and listener notifications are silently dropped, so a request on such a client can hang forever, and new connections registered on it fail with "event executor terminated". The client now recovers from that state: - TransportClient.isActive() returns false once the channel's event loop is shutting down, so the pool evicts the client instead of reusing it. - TransportClient.sendRpc() fails the callback up front rather than writing into a dead loop. Unlike pushes and fetches, an outstanding RPC has no timeout checker to fall back on. - TransportClientFactory, once a connection failure has revealed a dead loop, fails the outstanding requests of every pooled client still pinned to one. Marking a client inactive does not help an owner that keeps it for the lifetime of a stream -- e.g. Flink's CelebornBufferStream -- because the dead loop delivers neither the write listener nor channelInactive(). - TransportClientFactory replaces its worker group on the terminated-executor rejection and reconnects inline on the fresh group. The superseded group is retired once its channels drain, with close() as the backstop. - Adds celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop (default true). It gates the group recreation only; refusing to reuse or write to a dead loop is unconditional. Also logs the failure cause in CelebornBufferStream, which previously printed e.getCause() (null for these failures) or dropped the throwable entirely.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
A netty worker event-loop thread that dies (a
Throwableescapingrun()at therunIo()/select level; per-task exceptions are swallowed bysafeExecute) is driven toST_TERMINATEDbySingleThreadEventExecutor.doStartThread()'s finally block. Once that happens the thread is:MultithreadEventExecutorGroup(childrenis final, there is no repopulation),EventExecutorChooser, which has no liveness check, andstartThread()only starts fromST_NOT_STARTED/ST_SUSPENDED, never fromST_TERMINATED).So the dead loop permanently poisons any channel pinned to it, which surfaces in
TransportClientFactoryas two failure modes:RejectedExecutionException("event executor terminated")(caught byAbstractChannel.AbstractUnsafe.register), socreateClientthrowsIOException. Each connect round-robins across the N worker threads, so roughly 1 in N attempts binds to the dead loop and fails.TransportClientpinned to the dead loop still has an open socket, soisActive()was true andcreateClientkept returning it.writeAndFlush().addListener()then submits to the dead loop; netty'ssafeExecuteswallows theRejectedExecutionException(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()returnsfalsewhenchannel.eventLoop().isShuttingDown()is true, so a poisoned pooled client is no longer treated as active and is not reused —createClientcreates a new one instead.TransportClientFactory.createClient, when a connect fails and the cause chain contains aRejectedExecutionExceptionwhose message is exactly"event executor terminated"(the terminated-loop rejection only — the queue-full default handler throws with no message), replacesworkerGroupwith a fresh group and rethrows, so the existingIOExceptionretry path (e.g.RetryingBlockTransferor) reconnects onto a fresh, all-live group.recreateWorkerGroupissynchronizedand identity-guarded (workerGroup != connectGroup→ no-op), so N concurrent callers that all hit the same dead group swap it exactly once.workerGroupisvolatile.WeakReferenceand shut down best-effort inclose(); its threads are daemon, so a not-yet-collected group cannot block JVM shutdown.spark.network.recreateWorkerGroupOnDeadEventLoop, defaulttrue.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(defaulttrue); 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 reportsisShuttingDown()is not active even though the channel still reports open/active.TransportClientFactorySuite.recreatesWorkerGroupWhenEventLoopIsDead— shutting down the factory's worker group makes the nextcreateClientfail 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 TransportClientFactorySuitepasses (12 tests).corecompiles;network-commoncheckstyle (main + test) andcorescalastyle report no issues.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic)