SteNicholas opened a new pull request, #3778: URL: https://github.com/apache/celeborn/pull/3778
### What changes were proposed in this pull request? This PR backports https://github.com/apache/spark/pull/57462 ([SPARK-58292]) to Celeborn's network client stack. 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 push/fetch/RPC **hangs forever**. This PR makes the client network stack self-heal in-process, all within `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 (`retryCreateClient`) 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. - A closed factory never recreates a group: `close()` sets `closed` under the same lock before shutting the worker group down, so a `createClient` racing or following `close()` cannot resurrect the factory and leak threads. - Gated by a new config `celeborn.<module>.io.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 push/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 resolve a correctness bug? <!-- Check if yes. The `correctness` label will be added/removed automatically. --> - [ ] Yes ### Does this PR introduce _any_ user-facing change? <!-- Check if yes. --> - [x] Yes It adds a new network config `celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop` (default `true`), documented in `docs/configuration/network.md`. 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`: - `TransportClientSuiteJ.isActiveFalseWhenEventLoopIsShuttingDown` — a client whose event loop reports `isShuttingDown()` is not active even though the channel still reports open/active. - `TransportClientFactorySuiteJ.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 the built-in retry then succeeds. - `TransportClientFactorySuiteJ.doesNotRecreateWorkerGroupWhenDisabled` — negative control with the config off: the connect still fails and the worker group is left unchanged. - `TransportClientFactorySuiteJ.closeFactoryBeforeCreateClient` — extended to assert that a closed factory does not recreate a worker group. Relying on CI for the full verification run. -- 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]
