From fa199b435800707b4a8f854177fd121a2c815402 Mon Sep 17 00:00:00 2001 From: Venkata Krishnan Sowrirajan Date: Sun, 20 Sep 2026 13:55:09 -0700 Subject: [PATCH 1/6] [CELEBORN-XXXX] Log the hard split cause once per split file A hard split does not shrink the partition file, so the file keeps failing the threshold check in checkDiskFullAndSplit until a new epoch propagates to every pusher. Until then each concurrent push to that file logged the whole multi-line "CheckDiskFullAndSplit hardSplit" INFO again. On a cluster where celeborn.worker.shuffle.partitionSplit.max equalled celeborn.client.shuffle.partitionSplit.threshold, which makes the SOFT branch unreachable and every split hard, this produced 20.2M log events in 7 minutes from a handful of wedged partitions. Latch the log on the PartitionDataWriter so the cause is recorded once per split file. The latch lives and dies with the writer, so it needs no cleanup hook. WorkerSource.WRITE_DATA_HARD_SPLIT_COUNT already carries the rate. Also fix the sibling warning in handlePushMergedData, which fired once per hard-split partition per merged push and attributed every hard split to a full disk. checkDiskFullAndSplit returns HARD_SPLIT for three reasons and only one of them is disk pressure, so the message misdirected triage. It carries nothing the INFO does not, and the counter on the next line already covers rate, so it moves to debug. Behavior is unchanged, this is log noise only. Co-Authored-By: Claude Opus 5 (1M context) --- .../deploy/worker/storage/PartitionDataWriter.java | 9 +++++++++ .../celeborn/service/deploy/worker/PushDataHandler.scala | 9 ++++++--- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriter.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriter.java index cf7ee4e5f4c..6c818593dcb 100644 --- a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriter.java +++ b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriter.java @@ -19,6 +19,7 @@ import java.io.IOException; import java.util.Objects; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import scala.Option; @@ -51,6 +52,9 @@ public class PartitionDataWriter implements DeviceObserver { private final PartitionSplitMode splitMode; private final long memoryFileStorageMaxFileSize; private final AtomicInteger numPendingWrites = new AtomicInteger(0); + // Latches on the first hard split of this file so the hard split cause is logged once per + // split file instead of once per push. Dies with the writer, so it needs no cleanup. + private final AtomicBoolean hardSplitLogged = new AtomicBoolean(false); private final PartitionDataWriterContext writerContext; protected final AbstractSource source; // metrics private final String writerString; @@ -263,6 +267,11 @@ public PartitionSplitMode getSplitMode() { return splitMode; } + /** Returns true only for the first hard split of this file, false for every subsequent one. */ + public boolean shouldLogHardSplit() { + return hardSplitLogged.compareAndSet(false, true); + } + @Override public void notifyError(String mountPoint, DiskStatus diskStatus) { destroy( diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala index f6c32e77f34..3ef9a75115a 100644 --- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala +++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala @@ -589,8 +589,9 @@ class PushDataHandler(val workerSource: WorkerSource) extends BaseMessageHandler } else { val splitStatus = checkDiskFullAndSplit(fileWriter, isPrimary) if (splitStatus == StatusCode.HARD_SPLIT) { - logWarning( - s"return hard split for disk full with shuffle $shuffleKey map $mapId attempt $attemptId") + logDebug( + s"return hard split for shuffle $shuffleKey map $mapId attempt $attemptId; " + + s"see the CheckDiskFullAndSplit log for the cause") workerSource.incCounter(WorkerSource.WRITE_DATA_HARD_SPLIT_COUNT) pushMergedDataCallback.addSplitPartition(fileWriterIndex, StatusCode.HARD_SPLIT) } else if (splitStatus == StatusCode.SOFT_SPLIT) { @@ -1470,7 +1471,9 @@ class PushDataHandler(val workerSource: WorkerSource) extends BaseMessageHandler } else { StatusCode.NO_SPLIT } - if (splitStatus == StatusCode.HARD_SPLIT) { + // Log the cause once per split file. A split file keeps failing the threshold check until a + // new epoch propagates, so without the latch every concurrent push to it logs again. + if (splitStatus == StatusCode.HARD_SPLIT && fileWriter.shouldLogHardSplit()) { logInfo( s""" |CheckDiskFullAndSplit hardSplit From 5437a39218f99ed5882257790604c13806491cd3 Mon Sep 17 00:00:00 2001 From: Venkata Krishnan Sowrirajan Date: Sun, 20 Sep 2026 13:55:09 -0700 Subject: [PATCH 2/6] [CELEBORN-XXXX] Add exponential backoff with jitter to push data retries Push data retries are resubmitted to pushDataRetryPool immediately. The dueTime computed at each site is not a retry delay: it is the deadline for the revive busy-wait inside the retry task, so the effective retry cadence is bounded only by celeborn.client.push.revive.interval, 100ms by default. Every map task blocked on the same partition therefore retries in lockstep, and each retry calls dataClientFactory.createClient, so a stuck partition turns into a synchronised burst of connection setup against one worker. This is load-bearing rather than cosmetic on the HARD_SPLIT path, because the two HARD_SPLIT sites pass remainReviveTimes through without decrementing it. A partition wedged in HARD_SPLIT never exhausts its revive budget, so the only brake in the system never engages and the retry loop is unbounded. Schedule retries with an exponentially growing delay, capped and jittered. pushDataRetryPool becomes a ScheduledExecutorService so the delay costs no pool thread; sleeping in the task would be worse than the problem, since the pool defaults to 8 threads that already block on the revive wait. Backoff applies to all five retry sites, not just HARD_SPLIT. The failure sites are reached when a worker is already sick, and retrying an unreachable worker with no delay is what turns one bad worker into a connection storm. Three new configs, all client side: celeborn.client.push.retry.backoff.initial 500ms, 0 disables celeborn.client.push.retry.backoff.max 10s, well under the 120s push data timeout celeborn.client.push.retry.backoff.jitter 0.5, equal jitter so the delay never collapses to zero Not included, deliberately: making the HARD_SPLIT sites decrement remainReviveTimes would bound those retries and start failing tasks. That is a behaviour change and deserves its own ticket. The merged-data HARD_SPLIT site also has no per-batch attempt counter to grow from, so it takes the initial delay with jitter; threading a counter through doPushMergedData is left as follow-up. Co-Authored-By: Claude Opus 5 (1M context) --- .../celeborn/client/ShuffleClientImpl.java | 73 +++++++++++++++---- .../apache/celeborn/common/CelebornConf.scala | 34 +++++++++ docs/configuration/client.md | 3 + 3 files changed, 97 insertions(+), 13 deletions(-) diff --git a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java index 5e84f84ee42..2841888ea3b 100644 --- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java +++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java @@ -22,7 +22,8 @@ import java.util.*; import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; import scala.Tuple2; @@ -131,7 +132,10 @@ public class ShuffleClientImpl extends ShuffleClient { private boolean pushReplicateEnabled; private boolean fetchExcludeWorkerOnFailureEnabled; - private final ExecutorService pushDataRetryPool; + private final ScheduledExecutorService pushDataRetryPool; + private final long pushRetryBackoffInitialMs; + private final long pushRetryBackoffMaxMs; + private final double pushRetryBackoffJitter; private final Map> splitting = JavaUtils.newConcurrentHashMap(); @@ -235,7 +239,11 @@ public ShuffleClientImpl(String appUniqueId, CelebornConf conf, UserIdentifier u initDataClientFactoryIfNeeded(); int pushDataRetryThreads = conf.clientPushRetryThreads(); pushDataRetryPool = - ThreadUtils.newDaemonCachedThreadPool("celeborn-retry-sender", pushDataRetryThreads, 60); + ThreadUtils.newDaemonThreadPoolScheduledExecutor( + "celeborn-retry-sender", pushDataRetryThreads); + pushRetryBackoffInitialMs = conf.clientPushRetryBackoffInitial(); + pushRetryBackoffMaxMs = conf.clientPushRetryBackoffMax(); + pushRetryBackoffJitter = conf.clientPushRetryBackoffJitter(); reviveManager = new ReviveManager(this, conf); @@ -304,6 +312,28 @@ private boolean isPushTargetWorkerExcluded( } } + /** + * Delay before the {@code retryTimes}-th push retry: an exponentially growing base capped at + * {@code celeborn.client.push.retry.backoff.max}, spread by jitter so that map tasks blocked on + * the same partition do not retry in lockstep. Returns 0 when backoff is disabled. + */ + private long pushRetryBackoffMs(int retryTimes) { + if (pushRetryBackoffInitialMs <= 0 || retryTimes <= 0) { + return 0; + } + // Shift-loop rather than a power, because the hard split retry counter is not bounded. + long delay = pushRetryBackoffInitialMs; + for (int i = 1; i < retryTimes && delay < pushRetryBackoffMaxMs; i++) { + delay = delay << 1; + } + delay = Math.min(delay, pushRetryBackoffMaxMs); + long jitter = (long) (delay * pushRetryBackoffJitter); + if (jitter <= 0) { + return delay; + } + return delay - jitter + ThreadLocalRandom.current().nextLong(2 * jitter + 1); + } + private void submitRetryPushData( int shuffleId, byte[] body, @@ -532,7 +562,7 @@ private void submitRetryPushMergedData( } else { ReviveRequest[] requests = addAndGetReviveRequests(shuffleId, mapId, attemptId, reviveFailedBatchesMap, cause); - pushDataRetryPool.submit( + pushDataRetryPool.schedule( () -> submitRetryPushMergedData( pushState, @@ -545,7 +575,9 @@ private void submitRetryPushMergedData( requests, remainReviveTimes - 1, System.currentTimeMillis() - + conf.clientRpcRequestPartitionLocationAskTimeout().duration().toMillis())); + + conf.clientRpcRequestPartitionLocationAskTimeout().duration().toMillis()), + pushRetryBackoffMs(maxReviveTimes - (remainReviveTimes - 1)), + TimeUnit.MILLISECONDS); } } @@ -1133,6 +1165,9 @@ public void onFailure(Throwable e) { RpcResponseCallback wrappedCallback = new PushDataRpcResponseCallback() { int remainReviveTimes = maxReviveTimes; + // HARD_SPLIT retries deliberately do not consume remainReviveTimes, so they need their + // own counter to grow the backoff. + int hardSplitRetryTimes = 0; PartitionLocation latest = loc; @Override @@ -1199,7 +1234,7 @@ public void onSuccess(ByteBuffer response) { + conf.clientRpcRequestPartitionLocationAskTimeout() .duration() .toMillis(); - pushDataRetryPool.submit( + pushDataRetryPool.schedule( () -> submitRetryPushData( shuffleId, @@ -1209,7 +1244,9 @@ public void onSuccess(ByteBuffer response) { pushState, reviveRequest, remainReviveTimes, - dueTime)); + dueTime), + pushRetryBackoffMs(++hardSplitRetryTimes), + TimeUnit.MILLISECONDS); } else if (reason == StatusCode.PUSH_DATA_SUCCESS_PRIMARY_CONGESTED.getValue()) { logger.debug( "Push data to {} primary congestion required for shuffle {} map {} attempt {} partition {} batch {}.", @@ -1291,7 +1328,7 @@ public void onFailure(Throwable e) { long dueTime = System.currentTimeMillis() + conf.clientRpcRequestPartitionLocationAskTimeout().duration().toMillis(); - pushDataRetryPool.submit( + pushDataRetryPool.schedule( () -> submitRetryPushData( shuffleId, @@ -1301,7 +1338,9 @@ public void onFailure(Throwable e) { pushState, reviveRequest, remainReviveTimes, - dueTime)); + dueTime), + pushRetryBackoffMs(maxReviveTimes - remainReviveTimes), + TimeUnit.MILLISECONDS); } else { pushState.removeBatch(nextBatchId, latest.hostAndPushPort()); logger.info( @@ -1643,7 +1682,11 @@ public void onSuccess(ByteBuffer response) { ReviveRequest[] requests = addAndGetReviveRequests( shuffleId, mapId, attemptId, batchesNeedResubmit, StatusCode.HARD_SPLIT); - pushDataRetryPool.submit( + // remainReviveTimes is a parameter here and is not decremented on the HARD_SPLIT + // path, so there is no per-batch attempt counter to grow from. Apply the initial + // delay with jitter, which is what decorrelates the herd; growing this one too + // needs the counter threaded through doPushMergedData and is left as follow-up. + pushDataRetryPool.schedule( () -> submitRetryPushMergedData( pushState, @@ -1658,7 +1701,9 @@ public void onSuccess(ByteBuffer response) { System.currentTimeMillis() + conf.clientRpcRequestPartitionLocationAskTimeout() .duration() - .toMillis())); + .toMillis()), + pushRetryBackoffMs(1), + TimeUnit.MILLISECONDS); } } else if (reason == StatusCode.PUSH_DATA_SUCCESS_PRIMARY_CONGESTED.getValue()) { logger.debug( @@ -1731,7 +1776,7 @@ public void onFailure(Throwable e) { if (!mapperEnded(shuffleId, mapId)) { ReviveRequest[] requests = addAndGetReviveRequests(shuffleId, mapId, attemptId, batches, cause); - pushDataRetryPool.submit( + pushDataRetryPool.schedule( () -> submitRetryPushMergedData( pushState, @@ -1746,7 +1791,9 @@ public void onFailure(Throwable e) { System.currentTimeMillis() + conf.clientRpcRequestPartitionLocationAskTimeout() .duration() - .toMillis())); + .toMillis()), + pushRetryBackoffMs(maxReviveTimes - (remainReviveTimes - 1)), + TimeUnit.MILLISECONDS); } else { pushState.removeBatch(groupedBatchId, hostPort); logger.info( 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 1e34ac34f92..21ed36181bb 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -1094,6 +1094,9 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se def clientPushSortRandomizePartitionIdEnabled: Boolean = get(CLIENT_PUSH_SORT_RANDOMIZE_PARTITION_ENABLED) def clientPushRetryThreads: Int = get(CLIENT_PUSH_RETRY_THREADS) + def clientPushRetryBackoffInitial: Long = get(CLIENT_PUSH_RETRY_BACKOFF_INITIAL) + def clientPushRetryBackoffMax: Long = get(CLIENT_PUSH_RETRY_BACKOFF_MAX) + def clientPushRetryBackoffJitter: Double = get(CLIENT_PUSH_RETRY_BACKOFF_JITTER) def clientPushStageEndTimeout: Long = get(CLIENT_PUSH_STAGE_END_TIMEOUT) def clientPushUnsafeRowFastWrite: Boolean = get(CLIENT_PUSH_UNSAFEROW_FASTWRITE_ENABLED) def clientRpcCacheExpireTime: Long = get(CLIENT_RPC_CACHE_EXPIRE_TIME) @@ -5167,6 +5170,37 @@ object CelebornConf extends Logging { .intConf .createWithDefault(8) + val CLIENT_PUSH_RETRY_BACKOFF_MAX: ConfigEntry[Long] = + buildConf("celeborn.client.push.retry.backoff.max") + .categories("client") + .doc("Upper bound of the exponential push data retry delay.") + .version("0.7.1") + .timeConf(TimeUnit.MILLISECONDS) + .checkValue(_ > 0, "Value must be positive!") + .createWithDefaultString("10s") + + val CLIENT_PUSH_RETRY_BACKOFF_INITIAL: ConfigEntry[Long] = + buildConf("celeborn.client.push.retry.backoff.initial") + .categories("client") + .doc("Initial delay before re-sending a push data request that was rejected with " + + s"HARD_SPLIT or failed. The delay grows exponentially up to " + + s"`${CLIENT_PUSH_RETRY_BACKOFF_MAX.key}` so that map tasks blocked on the same " + + "partition do not retry in lockstep. Set to 0 to disable backoff.") + .version("0.7.1") + .timeConf(TimeUnit.MILLISECONDS) + .checkValue(_ >= 0, "Value must be zero or positive!") + .createWithDefaultString("500ms") + + val CLIENT_PUSH_RETRY_BACKOFF_JITTER: ConfigEntry[Double] = + buildConf("celeborn.client.push.retry.backoff.jitter") + .categories("client") + .doc("Fraction of the push data retry delay applied as random jitter. A value of 0.5 " + + "spreads retries uniformly over [0.5, 1.5] times the computed delay.") + .version("0.7.1") + .doubleConf + .checkValue(v => v >= 0.0 && v <= 1.0, "Value must be in [0.0, 1.0]!") + .createWithDefault(0.5) + val CLIENT_PUSH_TAKE_TASK_WAIT_INTERVAL: ConfigEntry[Long] = buildConf("celeborn.client.push.takeTaskWaitInterval") .categories("client") diff --git a/docs/configuration/client.md b/docs/configuration/client.md index dbb89561e43..a20a83d8554 100644 --- a/docs/configuration/client.md +++ b/docs/configuration/client.md @@ -66,6 +66,9 @@ license: | | celeborn.client.push.maxReqsInFlight.total | 256 | false | Amount of total Netty in-flight requests. The maximum memory is `celeborn.client.push.maxReqsInFlight.total` * `celeborn.client.push.buffer.max.size` * compression ratio(1 in worst case): 64KiB * 256 = 16MiB | 0.3.0 | celeborn.push.maxReqsInFlight | | celeborn.client.push.queue.capacity | 512 | false | Push buffer queue size for a task. The maximum memory is `celeborn.client.push.buffer.max.size` * `celeborn.client.push.queue.capacity`, default: 64KiB * 512 = 32MiB | 0.3.0 | celeborn.push.queue.capacity | | celeborn.client.push.replicate.enabled | false | false | When true, Celeborn worker will replicate shuffle data to another Celeborn worker asynchronously to ensure the pushed shuffle data won't be lost after the node failure. It's recommended to set `false` when `HDFS` is enabled in `celeborn.storage.availableTypes`. | 0.3.0 | celeborn.push.replicate.enabled | +| celeborn.client.push.retry.backoff.initial | 500ms | false | Initial delay before re-sending a push data request that was rejected with HARD_SPLIT or failed. The delay grows exponentially up to `celeborn.client.push.retry.backoff.max` so that map tasks blocked on the same partition do not retry in lockstep. Set to 0 to disable backoff. | 0.7.1 | | +| celeborn.client.push.retry.backoff.jitter | 0.5 | false | Fraction of the push data retry delay applied as random jitter. A value of 0.5 spreads retries uniformly over [0.5, 1.5] times the computed delay. | 0.7.1 | | +| celeborn.client.push.retry.backoff.max | 10s | false | Upper bound of the exponential push data retry delay. | 0.7.1 | | | celeborn.client.push.retry.threads | 8 | false | Thread number to process shuffle re-send push data requests. | 0.3.0 | celeborn.push.retry.threads | | celeborn.client.push.revive.batchSize | 2048 | false | Max number of partitions in one Revive request. | 0.3.0 | | | celeborn.client.push.revive.interval | 100ms | false | Interval for client to trigger Revive to LifecycleManager. The number of partitions in one Revive request is `celeborn.client.push.revive.batchSize`. | 0.3.0 | | From c88b05ff735951c8828494d71bf7ab998921d56d Mon Sep 17 00:00:00 2001 From: Venkata Krishnan Sowrirajan Date: Thu, 1 Oct 2026 05:26:33 +0000 Subject: [PATCH 3/6] [CELEBORN-XXXX] Add push retry backoff tests Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../celeborn/client/ShuffleClientImpl.java | 6 ++- .../celeborn/client/ShuffleClientSuiteJ.java | 41 +++++++++++++++++++ .../DiskMapPartitionDataWriterSuiteJ.java | 32 +++++++++++++++ 3 files changed, 77 insertions(+), 2 deletions(-) diff --git a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java index 2841888ea3b..b4cf4b47fc7 100644 --- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java +++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java @@ -317,14 +317,16 @@ private boolean isPushTargetWorkerExcluded( * {@code celeborn.client.push.retry.backoff.max}, spread by jitter so that map tasks blocked on * the same partition do not retry in lockstep. Returns 0 when backoff is disabled. */ - private long pushRetryBackoffMs(int retryTimes) { + @VisibleForTesting + long pushRetryBackoffMs(int retryTimes) { if (pushRetryBackoffInitialMs <= 0 || retryTimes <= 0) { return 0; } // Shift-loop rather than a power, because the hard split retry counter is not bounded. long delay = pushRetryBackoffInitialMs; for (int i = 1; i < retryTimes && delay < pushRetryBackoffMaxMs; i++) { - delay = delay << 1; + delay = + delay > pushRetryBackoffMaxMs / 2 ? pushRetryBackoffMaxMs : delay << 1; } delay = Math.min(delay, pushRetryBackoffMaxMs); long jitter = (long) (delay * pushRetryBackoffJitter); diff --git a/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java b/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java index 46d9a4f5b7c..7287d4fc640 100644 --- a/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java +++ b/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java @@ -112,6 +112,47 @@ public class ShuffleClientSuiteJ { private static final byte[] TEST_BUF1 = "hello world".getBytes(StandardCharsets.UTF_8); private final int BATCH_HEADER_SIZE = 4 * 4; + @Test + public void testPushRetryBackoff() { + CelebornConf conf = new CelebornConf(); + conf.set(CelebornConf.CLIENT_PUSH_RETRY_BACKOFF_INITIAL().key(), "100ms"); + conf.set(CelebornConf.CLIENT_PUSH_RETRY_BACKOFF_MAX().key(), "400ms"); + conf.set(CelebornConf.CLIENT_PUSH_RETRY_BACKOFF_JITTER().key(), "0"); + ShuffleClientImpl client = + new ShuffleClientImpl(TEST_APPLICATION_ID, conf, new UserIdentifier("mock", "mock")); + try { + assertEquals(0, client.pushRetryBackoffMs(0)); + assertEquals(100, client.pushRetryBackoffMs(1)); + assertEquals(200, client.pushRetryBackoffMs(2)); + assertEquals(400, client.pushRetryBackoffMs(3)); + assertEquals(400, client.pushRetryBackoffMs(Integer.MAX_VALUE)); + } finally { + client.shutdown(); + } + + conf.set(CelebornConf.CLIENT_PUSH_RETRY_BACKOFF_JITTER().key(), "0.5"); + client = new ShuffleClientImpl(TEST_APPLICATION_ID, conf, new UserIdentifier("mock", "mock")); + try { + Set delays = new java.util.HashSet<>(); + for (int i = 0; i < 100; i++) { + long delay = client.pushRetryBackoffMs(1); + assertTrue(delay >= 50 && delay <= 150); + delays.add(delay); + } + assertTrue("jitter should produce different retry delays", delays.size() > 1); + } finally { + client.shutdown(); + } + + conf.set(CelebornConf.CLIENT_PUSH_RETRY_BACKOFF_INITIAL().key(), "0"); + client = new ShuffleClientImpl(TEST_APPLICATION_ID, conf, new UserIdentifier("mock", "mock")); + try { + assertEquals(0, client.pushRetryBackoffMs(1)); + } finally { + client.shutdown(); + } + } + @Test public void testPushData() throws IOException, InterruptedException { for (CompressionCodec codec : CompressionCodec.values()) { diff --git a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/local/DiskMapPartitionDataWriterSuiteJ.java b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/local/DiskMapPartitionDataWriterSuiteJ.java index 5f750f89540..eff64812ff8 100644 --- a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/local/DiskMapPartitionDataWriterSuiteJ.java +++ b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/local/DiskMapPartitionDataWriterSuiteJ.java @@ -26,6 +26,7 @@ import java.nio.charset.StandardCharsets; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.IntStream; import scala.Function0; import scala.collection.mutable.ListBuffer; @@ -119,6 +120,37 @@ public static void afterAll() { } } + @Test + public void testShouldLogHardSplitOnlyOnce() throws IOException { + PartitionDataWriterContext context = + new PartitionDataWriterContext( + SPLIT_THRESHOLD, + splitMode, + false, + new PartitionLocation( + 1, 0, "host", 1111, 1112, 1113, 1114, PartitionLocation.Mode.PRIMARY, null), + "app1-1", + 1, + userIdentifier, + PartitionType.MAP, + false, + false); + PartitionDataWriter fileWriter = + new PartitionDataWriter( + PartitionDataWriterSuiteUtils.prepareDiskFileTestEnvironment( + tempDir, userIdentifier, localFlusher, false, CONF, storagePolicy, context), + source, + CONF, + DeviceMonitor$.MODULE$.EmptyMonitor(), + context); + try { + assertEquals( + 1, IntStream.range(0, 100).parallel().filter(i -> fileWriter.shouldLogHardSplit()).count()); + } finally { + fileWriter.close(); + } + } + @Test public void testMultiThreadWrite() throws IOException { PartitionLocation partitionLocation = Mockito.mock(PartitionLocation.class); From fddaf18dd085833ea53cd5698cba29f12e43c6b5 Mon Sep 17 00:00:00 2001 From: Venkata Krishnan Sowrirajan Date: Fri, 2 Oct 2026 16:06:41 +0000 Subject: [PATCH 4/6] [CELEBORN-2470] Test scheduled push retries Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../celeborn/client/ShuffleClientImpl.java | 33 +++-- .../celeborn/client/ShuffleClientSuiteJ.java | 132 +++++++++++++++++- .../DiskMapPartitionDataWriterSuiteJ.java | 3 +- 3 files changed, 149 insertions(+), 19 deletions(-) diff --git a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java index b4cf4b47fc7..b78b65232d0 100644 --- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java +++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java @@ -325,8 +325,7 @@ long pushRetryBackoffMs(int retryTimes) { // Shift-loop rather than a power, because the hard split retry counter is not bounded. long delay = pushRetryBackoffInitialMs; for (int i = 1; i < retryTimes && delay < pushRetryBackoffMaxMs; i++) { - delay = - delay > pushRetryBackoffMaxMs / 2 ? pushRetryBackoffMaxMs : delay << 1; + delay = delay > pushRetryBackoffMaxMs / 2 ? pushRetryBackoffMaxMs : delay << 1; } delay = Math.min(delay, pushRetryBackoffMaxMs); long jitter = (long) (delay * pushRetryBackoffJitter); @@ -336,6 +335,11 @@ long pushRetryBackoffMs(int retryTimes) { return delay - jitter + ThreadLocalRandom.current().nextLong(2 * jitter + 1); } + @VisibleForTesting + void scheduleRetry(Runnable retry, long delayMs) { + pushDataRetryPool.schedule(retry, delayMs, TimeUnit.MILLISECONDS); + } + private void submitRetryPushData( int shuffleId, byte[] body, @@ -564,7 +568,7 @@ private void submitRetryPushMergedData( } else { ReviveRequest[] requests = addAndGetReviveRequests(shuffleId, mapId, attemptId, reviveFailedBatchesMap, cause); - pushDataRetryPool.schedule( + scheduleRetry( () -> submitRetryPushMergedData( pushState, @@ -578,8 +582,7 @@ private void submitRetryPushMergedData( remainReviveTimes - 1, System.currentTimeMillis() + conf.clientRpcRequestPartitionLocationAskTimeout().duration().toMillis()), - pushRetryBackoffMs(maxReviveTimes - (remainReviveTimes - 1)), - TimeUnit.MILLISECONDS); + pushRetryBackoffMs(maxReviveTimes - (remainReviveTimes - 1))); } } @@ -1236,7 +1239,7 @@ public void onSuccess(ByteBuffer response) { + conf.clientRpcRequestPartitionLocationAskTimeout() .duration() .toMillis(); - pushDataRetryPool.schedule( + scheduleRetry( () -> submitRetryPushData( shuffleId, @@ -1247,8 +1250,7 @@ public void onSuccess(ByteBuffer response) { reviveRequest, remainReviveTimes, dueTime), - pushRetryBackoffMs(++hardSplitRetryTimes), - TimeUnit.MILLISECONDS); + pushRetryBackoffMs(++hardSplitRetryTimes)); } else if (reason == StatusCode.PUSH_DATA_SUCCESS_PRIMARY_CONGESTED.getValue()) { logger.debug( "Push data to {} primary congestion required for shuffle {} map {} attempt {} partition {} batch {}.", @@ -1330,7 +1332,7 @@ public void onFailure(Throwable e) { long dueTime = System.currentTimeMillis() + conf.clientRpcRequestPartitionLocationAskTimeout().duration().toMillis(); - pushDataRetryPool.schedule( + scheduleRetry( () -> submitRetryPushData( shuffleId, @@ -1341,8 +1343,7 @@ public void onFailure(Throwable e) { reviveRequest, remainReviveTimes, dueTime), - pushRetryBackoffMs(maxReviveTimes - remainReviveTimes), - TimeUnit.MILLISECONDS); + pushRetryBackoffMs(maxReviveTimes - remainReviveTimes)); } else { pushState.removeBatch(nextBatchId, latest.hostAndPushPort()); logger.info( @@ -1688,7 +1689,7 @@ public void onSuccess(ByteBuffer response) { // path, so there is no per-batch attempt counter to grow from. Apply the initial // delay with jitter, which is what decorrelates the herd; growing this one too // needs the counter threaded through doPushMergedData and is left as follow-up. - pushDataRetryPool.schedule( + scheduleRetry( () -> submitRetryPushMergedData( pushState, @@ -1704,8 +1705,7 @@ public void onSuccess(ByteBuffer response) { + conf.clientRpcRequestPartitionLocationAskTimeout() .duration() .toMillis()), - pushRetryBackoffMs(1), - TimeUnit.MILLISECONDS); + pushRetryBackoffMs(1)); } } else if (reason == StatusCode.PUSH_DATA_SUCCESS_PRIMARY_CONGESTED.getValue()) { logger.debug( @@ -1778,7 +1778,7 @@ public void onFailure(Throwable e) { if (!mapperEnded(shuffleId, mapId)) { ReviveRequest[] requests = addAndGetReviveRequests(shuffleId, mapId, attemptId, batches, cause); - pushDataRetryPool.schedule( + scheduleRetry( () -> submitRetryPushMergedData( pushState, @@ -1794,8 +1794,7 @@ public void onFailure(Throwable e) { + conf.clientRpcRequestPartitionLocationAskTimeout() .duration() .toMillis()), - pushRetryBackoffMs(maxReviveTimes - (remainReviveTimes - 1)), - TimeUnit.MILLISECONDS); + pushRetryBackoffMs(maxReviveTimes - (remainReviveTimes - 1))); } else { pushState.removeBatch(groupedBatchId, hostPort); logger.info( diff --git a/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java b/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java index 7287d4fc640..f5b23225991 100644 --- a/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java +++ b/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java @@ -20,10 +20,16 @@ import static org.junit.Assert.*; import static org.mockito.ArgumentMatchers.*; import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.IOException; +import java.lang.reflect.Field; +import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.HashMap; @@ -33,6 +39,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import scala.reflect.ClassTag; @@ -52,6 +59,7 @@ import org.apache.celeborn.common.CommitMetadata; import org.apache.celeborn.common.exception.CelebornIOException; import org.apache.celeborn.common.identity.UserIdentifier; +import org.apache.celeborn.common.network.client.RpcResponseCallback; import org.apache.celeborn.common.network.client.TransportClient; import org.apache.celeborn.common.network.client.TransportClientFactory; import org.apache.celeborn.common.network.protocol.SerdeVersion; @@ -59,6 +67,7 @@ import org.apache.celeborn.common.protocol.PartitionLocation; import org.apache.celeborn.common.protocol.PbReadReducerPartitionEnd; import org.apache.celeborn.common.protocol.PbReadReducerPartitionEndResponse; +import org.apache.celeborn.common.protocol.ReviveRequest; import org.apache.celeborn.common.protocol.message.ControlMessages.GetReducerFileGroupResponse$; import org.apache.celeborn.common.protocol.message.ControlMessages.RegisterShuffleResponse$; import org.apache.celeborn.common.protocol.message.StatusCode; @@ -153,6 +162,119 @@ public void testPushRetryBackoff() { } } + @Test + public void testHardSplitPushSchedulesRetryAndClearsBatchAfterRevive() throws Exception { + CelebornConf conf = retryIntegrationTestConf(); + setupEnv(conf, CompressionCodec.NONE, StatusCode.SUCCESS, false); + shuffleClient = spy(shuffleClient); + + AtomicReference scheduledRetry = new AtomicReference<>(); + AtomicLong scheduledDelay = new AtomicLong(-1); + doAnswer( + invocation -> { + scheduledRetry.set(invocation.getArgument(0)); + scheduledDelay.set(invocation.getArgument(1)); + return null; + }) + .when(shuffleClient) + .scheduleRetry(any(Runnable.class), anyLong()); + + shuffleClient.pushData( + TEST_SHUFFLE_ID, + TEST_MAP_ID, + TEST_ATTEMPT_ID, + TEST_REDUCRE_ID, + TEST_BUF1, + 0, + TEST_BUF1.length, + 1, + 1); + + ArgumentCaptor callbackCaptor = + ArgumentCaptor.forClass(RpcResponseCallback.class); + verify(client).pushData(any(), anyLong(), callbackCaptor.capture()); + callbackCaptor + .getValue() + .onSuccess(ByteBuffer.wrap(new byte[] {StatusCode.HARD_SPLIT.getValue()})); + + assertEquals(100, scheduledDelay.get()); + ReviveRequest reviveRequest = takeReviveRequest(shuffleClient); + reviveRequest.reviveStatus = StatusCode.SUCCESS.getValue(); + scheduledRetry.get().run(); + + verify(client, times(2)).pushData(any(), anyLong(), callbackCaptor.capture()); + callbackCaptor + .getAllValues() + .get(callbackCaptor.getAllValues().size() - 1) + .onSuccess(ByteBuffer.allocate(0)); + + PushState pushState = + shuffleClient.getPushState(Utils.makeMapKey(TEST_SHUFFLE_ID, TEST_MAP_ID, TEST_ATTEMPT_ID)); + assertFalse(pushState.limitZeroInFlight()); + } + + @Test + public void testHardSplitMergedPushSchedulesRetryAfterRevive() throws Exception { + CelebornConf conf = retryIntegrationTestConf(); + setupEnv(conf, CompressionCodec.NONE, StatusCode.SUCCESS, false); + shuffleClient = spy(shuffleClient); + + AtomicReference scheduledRetry = new AtomicReference<>(); + AtomicLong scheduledDelay = new AtomicLong(-1); + doAnswer( + invocation -> { + scheduledRetry.set(invocation.getArgument(0)); + scheduledDelay.set(invocation.getArgument(1)); + return null; + }) + .when(shuffleClient) + .scheduleRetry(any(Runnable.class), anyLong()); + + shuffleClient.mergeData( + TEST_SHUFFLE_ID, + TEST_MAP_ID, + TEST_ATTEMPT_ID, + TEST_REDUCRE_ID, + TEST_BUF1, + 0, + TEST_BUF1.length, + 1, + 1); + shuffleClient.pushMergedData(TEST_SHUFFLE_ID, TEST_MAP_ID, TEST_ATTEMPT_ID); + + ArgumentCaptor callbackCaptor = + ArgumentCaptor.forClass(RpcResponseCallback.class); + verify(client).pushMergedData(any(), anyLong(), callbackCaptor.capture()); + callbackCaptor + .getValue() + .onSuccess(ByteBuffer.wrap(new byte[] {StatusCode.HARD_SPLIT.getValue()})); + + assertEquals(100, scheduledDelay.get()); + ReviveRequest reviveRequest = takeReviveRequest(shuffleClient); + reviveRequest.reviveStatus = StatusCode.SUCCESS.getValue(); + scheduledRetry.get().run(); + + verify(client, times(2)).pushMergedData(any(), anyLong(), any()); + } + + private CelebornConf retryIntegrationTestConf() { + CelebornConf conf = new CelebornConf(); + conf.set(CelebornConf.CLIENT_PUSH_RETRY_BACKOFF_INITIAL().key(), "100ms"); + conf.set(CelebornConf.CLIENT_PUSH_RETRY_BACKOFF_MAX().key(), "400ms"); + conf.set(CelebornConf.CLIENT_PUSH_RETRY_BACKOFF_JITTER().key(), "0"); + conf.set(CelebornConf.CLIENT_PUSH_REVIVE_INTERVAL().key(), "1h"); + return conf; + } + + private ReviveRequest takeReviveRequest(ShuffleClientImpl client) throws Exception { + Field reviveManagerField = ShuffleClientImpl.class.getDeclaredField("reviveManager"); + reviveManagerField.setAccessible(true); + ReviveManager reviveManager = (ReviveManager) reviveManagerField.get(client); + ReviveRequest request = reviveManager.requestQueue.poll(); + assertNotNull(request); + return request; + } + @Test public void testPushData() throws IOException, InterruptedException { for (CompressionCodec codec : CompressionCodec.values()) { @@ -297,7 +419,15 @@ private CelebornConf setupEnv(CompressionCodec codec, StatusCode statusCode) private CelebornConf setupEnv( CompressionCodec codec, StatusCode statusCode, boolean interruptWhenPushData) throws IOException, InterruptedException { - CelebornConf conf = new CelebornConf(); + return setupEnv(new CelebornConf(), codec, statusCode, interruptWhenPushData); + } + + private CelebornConf setupEnv( + CelebornConf conf, + CompressionCodec codec, + StatusCode statusCode, + boolean interruptWhenPushData) + throws IOException, InterruptedException { conf.set(CelebornConf.SHUFFLE_COMPRESSION_CODEC().key(), codec.name()); conf.set(CelebornConf.CLIENT_PUSH_RETRY_THREADS().key(), "1"); conf.set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE().key(), "1K"); diff --git a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/local/DiskMapPartitionDataWriterSuiteJ.java b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/local/DiskMapPartitionDataWriterSuiteJ.java index eff64812ff8..94b7eec5e9a 100644 --- a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/local/DiskMapPartitionDataWriterSuiteJ.java +++ b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/local/DiskMapPartitionDataWriterSuiteJ.java @@ -145,7 +145,8 @@ public void testShouldLogHardSplitOnlyOnce() throws IOException { context); try { assertEquals( - 1, IntStream.range(0, 100).parallel().filter(i -> fileWriter.shouldLogHardSplit()).count()); + 1, + IntStream.range(0, 100).parallel().filter(i -> fileWriter.shouldLogHardSplit()).count()); } finally { fileWriter.close(); } From ecd7f65bf95c6a79b9169060880c62e6b86eb55b Mon Sep 17 00:00:00 2001 From: Venkata Krishnan Sowrirajan Date: Sun, 11 Oct 2026 17:04:06 +0000 Subject: [PATCH 5/6] [CELEBORN-2470] Address retry and split review feedback Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../org/apache/celeborn/client/ShuffleClientImpl.java | 8 +++++++- .../org/apache/celeborn/client/ShuffleClientSuiteJ.java | 8 ++++++++ .../celeborn/service/deploy/worker/PushDataHandler.scala | 6 ++++-- .../service/deploy/worker/PushDataHandlerSuite.scala | 9 +++++++-- 4 files changed, 26 insertions(+), 5 deletions(-) diff --git a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java index b78b65232d0..f3219247372 100644 --- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java +++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java @@ -332,7 +332,13 @@ long pushRetryBackoffMs(int retryTimes) { if (jitter <= 0) { return delay; } - return delay - jitter + ThreadLocalRandom.current().nextLong(2 * jitter + 1); + long minDelay = delay - jitter; + long maxDelay = delay > pushRetryBackoffMaxMs - jitter ? pushRetryBackoffMaxMs : delay + jitter; + long jitterRange = maxDelay - minDelay; + if (jitterRange == Long.MAX_VALUE) { + return ThreadLocalRandom.current().nextLong(Long.MAX_VALUE); + } + return minDelay + ThreadLocalRandom.current().nextLong(jitterRange + 1); } @VisibleForTesting diff --git a/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java b/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java index f5b23225991..82da671a310 100644 --- a/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java +++ b/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java @@ -149,6 +149,14 @@ public void testPushRetryBackoff() { delays.add(delay); } assertTrue("jitter should produce different retry delays", delays.size() > 1); + + delays.clear(); + for (int i = 0; i < 100; i++) { + long delay = client.pushRetryBackoffMs(3); + assertTrue(delay >= 200 && delay <= 400); + delays.add(delay); + } + assertTrue("jitter at the cap should produce different retry delays", delays.size() > 1); } finally { client.shutdown(); } diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala index 3ef9a75115a..7c64593d257 100644 --- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala +++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala @@ -1442,8 +1442,10 @@ class PushDataHandler(val workerSource: WorkerSource) extends BaseMessageHandler fileWriter: PartitionDataWriter, isPrimary: Boolean): StatusCode = { if (fileWriter.needHardSplitForMemoryShuffleStorage()) { - logInfo( - s"Do hardSplit for memory shuffle file fileLength:${fileWriter.getMemoryFileInfo.getFileLength}") + if (fileWriter.shouldLogHardSplit()) { + logInfo( + s"Do hardSplit for memory shuffle file fileLength:${fileWriter.getMemoryFileInfo.getFileLength}") + } return StatusCode.HARD_SPLIT } val diskFull = checkDiskFull(fileWriter) diff --git a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/PushDataHandlerSuite.scala b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/PushDataHandlerSuite.scala index 40639a24880..2e010de67c6 100644 --- a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/PushDataHandlerSuite.scala +++ b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/PushDataHandlerSuite.scala @@ -148,10 +148,15 @@ class PushDataHandlerSuite extends AnyFunSuite { test("returns HARD_SPLIT when memory shuffle storage needs hard split") { val handler = buildHandler(partitionSplitEnabled = true) val fileWriter = buildFileWriter(fileLength = 0L, needMemoryHardSplit = true) + when(fileWriter.shouldLogHardSplit()).thenReturn(true, false) - val result = handler.checkDiskFullAndSplit(fileWriter, isPrimary = true) + val firstResult = handler.checkDiskFullAndSplit(fileWriter, isPrimary = true) + val secondResult = handler.checkDiskFullAndSplit(fileWriter, isPrimary = true) - assert(result === StatusCode.HARD_SPLIT) + assert(firstResult === StatusCode.HARD_SPLIT) + assert(secondResult === StatusCode.HARD_SPLIT) + verify(fileWriter, times(2)).shouldLogHardSplit() + verify(fileWriter, times(1)).getMemoryFileInfo } test("returns NO_SPLIT when workerPartitionSplitEnabled is false") { From 79fa9fd7f86340f709635c26e1b117e574fdcf5b Mon Sep 17 00:00:00 2001 From: Venkata Krishnan Sowrirajan Date: Tue, 13 Oct 2026 03:04:48 +0000 Subject: [PATCH 6/6] [CELEBORN-2470] Document that backoff.max caps the jittered retry delay Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../scala/org/apache/celeborn/common/CelebornConf.scala | 6 ++++-- docs/configuration/client.md | 4 ++-- 2 files changed, 6 insertions(+), 4 deletions(-) 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 21ed36181bb..d59df647cd1 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -5173,7 +5173,7 @@ object CelebornConf extends Logging { val CLIENT_PUSH_RETRY_BACKOFF_MAX: ConfigEntry[Long] = buildConf("celeborn.client.push.retry.backoff.max") .categories("client") - .doc("Upper bound of the exponential push data retry delay.") + .doc("Upper bound of the push data retry delay, including jitter.") .version("0.7.1") .timeConf(TimeUnit.MILLISECONDS) .checkValue(_ > 0, "Value must be positive!") @@ -5195,7 +5195,9 @@ object CelebornConf extends Logging { buildConf("celeborn.client.push.retry.backoff.jitter") .categories("client") .doc("Fraction of the push data retry delay applied as random jitter. A value of 0.5 " + - "spreads retries uniformly over [0.5, 1.5] times the computed delay.") + "spreads retries uniformly over [0.5, 1.5] times the computed delay, capped at " + + s"`${CLIENT_PUSH_RETRY_BACKOFF_MAX.key}`. For example, once the computed delay reaches " + + "the default maximum of 10s, retries are spread over [5s, 10s].") .version("0.7.1") .doubleConf .checkValue(v => v >= 0.0 && v <= 1.0, "Value must be in [0.0, 1.0]!") diff --git a/docs/configuration/client.md b/docs/configuration/client.md index a20a83d8554..c5957be652c 100644 --- a/docs/configuration/client.md +++ b/docs/configuration/client.md @@ -67,8 +67,8 @@ license: | | celeborn.client.push.queue.capacity | 512 | false | Push buffer queue size for a task. The maximum memory is `celeborn.client.push.buffer.max.size` * `celeborn.client.push.queue.capacity`, default: 64KiB * 512 = 32MiB | 0.3.0 | celeborn.push.queue.capacity | | celeborn.client.push.replicate.enabled | false | false | When true, Celeborn worker will replicate shuffle data to another Celeborn worker asynchronously to ensure the pushed shuffle data won't be lost after the node failure. It's recommended to set `false` when `HDFS` is enabled in `celeborn.storage.availableTypes`. | 0.3.0 | celeborn.push.replicate.enabled | | celeborn.client.push.retry.backoff.initial | 500ms | false | Initial delay before re-sending a push data request that was rejected with HARD_SPLIT or failed. The delay grows exponentially up to `celeborn.client.push.retry.backoff.max` so that map tasks blocked on the same partition do not retry in lockstep. Set to 0 to disable backoff. | 0.7.1 | | -| celeborn.client.push.retry.backoff.jitter | 0.5 | false | Fraction of the push data retry delay applied as random jitter. A value of 0.5 spreads retries uniformly over [0.5, 1.5] times the computed delay. | 0.7.1 | | -| celeborn.client.push.retry.backoff.max | 10s | false | Upper bound of the exponential push data retry delay. | 0.7.1 | | +| celeborn.client.push.retry.backoff.jitter | 0.5 | false | Fraction of the push data retry delay applied as random jitter. A value of 0.5 spreads retries uniformly over [0.5, 1.5] times the computed delay, capped at `celeborn.client.push.retry.backoff.max`. For example, once the computed delay reaches the default maximum of 10s, retries are spread over [5s, 10s]. | 0.7.1 | | +| celeborn.client.push.retry.backoff.max | 10s | false | Upper bound of the push data retry delay, including jitter. | 0.7.1 | | | celeborn.client.push.retry.threads | 8 | false | Thread number to process shuffle re-send push data requests. | 0.3.0 | celeborn.push.retry.threads | | celeborn.client.push.revive.batchSize | 2048 | false | Max number of partitions in one Revive request. | 0.3.0 | | | celeborn.client.push.revive.interval | 100ms | false | Interval for client to trigger Revive to LifecycleManager. The number of partitions in one Revive request is `celeborn.client.push.revive.batchSize`. | 0.3.0 | |