This is an automated email from the ASF dual-hosted git repository. SteNicholas pushed a commit to branch CELEBORN-2400 in repository https://gitbox.apache.org/repos/asf/celeborn.git
commit a24e727d3520feb4764b1b492e8306bea55e76fb Author: Nicholas Jiang <[email protected]> AuthorDate: Mon Aug 3 16:11:31 2026 +0800 [CELEBORN-2400] Recover from dead netty client event loops 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: - TransportClient.isActive() now returns false when the channel's event loop is shutting down, so a poisoned client is evicted from the pool instead of being reused. - TransportClientFactory replaces its worker group when a connection fails with the terminated-executor rejection, so retries bind to fresh live threads. The superseded group is retained (weakly referenced) until close(), since its live threads may still serve already-open channels, and a closed factory never recreates a group. - Adds celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop (default true) to control the recreation. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> --- .../common/network/client/TransportClient.java | 10 +- .../network/client/TransportClientFactory.java | 168 ++++++++++++++++++--- .../common/network/util/TransportConf.java | 10 ++ .../org/apache/celeborn/common/CelebornConf.scala | 25 +++ .../network/TransportClientFactorySuiteJ.java | 61 +++++++- .../network/client/TransportClientSuiteJ.java | 56 +++++++ docs/configuration/network.md | 1 + 7 files changed, 305 insertions(+), 26 deletions(-) diff --git a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java index b0a61d054f..2ea19914a9 100644 --- a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java +++ b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java @@ -92,7 +92,15 @@ 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 request 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/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java index b79383277a..dac015b48a 100644 --- a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java +++ b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java @@ -19,11 +19,14 @@ package org.apache.celeborn.common.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.List; import java.util.Random; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -96,7 +99,27 @@ public class TransportClientFactory implements Closeable { private final int sendBuf; 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 a connection fails because the loop is dead, we replace this group with a + // fresh one so subsequent connections bind to live threads; volatile so the swap is visible to + // concurrent 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. + private boolean closed = false; protected ByteBufAllocator allocator; private final int maxClientConnectRetries; private final int maxClientConnectRetryWaitTimeMs; @@ -115,7 +138,7 @@ public class TransportClientFactory implements Closeable { this.sendBuf = conf.sendBuf(); this.rand = new Random(); - IOMode ioMode = IOMode.valueOf(conf.ioMode()); + this.ioMode = IOMode.valueOf(conf.ioMode()); this.socketChannelClass = NettyUtils.getClientChannelClass(ioMode); logger.info("Module {} mode {} threads {}", conf.getModuleName(), ioMode, conf.clientThreads()); this.workerGroup = @@ -124,6 +147,7 @@ public class TransportClientFactory implements Closeable { conf.clientThreads(), conf.conflictAvoidChooserEnable(), conf.getModuleName() + "-client"); + this.recreateWorkerGroupOnDeadEventLoop = conf.recreateWorkerGroupOnDeadEventLoop(); // Always disable thread-local cache when creating pooled ByteBuf allocator for TransportClients // because the ByteBufs are allocated by the event loop thread, but released by the executor // thread rather than the event loop thread. Those thread-local caches actually delay the @@ -134,6 +158,11 @@ public class TransportClientFactory implements Closeable { this.maxClientConnectRetryWaitTimeMs = conf.ioRetryWaitTimeMs(); } + @VisibleForTesting + public EventLoopGroup getWorkerGroup() { + return workerGroup; + } + /** * Create a {@link TransportClient} connecting to the given remote host / port. * @@ -286,8 +315,11 @@ public class TransportClientFactory implements Closeable { InetSocketAddress address, ChannelInboundHandlerAdapter decoder) throws IOException, InterruptedException { Bootstrap bootstrap = new Bootstrap(); + // 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(workerGroup) + .group(connectGroup) .channel(socketChannelClass) // Disable Nagle's Algorithm since we don't want packets to wait .option(ChannelOption.TCP_NODELAY, true) @@ -317,28 +349,39 @@ public class TransportClientFactory implements Closeable { // Connect to the remote server long preConnect = System.nanoTime(); ChannelFuture cf = bootstrap.connect(address); - if (connectTimeoutMs <= 0) { - awaitWithChannelCleanup( - () -> { - cf.await(); - return true; - }, - cf); - assert cf.isDone(); - if (cf.isCancelled()) { + try { + if (connectTimeoutMs <= 0) { + awaitWithChannelCleanup( + () -> { + cf.await(); + return true; + }, + cf); + assert cf.isDone(); + if (cf.isCancelled()) { + closeChannel(cf); + throw new IOException(String.format("Connecting to %s cancelled", address)); + } else if (!cf.isSuccess()) { + closeChannel(cf); + throw new IOException(String.format("Failed to connect to %s", address), cf.cause()); + } + } else if (!awaitWithChannelCleanup(() -> cf.await(connectTimeoutMs), cf)) { closeChannel(cf); - throw new IOException(String.format("Connecting to %s cancelled", address)); - } else if (!cf.isSuccess()) { + throw new CelebornIOException( + String.format("Connecting to %s timed out (%s ms)", address, connectTimeoutMs)); + } else if (cf.cause() != null) { closeChannel(cf); - throw new IOException(String.format("Failed to connect to %s", address), cf.cause()); + throw new CelebornIOException( + String.format("Failed to connect to %s", address), cf.cause()); } - } else if (!awaitWithChannelCleanup(() -> cf.await(connectTimeoutMs), cf)) { - closeChannel(cf); - throw new CelebornIOException( - String.format("Connecting to %s timed out (%s ms)", address, connectTimeoutMs)); - } else if (cf.cause() != null) { - closeChannel(cf); - throw new CelebornIOException(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. retryCreateClient) retries. + recreateWorkerGroupIfEventLoopDead(connectGroup, cf.cause()); + throw e; } if (context.sslEncryptionEnabled()) { final SslHandler sslHandler = cf.channel().pipeline().get(SslHandler.class); @@ -432,6 +475,71 @@ public class TransportClientFactory implements Closeable { } } + /** + * 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++; + TransportConf conf = context.getConf(); + workerGroup = + NettyUtils.createEventLoop( + ioMode, + conf.clientThreads(), + conf.conflictAvoidChooserEnable(), + 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() { @@ -447,10 +555,26 @@ 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). + synchronized (this) { + closed = true; + } + // SPARK-19147 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(); } public TransportContext getContext() { diff --git a/common/src/main/java/org/apache/celeborn/common/network/util/TransportConf.java b/common/src/main/java/org/apache/celeborn/common/network/util/TransportConf.java index ce37ae9788..8cb809fa35 100644 --- a/common/src/main/java/org/apache/celeborn/common/network/util/TransportConf.java +++ b/common/src/main/java/org/apache/celeborn/common/network/util/TransportConf.java @@ -92,6 +92,16 @@ public class TransportConf { return celebornConf.networkIoConflictAvoidChooserEnable(module); } + /** + * 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 celebornConf.networkIoRecreateWorkerGroupOnDeadEventLoop(module); + } + /** * Receive buffer size (SO_RCVBUF). Note: the optimal size for receive buffer and send buffer * should be latency * network_bandwidth. Assuming latency = 1ms, network_bandwidth = 10Gbps diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala index ea6b819fcf..9158144c33 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -585,6 +585,10 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se getBoolean(key, NETWORK_IO_CLIENT_CONFLICT_AVOID_CHOOSER_ENABLE.defaultValue.get) } + def networkIoRecreateWorkerGroupOnDeadEventLoop(module: String): Boolean = { + getTransportConfBoolean(module, NETWORK_IO_RECREATE_WORKER_GROUP_ON_DEAD_EVENT_LOOP) + } + def networkIoReceiveBuf(module: String): Int = { getTransportConfSizeAsBytes(module, NETWORK_IO_RECEIVE_BUFFER).toInt } @@ -2296,6 +2300,27 @@ object CelebornConf extends Logging { .booleanConf .createWithDefault(false) + val NETWORK_IO_RECREATE_WORKER_GROUP_ON_DEAD_EVENT_LOOP: ConfigEntry[Boolean] = + buildConf("celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop") + .categories("network") + .version("1.0.0") + .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 request on it can hang forever. When enabled, such a failure recreates the " + + "worker group so subsequent connections bind to live threads and the retry can succeed. " + + s"If setting <module> to `${TransportModuleConstants.RPC_APP_MODULE}`, " + + s"works for shuffle client. " + + s"If setting <module> to `${TransportModuleConstants.RPC_SERVICE_MODULE}`, " + + s"works for master or worker. " + + s"If setting <module> to `${TransportModuleConstants.DATA_MODULE}`, " + + s"it works for shuffle client push and fetch data. " + + s"If setting <module> to `${TransportModuleConstants.REPLICATE_MODULE}`, " + + s"it works for replicate client of worker replicating data to peer worker.") + .booleanConf + .createWithDefault(true) + val NETWORK_IO_RECEIVE_BUFFER: ConfigEntry[Long] = buildConf("celeborn.<module>.io.receiveBuffer") .categories("network") diff --git a/common/src/test/java/org/apache/celeborn/common/network/TransportClientFactorySuiteJ.java b/common/src/test/java/org/apache/celeborn/common/network/TransportClientFactorySuiteJ.java index 3c51a175d3..a81b3a9b4f 100644 --- a/common/src/test/java/org/apache/celeborn/common/network/TransportClientFactorySuiteJ.java +++ b/common/src/test/java/org/apache/celeborn/common/network/TransportClientFactorySuiteJ.java @@ -28,6 +28,7 @@ import java.io.IOException; import java.util.*; import java.util.concurrent.atomic.AtomicInteger; +import io.netty.channel.EventLoopGroup; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -214,11 +215,16 @@ public class TransportClientFactorySuiteJ { context.close(); } - @Test(expected = IOException.class) - public void closeFactoryBeforeCreateClient() throws IOException, InterruptedException { + @Test + public void closeFactoryBeforeCreateClient() { TransportClientFactory factory = context.createClientFactory(); + EventLoopGroup groupBeforeClose = factory.getWorkerGroup(); factory.close(); - factory.createClient(getLocalHost(), server1.getPort()); + assertThrows(IOException.class, () -> factory.createClient(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 @@ -248,6 +254,55 @@ public class TransportClientFactorySuiteJ { } } + @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 connection's channel registration is rejected with "event executor terminated", so + // the factory must swap in a fresh worker group and the built-in retry must succeed. + CelebornConf _conf = new CelebornConf(); + _conf.set("celeborn.shuffle.io.retryWait", "100ms"); + TransportConf conf = new TransportConf(TEST_MODULE, _conf); + TransportContext ctx = new TransportContext(conf, new BaseMessageHandler()); + try (TransportClientFactory factory = ctx.createClientFactory()) { + EventLoopGroup deadGroup = factory.getWorkerGroup(); + deadGroup.shutdownGracefully().sync(); + assertTrue(deadGroup.isShuttingDown()); + + // The first connect attempt fails because the (dead) group rejects the channel + // registration; the factory replaces the worker group and the retry then succeeds. + TransportClient client = factory.createClient(getLocalHost(), server1.getPort()); + assertTrue(client.isActive()); + + EventLoopGroup freshGroup = factory.getWorkerGroup(); + assertNotSame(deadGroup, freshGroup); + assertFalse(freshGroup.isShuttingDown()); + } finally { + ctx.close(); + } + } + + @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. + CelebornConf _conf = new CelebornConf(); + _conf.set("celeborn.shuffle.io.recreateWorkerGroupOnDeadEventLoop", "false"); + _conf.set("celeborn.shuffle.io.retryWait", "100ms"); + TransportConf conf = new TransportConf(TEST_MODULE, _conf); + TransportContext ctx = new TransportContext(conf, new BaseMessageHandler()); + try (TransportClientFactory factory = ctx.createClientFactory()) { + EventLoopGroup deadGroup = factory.getWorkerGroup(); + deadGroup.shutdownGracefully().sync(); + + assertThrows( + IOException.class, () -> factory.createClient(getLocalHost(), server1.getPort())); + assertSame(deadGroup, factory.getWorkerGroup()); + } finally { + ctx.close(); + } + } + @Test public void testRetryCreateClient() throws IOException, InterruptedException { TransportClientFactory factory = Mockito.spy(context.createClientFactory()); diff --git a/common/src/test/java/org/apache/celeborn/common/network/client/TransportClientSuiteJ.java b/common/src/test/java/org/apache/celeborn/common/network/client/TransportClientSuiteJ.java new file mode 100644 index 0000000000..c04b4c8eb1 --- /dev/null +++ b/common/src/test/java/org/apache/celeborn/common/network/client/TransportClientSuiteJ.java @@ -0,0 +1,56 @@ +/* + * 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.celeborn.common.network.client; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import io.netty.channel.Channel; +import io.netty.channel.EventLoop; +import org.junit.Test; + +import org.apache.celeborn.common.CelebornConf; +import org.apache.celeborn.common.network.util.TransportConf; + +public class TransportClientSuiteJ { + + @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). + TransportConf conf = new TransportConf("shuffle", new CelebornConf()); + 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(conf, 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/docs/configuration/network.md b/docs/configuration/network.md index 636bca1bcb..9ee360e11a 100644 --- a/docs/configuration/network.md +++ b/docs/configuration/network.md @@ -33,6 +33,7 @@ license: | | celeborn.<module>.io.numConnectionsPerPeer | 1 | false | Number of concurrent connections between two nodes. If setting <module> to `rpc_app`, works for shuffle client. If setting <module> to `rpc_service`, works for master or worker. If setting <module> to `data`, it works for shuffle client push and fetch data. If setting <module> to `replicate`, it works for replicate client of worker replicating data to peer worker. | | | | celeborn.<module>.io.preferDirectBufs | true | false | If true, we will prefer allocating off-heap byte buffers within Netty. If setting <module> to `rpc_app`, works for shuffle client. If setting <module> to `rpc_service`, works for master or worker. If setting <module> to `data`, it works for shuffle client push and fetch data. If setting <module> to `push`, it works for worker receiving push data. If setting <module> to `replicate`, it works for replicate server or client of w [...] | celeborn.<module>.io.receiveBuffer | 0b | false | Receive buffer size (SO_RCVBUF). Note: the optimal size for receive buffer and send buffer should be latency * network_bandwidth. Assuming latency = 1ms, network_bandwidth = 10Gbps buffer size should be ~ 1.25MB. If setting <module> to `rpc_app`, works for shuffle client. If setting <module> to `rpc_service`, works for master or worker. If setting <module> to `data`, it works for shuffle client push and fetch data. If setting <mod [...] +| celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop | true | false | 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 request on it can hang forever. When enabled, such a failure recrea [...] | celeborn.<module>.io.retryWait | 5s | false | Time that we will wait in order to perform a retry after an IOException. Only relevant if maxIORetries > 0. If setting <module> to `data`, it works for shuffle client push and fetch data. If setting <module> to `replicate`, it works for replicate client of worker replicating data to peer worker. If setting <module> to `push`, it works for Flink shuffle client push data. | 0.2.0 | | | celeborn.<module>.io.saslTimeout | 30s | false | Timeout for a single round trip of auth message exchange, in milliseconds. | 0.5.0 | | | celeborn.<module>.io.sendBuffer | 0b | false | Send buffer size (SO_SNDBUF). If setting <module> to `rpc_app`, works for shuffle client. If setting <module> to `rpc_service`, works for master or worker. If setting <module> to `data`, it works for shuffle client push and fetch data. If setting <module> to `push`, it works for worker receiving push data. If setting <module> to `replicate`, it works for replicate server or client of worker replicating data to peer worker. If setting [...]
