[CELEBORN-2400] Recreate the netty worker EventLoopGroup when a worker event loop thread dies

[SteNicholas]

What changes were proposed in this pull request?

This ports SPARK-58292 (apache/spark#57462, merged as 2c8570a1d1; the PR shows as Closed because Spark merges via merge_spark_pr.py) to Celeborn's network client stack, and extends it to also rescue clients that are already poisoned.

The netty behaviour being worked around

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 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 lands on the dead loop.
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 rejection, the listener is orphaned, and the push/fetch/RPC hangs forever.

The change

Not reusing a poisoned client

• TransportClient.isActive() returns false once the channel's event loop is shutting down, so the pool stops handing the client out and createClient builds a new one instead.

Unblocking whoever already holds one

Evicting from the pool does nothing for an owner that keeps a client for the lifetime of a stream and never reacquires it — e.g. Flink's CelebornBufferStream, which calls addCredit()/notifyRequiredSegment() on the TransportClient it captured. The dead loop delivers neither the write listener nor channelInactive(), and the client cannot be force-closed either (close() is itself submitted to the dead loop), so nothing would ever complete those callbacks.

• TransportClient.sendRpc() fails the callback up front instead of writing into a dead loop. Unlike pushes and fetches, an outstanding RPC has no timeout checker to fall back on (failExpiredPushRequest/failExpiredFetchRequest cover outstandingPushes/outstandingFetches only).
• TransportClientFactory, once a connection failure has revealed a dead loop, synchronously fails the outstanding requests of every pooled client still pinned to one. The sweep runs outside the connection-pool lock: failing a request invokes its callback on the calling thread, and callbacks re-enter the factory.

Restoring the worker group

• 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 reconnects inline on it. 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; and a closed factory never recreates one (close() sets closed under the same lock before shutting the group down).
• The superseded group is not shut down eagerly — its still-live threads may be serving already-open channels. Its channels are tracked and the group is shut down once they drain, with close() as the backstop.
• Gated by a new config celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop, default true.

What differs from SPARK-58292

Celeborn's client stack has properties Spark's does not, so a straight port would not have been enough:

	SPARK-58292	This PR
Recovering the failed connect	rethrows IOException and relies on RetryingBlockTransferor to reconnect	reconnects inline, because celeborn.<module>.io.maxRetries may be as low as 1 (leaving no retry to spend on the replacement), it sleeps io.retryWait between attempts, and createUnmanagedClient has no retry wrapper at all
Clients a caller already holds	not addressed	sendRpc() fails fast, plus a sweep that fails the outstanding requests of pooled clients pinned to dead loops — needed because Flink's CelebornBufferStream holds one client for a stream's lifetime and never reacquires
Retiring the superseded group	WeakReference, shut down best-effort in close()	channels tracked, group shut down deterministically once drained (close() as backstop). It cannot be left to the GC: a netty thread keeps its executor and the executor its parent group strongly reachable, so a dropped group would retain clientThreads() - 1 selector threads for the factory's lifetime and accumulate across repeated recoveries

Why are the changes needed?

A single dead netty worker thread degrades the client network stack for the lifetime of the JVM, and in Celeborn the blast radius is wider than in Spark:

• A push or fetch on a reused pooled client hangs until its own timeout checker fires; an RPC has no such checker and hangs indefinitely, taking its caller with it.
• A Flink CelebornBufferStream that captured the client keeps sending credits into the dead loop and is never told anything failed, so the read stalls with nothing surfaced to the job.
• createUnmanagedClient, which has no retry wrapper, fails outright on roughly 1 in N attempts.

Only a fresh JVM fully clears the poison. Recreating the worker group on the terminated-loop rejection, and failing the requests that can no longer complete, lets the client recover in-process instead.

Does this PR resolve a correctness bug?

• Yes

Left unchecked: the failure mode is liveness/reliability — requests hang or connections fail, but no result is ever wrong.

Does this PR introduce any user-facing change?

• Yes

It adds a new network config celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop (default true), documented in docs/configuration/network.md.

Note on the config's scope: it gates the worker-group recreation and its channel bookkeeping only. Refusing to reuse a client on a dead loop (isActive()) and refusing to write to one (sendRpc()) are unconditional, since such a request cannot succeed either way. When no event loop dies, behavior is unchanged.

Two further behaviour notes:

• A best-effort message guarded by isActive() — e.g. the BUFFER_STREAM_END a reader sends on close — is now skipped while the owning factory is closing, because isActive() keys on isShuttingDown() rather than the exact isShutdown() that netty's rejection uses. The server reclaims those streams when the connection drops. This is deliberate: refusing new work on a group that is going away is what we want, and it keeps the dead-loop sweep testable.
• Conversely, WorkerPartitionReader already reacquires its client when isActive() is false, so the Spark fetch path now transparently drops a client pinned to a dead loop.

How was this patch tested?

New TransportClientSuiteJ in common:

• isActiveFalseWhenEventLoopIsShuttingDown — a client whose event loop reports isShuttingDown() is not active even though the channel still reports open/active.
• sendRpcFailsFastWhenEventLoopIsDead — the callback is failed rather than left outstanding, and nothing is handed to the dead loop.
• invalidatesClientWhenEventLoopIsDead — a healthy client is left alone; a dead-loop client has its outstanding requests failed synchronously; the operation is idempotent.

New in 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 connection then succeeds.
• recreatedWorkerGroupIsUsedWithoutConsumingTheRetryBudget — with io.maxRetries=1 there is no retry left to spend, and createClient still succeeds.
• createUnmanagedClientRecoversFromDeadEventLoop — the path with no retry wrapper recovers too.
• failsOutstandingRequestsOfPooledClientsOnDeadEventLoops — a pooled client with an in-flight RPC has it failed by the sweep. Shutting the worker group down only approximates a dead loop: it closes the group's channels and delivers channelInactive(), which would fail the request by itself, whereas a dead loop does neither. The test therefore registers the request once the group has terminated, so nothing but the sweep is left that can complete it.
• retiresSupersededWorkerGroupWithNoChannelsLeft — the superseded group is retired at once rather than retained until close().
• doesNotRecreateWorkerGroupWhenDisabled — negative control with the config off: the connect still fails, the worker group is unchanged, and no retirement bookkeeping runs.
• closeFactoryBeforeCreateClient — extended to assert that a closed factory does not recreate a worker group.

TransportClientFactorySuiteJ builds its own TransportContext in these tests, so a newCelebornConf() hook was added and overridden in SSLTransportClientFactorySuiteJ; otherwise the inherited dead-loop tests would run a plain client against the subclass's SSL servers.

Two behaviours are not covered, both because they cannot be constructed deterministically: close() shutting down a superseded group that never drained, and concurrent callers swapping the same dead group exactly once.

[SteNicholas]

Ping @pan3793, @cxzl25, @RexXiong, @sunchao.

[sunchao]

Three issues remain in the dead-event-loop recovery path: existing Flink streams and pending RPCs can hang, superseded worker groups retain live threads, and the triggering request fails when no retry remains. I reproduced the first two behaviors against Netty 4.2.10.

[SteNicholas]

@sunchao, thanks for review. I have addressed above comments. PTAL.

[sunchao]

Follow-up review of 037c1309e37b: four P1 liveness failures and three P2 connection-recovery or worker-group-lifecycle issues remain. I reproduced each issue against the exact-head Java code and Netty 4.2.10; the existing direct-sweep regression test still passes while the real same-peer replacement path leaves its original RPC orphaned.

[sunchao]

[P1] Preserve the displaced client until its outstanding work is failed.

With the default one connection per peer, reconnecting to the same peer replaces the dead cachedClient here before the finally block runs failClientsOnDeadEventLoopsIfRecreated(). That sweep walks only the current pool slots, so the displaced client is already unreachable: its outstanding RPC never completes, and its dead channel remains tracked. I reproduced the actual recovery path on this head: the replacement was active while the old callback was never failed and its RPC remained outstanding. The new regression test calls failClientsOnDeadEventLoops() directly while the old client is still pooled, so it misses this case. Please retain the evicted client and invalidate/untrack it after releasing the pool lock, including when reconnection succeeds on another live loop without recreating the group.

[sunchao]

[P1] Notify established inbound streams when invalidating a dead client.

An established Flink credit stream normally has no outstanding RPC: OPEN_STREAM and subsequent credit/segment updates have already been acknowledged. Its ownership instead lives in ReadClientHandler.streamClients, and ReadClientHandler.channelInactive() is what sends TransportableError to the corresponding stream readers. This method only drains the response handler and never invokes the inbound request handler, while a genuinely dead loop cannot deliver channelInactive() itself. Consequently, the recovery sweep notifies nobody and an already-open Flink reader remains stalled indefinitely. I reproduced that dead-loop invalidation leaves an installed inbound handler unnotified, whereas the normal TransportChannelHandler.channelInactive() path notifies it. Please propagate dead-client invalidation to the request/inbound stream handler exactly once.

[sunchao]

[P1] Close the race between the dead-loop check and RPC registration.

sendRpc() checks the loop before adding its callback to outstandingRpcs. A sender can observe a live loop, pause, and let another thread observe the loop die, recreate the group, and finish the entire invalidation sweep while this callback is not yet in the map. The sender then registers its callback and writes to the dead loop; neither the write listener nor channelInactive() runs, RPCs have no timeout checker, and no further sweep occurs unless another recreation happens. A controlled interleaving on this head leaves the RPC outstanding after the sweep with its callback never invoked. Please register before rechecking/invalidation, or synchronize registration with terminal invalidation so neither operation can miss the other.

[sunchao]

[P1] Keep timeout checkers alive until new push/fetch requests are rejected.

Dead-loop invalidation cancels the per-client push/fetch timeout checkers, but TransportClient.fetchChunk(), pushData(), and pushMergedData() still register and write new requests without a dead-loop guard. For example, WorkerPartitionReader can observe its client active, another thread can then invalidate the now-dead client and cancel this checker, and the reader can resume into fetchChunk(). The write listener never runs and the timeout backstop has been removed, so next() polls forever. I reproduced an expired fetch remaining outstanding after cancellation while an otherwise identical uncanceled checker fails the request normally. Please keep the checkers running until actual channel closure, or atomically mark the handler terminal and immediately fail every later push/fetch registration.

[sunchao]

[P2] Create a fresh decoder for the replacement connection.

The comment assumes "event executor terminated" means the original ChannelInitializer never ran, but Netty 4.2.10 can initialize/register the channel and install this decoder before Bootstrap.doConnect() submits the actual connect task to the event loop. If the loop dies between those steps, the same terminated-executor rejection triggers this retry with a decoder that is already present in the old pipeline. Both TransportFrameDecoder and the Flink decoder are non-@Sharable; I reproduced Netty rejecting the second installation with ChannelPipelineException. Recovery therefore fails despite creating a healthy replacement, including for unmanaged clients and maxRetries=1. Please pass a decoder supplier through the retry and allocate a fresh handler for the second channel.

[sunchao]

[P2] Include in-flight registrations when deciding whether the old group drained.

trackChannel() is called only after a connect completes, but this immediate retirement treats an empty tracked-channel set as proof that the old group has no users. With multiple loops, a concurrent connection can already be registered on a still-healthy loop in that group while its connect is pending and its channel has not yet reached trackChannel(). Recovery on another dead loop then shuts down the entire old group and closes the otherwise valid in-flight connection. I reproduced this with a real registered NioSocketChannel: recreation observed no tracked channels, shut down the old group, and closed the registered channel. Please track pending registrations under the same retirement coordination, or register channels before testing whether the group has drained.

[sunchao]

[P2] Untrack dead unmanaged channels before retiring superseded groups.

createUnmanagedClient() still registers its channel in workerGroupChannels, but this sweep visits only connectionPool. If an unmanaged channel is pinned to the dead loop, its closeFuture() never completes, so it is never untracked; even calling TransportClient.invalidateIfEventLoopDead() only fails response callbacks and does not inform the factory. The superseded group therefore appears permanently nonempty, keeping its remaining selector threads alive until the whole factory closes. I reproduced a tracked dead unmanaged channel surviving both the pool sweep and owner-side invalidation while supersededWorkerGroupCount() remains one. Please sweep the tracked channels themselves and explicitly remove dead-loop channels regardless of whether their clients were pooled.