This is an automated email from the ASF dual-hosted git repository.
zaynt4606 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new b34cd83a58 [CELEBORN-2470] Add backoff with jitter to push retries and
log hard split once per file
b34cd83a58 is described below
commit b34cd83a58d0dc0d7f579ea5476b568aeb56d5fb
Author: Venkata Krishnan Sowrirajan <[email protected]>
AuthorDate: Tue Oct 13 11:34:28 2026 +0800
[CELEBORN-2470] Add backoff with jitter to push retries and log hard split
once per file
JIRA: [CELEBORN-2470](https://issues.apache.org/jira/browse/CELEBORN-2470)
## Summary
Push retries currently run immediately and can synchronize across many
tasks, increasing connection and DNS load when a worker is unhealthy. Hard
splits can also emit the same INFO log repeatedly while pushes continue
targeting the old file.
This PR:
- adds exponential backoff with jitter to push retries
- logs the hard-split cause once per `PartitionDataWriter`, for both disk
and memory files
- moves the repeated merged-push hard-split message to DEBUG
## Configuration
| Config | Default |
|---|---|
| `celeborn.client.push.retry.backoff.initial` | `500ms` |
| `celeborn.client.push.retry.backoff.max` | `10s` |
| `celeborn.client.push.retry.backoff.jitter` | `0.5` |
Setting `celeborn.client.push.retry.backoff.initial` to `0` disables
backoff. `celeborn.client.push.retry.backoff.max` caps the final delay,
including jitter, so with the defaults a retry at the cap waits between 5s and
10s.
## Tests
- added coverage for exponential growth, maximum delay, jitter bounds
(including at the cap), disabled backoff, and large retry counts
- added callback-level coverage for scheduled single and merged
`HARD_SPLIT` retries and revive completion during backoff
- added concurrent coverage verifying the hard-split log latch triggers
once per writer
- added coverage that repeated memory-shuffle `HARD_SPLIT` checks log only
once
Closes #3856 from venkata91/celeborn-push-retry-backoff.
Authored-by: Venkata Krishnan Sowrirajan <[email protected]>
Signed-off-by: zhengtao <[email protected]>
AI-Contributed/Feature: 0/143
AI-Contributed/UT: 0/223
Co-authored-by: Copilot <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../apache/celeborn/client/ShuffleClientImpl.java | 80 +++++++--
.../celeborn/client/ShuffleClientSuiteJ.java | 181 ++++++++++++++++++++-
.../org/apache/celeborn/common/CelebornConf.scala | 36 ++++
docs/configuration/client.md | 3 +
.../deploy/worker/storage/PartitionDataWriter.java | 9 +
.../service/deploy/worker/PushDataHandler.scala | 15 +-
.../local/DiskMapPartitionDataWriterSuiteJ.java | 33 ++++
.../deploy/worker/PushDataHandlerSuite.scala | 9 +-
8 files changed, 345 insertions(+), 21 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 5e84f84ee4..f321924737 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.nio.ByteBuffer;
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<Integer, Set<Integer>> splitting =
JavaUtils.newConcurrentHashMap();
@@ -235,7 +239,11 @@ public class ShuffleClientImpl extends ShuffleClient {
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,40 @@ public class ShuffleClientImpl extends ShuffleClient {
}
}
+ /**
+ * 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.
+ */
+ @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 > pushRetryBackoffMaxMs / 2 ? pushRetryBackoffMaxMs :
delay << 1;
+ }
+ delay = Math.min(delay, pushRetryBackoffMaxMs);
+ long jitter = (long) (delay * pushRetryBackoffJitter);
+ if (jitter <= 0) {
+ return delay;
+ }
+ 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
+ void scheduleRetry(Runnable retry, long delayMs) {
+ pushDataRetryPool.schedule(retry, delayMs, TimeUnit.MILLISECONDS);
+ }
+
private void submitRetryPushData(
int shuffleId,
byte[] body,
@@ -532,7 +574,7 @@ public class ShuffleClientImpl extends ShuffleClient {
} else {
ReviveRequest[] requests =
addAndGetReviveRequests(shuffleId, mapId, attemptId,
reviveFailedBatchesMap, cause);
- pushDataRetryPool.submit(
+ scheduleRetry(
() ->
submitRetryPushMergedData(
pushState,
@@ -545,7 +587,8 @@ public class ShuffleClientImpl extends ShuffleClient {
requests,
remainReviveTimes - 1,
System.currentTimeMillis()
- +
conf.clientRpcRequestPartitionLocationAskTimeout().duration().toMillis()));
+ +
conf.clientRpcRequestPartitionLocationAskTimeout().duration().toMillis()),
+ pushRetryBackoffMs(maxReviveTimes - (remainReviveTimes - 1)));
}
}
@@ -1133,6 +1176,9 @@ public class ShuffleClientImpl extends ShuffleClient {
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 +1245,7 @@ public class ShuffleClientImpl extends ShuffleClient {
+ conf.clientRpcRequestPartitionLocationAskTimeout()
.duration()
.toMillis();
- pushDataRetryPool.submit(
+ scheduleRetry(
() ->
submitRetryPushData(
shuffleId,
@@ -1209,7 +1255,8 @@ public class ShuffleClientImpl extends ShuffleClient {
pushState,
reviveRequest,
remainReviveTimes,
- dueTime));
+ dueTime),
+ 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 {}.",
@@ -1291,7 +1338,7 @@ public class ShuffleClientImpl extends ShuffleClient {
long dueTime =
System.currentTimeMillis()
+
conf.clientRpcRequestPartitionLocationAskTimeout().duration().toMillis();
- pushDataRetryPool.submit(
+ scheduleRetry(
() ->
submitRetryPushData(
shuffleId,
@@ -1301,7 +1348,8 @@ public class ShuffleClientImpl extends ShuffleClient {
pushState,
reviveRequest,
remainReviveTimes,
- dueTime));
+ dueTime),
+ pushRetryBackoffMs(maxReviveTimes - remainReviveTimes));
} else {
pushState.removeBatch(nextBatchId, latest.hostAndPushPort());
logger.info(
@@ -1643,7 +1691,11 @@ public class ShuffleClientImpl extends ShuffleClient {
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.
+ scheduleRetry(
() ->
submitRetryPushMergedData(
pushState,
@@ -1658,7 +1710,8 @@ public class ShuffleClientImpl extends ShuffleClient {
System.currentTimeMillis()
+
conf.clientRpcRequestPartitionLocationAskTimeout()
.duration()
- .toMillis()));
+ .toMillis()),
+ pushRetryBackoffMs(1));
}
} else if (reason ==
StatusCode.PUSH_DATA_SUCCESS_PRIMARY_CONGESTED.getValue()) {
logger.debug(
@@ -1731,7 +1784,7 @@ public class ShuffleClientImpl extends ShuffleClient {
if (!mapperEnded(shuffleId, mapId)) {
ReviveRequest[] requests =
addAndGetReviveRequests(shuffleId, mapId, attemptId,
batches, cause);
- pushDataRetryPool.submit(
+ scheduleRetry(
() ->
submitRetryPushMergedData(
pushState,
@@ -1746,7 +1799,8 @@ public class ShuffleClientImpl extends ShuffleClient {
System.currentTimeMillis()
+
conf.clientRpcRequestPartitionLocationAskTimeout()
.duration()
- .toMillis()));
+ .toMillis()),
+ 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 46d9a4f5b7..82da671a31 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 @@ package org.apache.celeborn.client;
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.CountDownLatch;
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.CelebornConf;
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.CompressionCodec;
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;
@@ -112,6 +121,168 @@ 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<Long> 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);
+
+ 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();
+ }
+
+ 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 testHardSplitPushSchedulesRetryAndClearsBatchAfterRevive()
throws Exception {
+ CelebornConf conf = retryIntegrationTestConf();
+ setupEnv(conf, CompressionCodec.NONE, StatusCode.SUCCESS, false);
+ shuffleClient = spy(shuffleClient);
+
+ AtomicReference<Runnable> 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<RpcResponseCallback> 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<Runnable> 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<RpcResponseCallback> 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()) {
@@ -256,7 +427,15 @@ public class ShuffleClientSuiteJ {
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/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 3391e3d169..22d1a066db 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -1097,6 +1097,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)
@@ -5195,6 +5198,39 @@ 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 push data retry delay, including jitter.")
+ .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,
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]!")
+ .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 dbb89561e4..c5957be652 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, 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 | |
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 720423829b..233b7a0efc 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 @@ package org.apache.celeborn.service.deploy.worker.storage;
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 class PartitionDataWriter implements DeviceObserver
{
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 synchronized void notifyError(String mountPoint, DiskStatus
diskStatus) {
// A committed file is already flushed and visible to fetch requests,
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 f6c32e77f3..7c64593d25 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) {
@@ -1441,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)
@@ -1470,7 +1473,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
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 5f750f8954..94b7eec5e9 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.ByteOrder;
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,38 @@ public class DiskMapPartitionDataWriterSuiteJ {
}
}
+ @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);
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 40639a2488..2e010de67c 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") {
