You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/07/29 12:57:07 UTC
(spark) branch master updated: [SPARK-58292][CORE] Recreate the netty worker EventLoopGroup when a worker event loop thread dies
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 2c8570a1d108 [SPARK-58292][CORE] Recreate the netty worker EventLoopGroup when a worker event loop thread dies
2c8570a1d108 is described below
commit 2c8570a1d1080a03d828f8c1b7de025166d62d64
Author: ChuckLin2025 <lz...@gmail.com>
AuthorDate: Wed Jul 29 20:56:51 2026 +0800
[SPARK-58292][CORE] Recreate the netty worker EventLoopGroup when a worker 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 fo [...]
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 <lz...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/network/client/TransportClient.java | 9 +-
.../network/client/TransportClientFactory.java | 148 ++++++++++++++++++---
.../apache/spark/network/util/TransportConf.java | 10 ++
.../client/TransportClientFactorySuite.java | 50 +++++++
.../spark/network/client/TransportClientSuite.java | 51 +++++++
.../org/apache/spark/internal/config/Network.scala | 13 ++
6 files changed, 265 insertions(+), 16 deletions(-)
diff --git a/common/network-common/src/main/java/org/apache/spark/network/client/TransportClient.java b/common/network-common/src/main/java/org/apache/spark/network/client/TransportClient.java
index f194a4cb94eb..4a19bccf06a3 100644
--- a/common/network-common/src/main/java/org/apache/spark/network/client/TransportClient.java
+++ b/common/network-common/src/main/java/org/apache/spark/network/client/TransportClient.java
@@ -92,7 +92,14 @@ public class TransportClient implements Closeable {
}
public boolean isActive() {
- return !timedOut && (channel.isOpen() || channel.isActive());
+ // A channel is pinned to one netty event-loop thread for its lifetime. If that event loop has
+ // terminated (e.g. an uncaught error killed the thread; netty does not replace it in a
+ // fixed-size EventLoopGroup), the channel can no longer send or complete anything: writes and
+ // listener notifications route to the dead loop and are silently dropped. Such a client must
+ // not be treated as active/reused, otherwise a fetch on it can orphan and hang. See
+ // SPARK-58292.
+ return !timedOut && !channel.eventLoop().isShuttingDown()
+ && (channel.isOpen() || channel.isActive());
}
public SocketAddress getSocketAddress() {
diff --git a/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java b/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java
index 2137b5f3136e..1f886e2a7752 100644
--- a/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java
+++ b/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java
@@ -19,6 +19,7 @@ package org.apache.spark.network.client;
import java.io.Closeable;
import java.io.IOException;
+import java.lang.ref.WeakReference;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
@@ -26,6 +27,8 @@ import java.util.List;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicReference;
import com.codahale.metrics.MetricSet;
@@ -91,7 +94,27 @@ public class TransportClientFactory implements Closeable {
private final int numConnectionsPerPeer;
private final Class<? extends Channel> socketChannelClass;
- private EventLoopGroup workerGroup;
+ private final IOMode ioMode;
+ // The client worker EventLoopGroup. A netty event-loop thread that dies (e.g. an uncaught error)
+ // is never replaced within a fixed-size group, permanently poisoning any channel pinned to it
+ // (SPARK-58292). When createClient sees a connection fail because the loop is dead, it replaces
+ // this group with a fresh one so subsequent connections bind to live threads; volatile so the
+ // swap is visible to concurrent createClient callers.
+ private volatile EventLoopGroup workerGroup;
+ // Superseded worker groups are not shut down eagerly: their still-live threads may be serving
+ // channels that are already open. We keep weak references and shut them down best-effort at
+ // close(); their threads are daemon, so a not-yet-collected group cannot block JVM shutdown.
+ private final List<WeakReference<EventLoopGroup>> deprecatedWorkerGroups =
+ new CopyOnWriteArrayList<>();
+ // Whether to recreate the worker group on a dead event loop (SPARK-58292); on by default.
+ private final boolean recreateWorkerGroupOnDeadEventLoop;
+ // How many times the worker group has been recreated after a dead event loop. Used only to make
+ // each recreated group's thread names distinct; mutated under the recreateWorkerGroup lock.
+ private int workerGroupRecreationCount;
+ // Set once close() has shut the worker group down, so the dead-event-loop path does not
+ // resurrect a closed factory by recreating a fresh (leaked) group. Guarded by the same lock as
+ // recreateWorkerGroup. See SPARK-58292.
+ private boolean closed = false;
private final PooledByteBufAllocator pooledAllocator;
private final NettyMemoryMetrics metrics;
private final int fastFailTimeWindow;
@@ -106,7 +129,7 @@ public class TransportClientFactory implements Closeable {
this.numConnectionsPerPeer = conf.numConnectionsPerPeer();
this.rand = new Random();
- IOMode ioMode = IOMode.valueOf(conf.ioMode());
+ this.ioMode = IOMode.valueOf(conf.ioMode());
this.socketChannelClass = NettyUtils.getClientChannelClass(ioMode);
this.workerGroup = NettyUtils.createEventLoop(
ioMode,
@@ -121,6 +144,7 @@ public class TransportClientFactory implements Closeable {
}
this.metrics = new NettyMemoryMetrics(
this.pooledAllocator, conf.getModuleName() + "-client", conf);
+ this.recreateWorkerGroupOnDeadEventLoop = conf.recreateWorkerGroupOnDeadEventLoop();
fastFailTimeWindow = (int)(conf.ioRetryWaitTimeMs() * 0.95);
}
@@ -128,6 +152,11 @@ public class TransportClientFactory implements Closeable {
return metrics;
}
+ @VisibleForTesting
+ EventLoopGroup getWorkerGroup() {
+ return workerGroup;
+ }
+
/**
* Create a {@link TransportClient} connecting to the given remote host / port.
*
@@ -256,7 +285,10 @@ public class TransportClientFactory implements Closeable {
Bootstrap bootstrap = new Bootstrap();
int connCreateTimeout = conf.connectionCreationTimeoutMs();
- bootstrap.group(workerGroup)
+ // Capture the group this connection uses, so that on a dead-event-loop failure we replace
+ // exactly this group (and not one a concurrent caller already swapped in).
+ final EventLoopGroup connectGroup = workerGroup;
+ bootstrap.group(connectGroup)
.channel(socketChannelClass)
// Disable Nagle's Algorithm since we don't want packets to wait
.option(ChannelOption.TCP_NODELAY, true)
@@ -288,20 +320,30 @@ public class TransportClientFactory implements Closeable {
long preConnect = System.nanoTime();
ChannelFuture cf = bootstrap.connect(address);
- if (connCreateTimeout <= 0) {
- cf.await();
- assert cf.isDone();
- if (cf.isCancelled()) {
- throw new IOException(String.format("Connecting to %s cancelled", address));
- } else if (!cf.isSuccess()) {
+ try {
+ if (connCreateTimeout <= 0) {
+ cf.await();
+ assert cf.isDone();
+ if (cf.isCancelled()) {
+ throw new IOException(String.format("Connecting to %s cancelled", address));
+ } else if (!cf.isSuccess()) {
+ throw new IOException(String.format("Failed to connect to %s", address), cf.cause());
+ }
+ } else if (!cf.await(connCreateTimeout)) {
+ throw new IOException(
+ String.format("Connecting to %s timed out (%s ms)",
+ address, connCreateTimeout));
+ } else if (cf.cause() != null) {
throw new IOException(String.format("Failed to connect to %s", address), cf.cause());
}
- } else if (!cf.await(connCreateTimeout)) {
- throw new IOException(
- String.format("Connecting to %s timed out (%s ms)",
- address, connCreateTimeout));
- } else if (cf.cause() != null) {
- throw new IOException(String.format("Failed to connect to %s", address), cf.cause());
+ } catch (IOException e) {
+ // If the connection failed because the channel could not be registered on its netty event
+ // loop (the loop's thread has died and netty rejects new tasks with "event executor
+ // 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());
+ throw e;
}
if (context.sslEncryptionEnabled()) {
final SslHandler sslHandler = cf.channel().pipeline().get(SslHandler.class);
@@ -354,6 +396,66 @@ public class TransportClientFactory implements Closeable {
return client;
}
+ /**
+ * If the given connection-failure cause was a rejection by a dead netty event loop (its worker
+ * thread terminated and netty rejects new registrations with a
+ * {@link RejectedExecutionException}), replace the worker group so subsequent connections bind to
+ * fresh, live threads. A dead loop is never replaced within a fixed-size group and keeps being
+ * selected by the round-robin chooser, so without this the degradation is permanent. See
+ * SPARK-58292.
+ */
+ private void recreateWorkerGroupIfEventLoopDead(EventLoopGroup connectGroup, Throwable cause) {
+ if (!recreateWorkerGroupOnDeadEventLoop) {
+ return;
+ }
+ boolean eventLoopDead = false;
+ for (Throwable t = cause; t != null; t = t.getCause()) {
+ // Match ONLY the terminated-loop rejection, not a transient task-queue-full rejection.
+ // netty's SingleThreadEventExecutor.reject() throws exactly this message when isShutdown();
+ // the queue-full handler path throws a RejectedExecutionException with no message.
+ if (t instanceof RejectedExecutionException
+ && "event executor terminated".equals(t.getMessage())) {
+ eventLoopDead = true;
+ break;
+ }
+ }
+ if (eventLoopDead) {
+ recreateWorkerGroup(connectGroup);
+ }
+ }
+
+ /**
+ * Replace the worker group with a fresh one, if it is still the group the failed connection used
+ * ({@code connectGroup}). The superseded group is not shut down here: its still-live threads may
+ * be serving channels that are already open. We keep a weak reference and shut it down
+ * best-effort at {@link #close()}; its threads are daemon, so a not-yet-collected group cannot
+ * block JVM shutdown. Synchronized and identity-guarded so concurrent callers that all hit the
+ * same dead group replace it exactly once rather than spawning many groups.
+ */
+ private synchronized void recreateWorkerGroup(EventLoopGroup connectGroup) {
+ // The factory is closed (or closing): its worker group was shut down by close(), so a
+ // createClient() racing or following close() must not recreate a fresh group and resurrect a
+ // closed factory (which would leak threads that close() will never reap again).
+ if (closed) {
+ return;
+ }
+ // A concurrent caller that hit the same dead group already swapped it out; nothing to do.
+ if (workerGroup != connectGroup) {
+ return;
+ }
+ // Use a distinct thread-name prefix so the recreated group is self-identifying in thread
+ // dumps and logs. Netty's DefaultThreadFactory also appends an incrementing pool id, so the
+ // names would not collide even with the same prefix, but tagging it "-recreated-<n>" makes the
+ // dead-event-loop recovery obvious to anyone inspecting the process, and the <n> distinguishes
+ // successive recreations if a loop dies more than once.
+ workerGroupRecreationCount++;
+ workerGroup = NettyUtils.createEventLoop(ioMode, conf.clientThreads(),
+ conf.getModuleName() + "-client-recreated-" + workerGroupRecreationCount);
+ deprecatedWorkerGroups.add(new WeakReference<>(connectGroup));
+ logger.warn("Detected a dead netty client event loop; replaced the worker group. The " +
+ "previous group is retained until its open channels drain (SPARK-58292).");
+ }
+
/** Close all connections in the connection pool, and shutdown the worker thread pool. */
@Override
public void close() {
@@ -369,8 +471,24 @@ public class TransportClientFactory implements Closeable {
}
connectionPool.clear();
+ // Mark closed under the recreateWorkerGroup lock before shutting the group down, so a
+ // concurrent createClient() hitting the dead-event-loop path cannot recreate a fresh group
+ // after we have decided to close (which would leak threads). See SPARK-58292.
+ synchronized (this) {
+ closed = true;
+ }
+
if (workerGroup != null && !workerGroup.isShuttingDown()) {
workerGroup.shutdownGracefully();
}
+
+ // Shut down any worker groups superseded after a dead-event-loop recreation, if not yet GC'd.
+ for (WeakReference<EventLoopGroup> ref : deprecatedWorkerGroups) {
+ EventLoopGroup group = ref.get();
+ if (group != null && !group.isShuttingDown()) {
+ group.shutdownGracefully();
+ }
+ }
+ deprecatedWorkerGroups.clear();
}
}
diff --git a/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java b/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java
index 22dd3e2b7bdf..a37979bf30ac 100644
--- a/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java
+++ b/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java
@@ -534,6 +534,16 @@ public class TransportConf {
return conf.getBoolean("spark.shuffle.useOldFetchProtocol", false);
}
+ /**
+ * Whether to replace the client worker EventLoopGroup when a netty worker event loop is detected
+ * as dead (a connection is rejected with "event executor terminated"). A dead loop is never
+ * replaced within a fixed-size group and keeps being handed out by the round-robin chooser, so
+ * without this the degradation is permanent. On by default. See SPARK-58292.
+ */
+ public boolean recreateWorkerGroupOnDeadEventLoop() {
+ return conf.getBoolean("spark.network.recreateWorkerGroupOnDeadEventLoop", true);
+ }
+
/** Whether to enable sasl retries or not. The number of retries is dictated by the config
* `spark.shuffle.io.maxRetries`.
*/
diff --git a/common/network-common/src/test/java/org/apache/spark/network/client/TransportClientFactorySuite.java b/common/network-common/src/test/java/org/apache/spark/network/client/TransportClientFactorySuite.java
index 58faea2cf2dd..8081579bfaee 100644
--- a/common/network-common/src/test/java/org/apache/spark/network/client/TransportClientFactorySuite.java
+++ b/common/network-common/src/test/java/org/apache/spark/network/client/TransportClientFactorySuite.java
@@ -219,9 +219,14 @@ public class TransportClientFactorySuite {
@Test
public void closeFactoryBeforeCreateClient() {
TransportClientFactory factory = context.createClientFactory();
+ io.netty.channel.EventLoopGroup groupBeforeClose = factory.getWorkerGroup();
factory.close();
Assertions.assertThrows(IOException.class,
() -> factory.createClient(TestUtils.getLocalHost(), server1.getPort()));
+ // SPARK-58292: createClient on a closed factory fails with the terminated-executor cause, but
+ // the closed factory must NOT recreate a fresh worker group (which would leak threads). The
+ // group is left as-is, i.e. the shut-down one from before close().
+ assertSame(groupBeforeClose, factory.getWorkerGroup());
}
@Test
@@ -265,4 +270,49 @@ public class TransportClientFactorySuite {
assertNotEquals(exception.getCause(), null);
}
}
+
+ @Test
+ public void recreatesWorkerGroupWhenEventLoopIsDead() throws Exception {
+ // SPARK-58292: a dead netty worker event loop is never replaced within a fixed-size group and
+ // permanently poisons connections. Simulate it by shutting down the factory's worker group:
+ // the next createClient's channel registration is rejected with "event executor terminated", so
+ // createClient must swap in a fresh worker group (and rethrow so the caller can retry).
+ try (TransportClientFactory factory = context.createClientFactory()) {
+ io.netty.channel.EventLoopGroup deadGroup = factory.getWorkerGroup();
+ deadGroup.shutdownGracefully().sync();
+ assertTrue(deadGroup.isShuttingDown());
+
+ // The connect fails because the (dead) group rejects the channel registration.
+ assertThrows(IOException.class,
+ () -> factory.createClient(TestUtils.getLocalHost(), server1.getPort()));
+
+ // The worker group was replaced with a fresh, live one.
+ io.netty.channel.EventLoopGroup freshGroup = factory.getWorkerGroup();
+ assertNotSame(deadGroup, freshGroup);
+ assertFalse(freshGroup.isShuttingDown());
+
+ // And a subsequent connection now succeeds on the fresh group.
+ TransportClient client = factory.createClient(TestUtils.getLocalHost(), server1.getPort());
+ assertTrue(client.isActive());
+ }
+ }
+
+ @Test
+ public void doesNotRecreateWorkerGroupWhenDisabled() throws Exception {
+ // With the recreation disabled, a dead worker group is NOT recreated: createClient still
+ // fails, but the worker group is left unchanged.
+ Map<String, String> configMap = new HashMap<>();
+ configMap.put("spark.network.recreateWorkerGroupOnDeadEventLoop", "false");
+ TransportConf disabledConf = new TransportConf("shuffle", new MapConfigProvider(configMap));
+ try (TransportContext ctx = new TransportContext(disabledConf, new NoOpRpcHandler());
+ TransportClientFactory factory = ctx.createClientFactory()) {
+ io.netty.channel.EventLoopGroup deadGroup = factory.getWorkerGroup();
+ deadGroup.shutdownGracefully().sync();
+
+ assertThrows(IOException.class,
+ () -> factory.createClient(TestUtils.getLocalHost(), server1.getPort()));
+
+ assertSame(deadGroup, factory.getWorkerGroup());
+ }
+ }
}
diff --git a/common/network-common/src/test/java/org/apache/spark/network/client/TransportClientSuite.java b/common/network-common/src/test/java/org/apache/spark/network/client/TransportClientSuite.java
new file mode 100644
index 000000000000..3dec923a63b4
--- /dev/null
+++ b/common/network-common/src/test/java/org/apache/spark/network/client/TransportClientSuite.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.client;
+
+import io.netty.channel.Channel;
+import io.netty.channel.EventLoop;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TransportClientSuite {
+
+ @Test
+ public void isActiveFalseWhenEventLoopIsShuttingDown() {
+ // SPARK-58292: a client whose netty event loop has terminated must not be treated as active,
+ // even though the TCP channel still reports open/active (nothing can close it -- closing runs
+ // on the now-dead loop).
+ Channel channel = mock(Channel.class);
+ EventLoop eventLoop = mock(EventLoop.class);
+ when(channel.eventLoop()).thenReturn(eventLoop);
+ when(channel.isOpen()).thenReturn(true);
+ when(channel.isActive()).thenReturn(true);
+ TransportClient client = new TransportClient(channel, new TransportResponseHandler(channel));
+
+ // Live event loop -> active.
+ when(eventLoop.isShuttingDown()).thenReturn(false);
+ assertTrue(client.isActive());
+
+ // Terminated event loop -> NOT active, so it will be evicted from the pool and not reused.
+ when(eventLoop.isShuttingDown()).thenReturn(true);
+ assertFalse(client.isActive());
+ }
+}
diff --git a/core/src/main/scala/org/apache/spark/internal/config/Network.scala b/core/src/main/scala/org/apache/spark/internal/config/Network.scala
index 568394208390..12a56d5c8cab 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/Network.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/Network.scala
@@ -39,6 +39,19 @@ private[spark] object Network {
.booleanConf
.createWithDefault(false)
+ private[spark] val NETWORK_RECREATE_WORKER_GROUP_ON_DEAD_EVENT_LOOP =
+ ConfigBuilder("spark.network.recreateWorkerGroupOnDeadEventLoop")
+ .doc("Whether to replace the netty client worker EventLoopGroup when one of its event-loop " +
+ "threads is detected as dead (a connection is rejected with \"event executor " +
+ "terminated\"). A dead event loop is never replaced within a fixed-size group and keeps " +
+ "being handed out by the round-robin chooser, permanently poisoning any channel pinned " +
+ "to it, so a fetch can hang forever. When enabled, such a failure recreates the worker " +
+ "group so subsequent connections bind to live threads and the caller can retry.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(true)
+
private[spark] val NETWORK_TIMEOUT =
ConfigBuilder("spark.network.timeout")
.version("1.3.0")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org