You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by ni...@apache.org on 2026/04/03 10:00:56 UTC
(celeborn) branch branch-0.6 updated: [CELEBORN-2295] CommitHandler should support retry interval
This is an automated email from the ASF dual-hosted git repository.
nicholasjiang pushed a commit to branch branch-0.6
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-0.6 by this push:
new 244cfc6d7 [CELEBORN-2295] CommitHandler should support retry interval
244cfc6d7 is described below
commit 244cfc6d75e3bf7fa16272ecfe8cdb6f4283e053
Author: Xianming Lei <je...@apache.org>
AuthorDate: Fri Apr 3 18:00:01 2026 +0800
[CELEBORN-2295] CommitHandler should support retry interval
### What changes were proposed in this pull request?
`CommitHandler` should support retry interval for retry of committing file.
### Why are the changes needed?
When commitFiles RPC fails, the current implementation retries immediately without any backoff. If the worker is experiencing transient network issues, immediate retries are likely to fail again. Adding a configurable retry interval (`celeborn.client.requestCommitFiles.retryInterval`, default 10s) gives the worker time to recover before the next attempt, significantly improving the success rate of retries. A dedicated `ScheduledExecutorService` is used to avoid blocking threads in the [...]
### Does this PR resolve a correctness bug?
No.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
CI.
Closes #3642 from leixm/CELEBORN-2295.
Lead-authored-by: Xianming Lei <je...@apache.org>
Co-authored-by: Xianming Lei <xi...@shopee.com>
Signed-off-by: SteNicholas <pr...@163.com>
(cherry picked from commit 37aed7ea5421e0de9bfdfbd452746103a579232e)
Signed-off-by: SteNicholas <pr...@163.com>
---
.../org/apache/celeborn/client/CommitManager.scala | 7 ++-
.../celeborn/client/commit/CommitHandler.scala | 56 ++++++++++++++++++++--
.../client/commit/MapPartitionCommitHandler.scala | 13 +++--
.../commit/ReducePartitionCommitHandler.scala | 6 ++-
.../org/apache/celeborn/common/CelebornConf.scala | 9 ++++
docs/configuration/client.md | 1 +
6 files changed, 81 insertions(+), 11 deletions(-)
diff --git a/client/src/main/scala/org/apache/celeborn/client/CommitManager.scala b/client/src/main/scala/org/apache/celeborn/client/CommitManager.scala
index 1a71d3ccd..b9a937d6f 100644
--- a/client/src/main/scala/org/apache/celeborn/client/CommitManager.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/CommitManager.scala
@@ -92,6 +92,8 @@ class CommitManager(appUniqueId: String, val conf: CelebornConf, lifecycleManage
None
}
private var batchHandleCommitPartition: Option[ScheduledFuture[_]] = _
+ private val commitRetryScheduler: ScheduledExecutorService =
+ ThreadUtils.newDaemonSingleThreadScheduledExecutor("commit-retry-scheduler")
private val commitHandlers = JavaUtils.newConcurrentHashMap[PartitionType, CommitHandler]()
def start(): Unit = {
@@ -173,6 +175,7 @@ class CommitManager(appUniqueId: String, val conf: CelebornConf, lifecycleManage
def stop(): Unit = {
batchHandleCommitPartition.foreach(_.cancel(true))
batchHandleCommitPartitionSchedulerThread.foreach(ThreadUtils.shutdown(_))
+ ThreadUtils.shutdown(commitRetryScheduler)
}
def registerShuffle(
@@ -308,6 +311,7 @@ class CommitManager(appUniqueId: String, val conf: CelebornConf, lifecycleManage
committedPartitionInfo,
lifecycleManager.workerStatusTracker,
lifecycleManager.rpcSharedThreadPool,
+ commitRetryScheduler,
lifecycleManager)
case PartitionType.MAP => new MapPartitionCommitHandler(
appUniqueId,
@@ -315,7 +319,8 @@ class CommitManager(appUniqueId: String, val conf: CelebornConf, lifecycleManage
lifecycleManager.shuffleAllocatedWorkers,
committedPartitionInfo,
lifecycleManager.workerStatusTracker,
- lifecycleManager.rpcSharedThreadPool)
+ lifecycleManager.rpcSharedThreadPool,
+ commitRetryScheduler)
case _ => throw new UnsupportedOperationException(
s"Unexpected ShufflePartitionType for CommitManager: $partitionType")
}
diff --git a/client/src/main/scala/org/apache/celeborn/client/commit/CommitHandler.scala b/client/src/main/scala/org/apache/celeborn/client/commit/CommitHandler.scala
index 735ebb99c..4795faea1 100644
--- a/client/src/main/scala/org/apache/celeborn/client/commit/CommitHandler.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/commit/CommitHandler.scala
@@ -18,14 +18,14 @@
package org.apache.celeborn.client.commit
import java.util
-import java.util.concurrent.{ConcurrentHashMap, LinkedBlockingQueue, ThreadPoolExecutor, TimeUnit}
+import java.util.concurrent.{ConcurrentHashMap, LinkedBlockingQueue, ScheduledExecutorService, ThreadPoolExecutor, TimeUnit}
import java.util.concurrent.atomic.{AtomicLong, LongAdder}
import scala.collection.JavaConverters._
import scala.collection.generic.CanBuildFrom
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
-import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.concurrent.duration.Duration
import org.apache.celeborn.client.{ShuffleCommittedInfo, WorkerStatusTracker}
@@ -67,7 +67,8 @@ abstract class CommitHandler(
conf: CelebornConf,
committedPartitionInfo: CommittedPartitionInfo,
workerStatusTracker: WorkerStatusTracker,
- val sharedRpcPool: ThreadPoolExecutor) extends Logging {
+ val sharedRpcPool: ThreadPoolExecutor,
+ commitRetryScheduler: ScheduledExecutorService) extends Logging {
private val pushReplicateEnabled = conf.clientPushReplicateEnabled
private val clientRpcCommitFilesAskTimeout = conf.clientRpcCommitFilesAskTimeout
@@ -81,6 +82,7 @@ abstract class CommitHandler(
val ec = ExecutionContext.fromExecutor(sharedRpcPool)
val maxRetries = conf.clientRequestCommitFilesMaxRetries
+ val retryWait = conf.clientRequestCommitFilesRetryWait
val mockCommitFilesFailure = conf.testMockCommitFilesFailure
def getPartitionType(): PartitionType
@@ -250,7 +252,8 @@ abstract class CommitHandler(
status.message.copy(mockFailure = mockFailure)
status.future = commitFiles(
status.workerInfo,
- msg)
+ msg,
+ isRetry = true)
}
def createFailResponse(status: CommitFutureWithStatus): CommitFilesResponse = {
@@ -498,7 +501,8 @@ abstract class CommitHandler(
def commitFiles(
worker: WorkerInfo,
- message: CommitFiles): Future[CommitFilesResponse] = {
+ message: CommitFiles,
+ isRetry: Boolean = false): Future[CommitFilesResponse] = {
if (conf.clientCommitFilesIgnoreExcludedWorkers &&
workerStatusTracker.excludedWorkers.containsKey(worker)) {
@@ -510,6 +514,48 @@ abstract class CommitHandler(
message.primaryIds,
message.replicaIds)
}(ec)
+ } else if (isRetry) {
+ val promise = Promise[CommitFilesResponse]()
+ try {
+ commitRetryScheduler.schedule(
+ new Runnable {
+ override def run(): Unit = {
+ try {
+ worker.endpoint.ask[CommitFilesResponse](message, clientRpcCommitFilesAskTimeout)
+ .onComplete {
+ result => promise.complete(result)
+ }(ec)
+ } catch {
+ case e: Exception =>
+ logError(
+ s"Failed to send CommitFiles to worker($worker) " +
+ s"for ${message.shuffleId} after retry delay.",
+ e)
+ promise.success(CommitFilesResponse(
+ StatusCode.REQUEST_FAILED,
+ List.empty.asJava,
+ List.empty.asJava,
+ message.primaryIds,
+ message.replicaIds))
+ }
+ }
+ },
+ retryWait,
+ TimeUnit.MILLISECONDS)
+ } catch {
+ case e: Exception =>
+ logError(
+ s"Failed to schedule CommitFiles retry to worker($worker) " +
+ s"for ${message.shuffleId}.",
+ e)
+ promise.success(CommitFilesResponse(
+ StatusCode.REQUEST_FAILED,
+ List.empty.asJava,
+ List.empty.asJava,
+ message.primaryIds,
+ message.replicaIds))
+ }
+ promise.future
} else {
worker.endpoint.ask[CommitFilesResponse](message, clientRpcCommitFilesAskTimeout)
}
diff --git a/client/src/main/scala/org/apache/celeborn/client/commit/MapPartitionCommitHandler.scala b/client/src/main/scala/org/apache/celeborn/client/commit/MapPartitionCommitHandler.scala
index 490b9450a..4ae01139b 100644
--- a/client/src/main/scala/org/apache/celeborn/client/commit/MapPartitionCommitHandler.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/commit/MapPartitionCommitHandler.scala
@@ -19,7 +19,7 @@ package org.apache.celeborn.client.commit
import java.util
import java.util.Collections
-import java.util.concurrent.{ConcurrentHashMap, ThreadPoolExecutor}
+import java.util.concurrent.{ConcurrentHashMap, ScheduledExecutorService, ThreadPoolExecutor}
import java.util.concurrent.atomic.{AtomicInteger, AtomicIntegerArray}
import scala.collection.JavaConverters._
@@ -57,8 +57,15 @@ class MapPartitionCommitHandler(
shuffleAllocatedWorkers: ShuffleAllocatedWorkers,
committedPartitionInfo: CommittedPartitionInfo,
workerStatusTracker: WorkerStatusTracker,
- sharedRpcPool: ThreadPoolExecutor)
- extends CommitHandler(appId, conf, committedPartitionInfo, workerStatusTracker, sharedRpcPool)
+ sharedRpcPool: ThreadPoolExecutor,
+ commitRetryScheduler: ScheduledExecutorService)
+ extends CommitHandler(
+ appId,
+ conf,
+ committedPartitionInfo,
+ workerStatusTracker,
+ sharedRpcPool,
+ commitRetryScheduler)
with Logging {
private val shuffleSucceedPartitionIds = JavaUtils.newConcurrentHashMap[Int, util.Set[Integer]]()
diff --git a/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala b/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala
index 7fe781da6..0f3b77524 100644
--- a/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala
@@ -19,7 +19,7 @@ package org.apache.celeborn.client.commit
import java.nio.ByteBuffer
import java.util
-import java.util.concurrent.{Callable, ConcurrentHashMap, ThreadPoolExecutor, TimeUnit}
+import java.util.concurrent.{Callable, ConcurrentHashMap, ScheduledExecutorService, ThreadPoolExecutor, TimeUnit}
import java.util.function
import scala.collection.JavaConverters._
@@ -58,13 +58,15 @@ class ReducePartitionCommitHandler(
committedPartitionInfo: CommittedPartitionInfo,
workerStatusTracker: WorkerStatusTracker,
sharedRpcPool: ThreadPoolExecutor,
+ commitRetryScheduler: ScheduledExecutorService,
lifecycleManager: LifecycleManager)
extends CommitHandler(
appUniqueId,
conf,
committedPartitionInfo,
workerStatusTracker,
- sharedRpcPool)
+ sharedRpcPool,
+ commitRetryScheduler)
with Logging {
class MultiSerdeVersionRpcContext(val ctx: RpcCallContext, val serdeVersion: SerdeVersion) {}
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 6a5f0ed7e..1dff51cfa 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -933,6 +933,7 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
def clientReserveSlotsRackAwareEnabled: Boolean = get(CLIENT_RESERVE_SLOTS_RACKAWARE_ENABLED)
def clientReserveSlotsMaxRetries: Int = get(CLIENT_RESERVE_SLOTS_MAX_RETRIES)
def clientReserveSlotsRetryWait: Long = get(CLIENT_RESERVE_SLOTS_RETRY_WAIT)
+ def clientRequestCommitFilesRetryWait: Long = get(CLIENT_COMMIT_FILE_REQUEST_RETRY_WAIT)
def clientRequestCommitFilesMaxRetries: Int = get(CLIENT_COMMIT_FILE_REQUEST_MAX_RETRY)
def clientCommitFilesIgnoreExcludedWorkers: Boolean = get(CLIENT_COMMIT_IGNORE_EXCLUDED_WORKERS)
def clientShuffleDynamicResourceEnabled: Boolean =
@@ -5309,6 +5310,14 @@ object CelebornConf extends Logging {
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("3s")
+ val CLIENT_COMMIT_FILE_REQUEST_RETRY_WAIT: ConfigEntry[Long] =
+ buildConf("celeborn.client.requestCommitFiles.retryWait")
+ .categories("client")
+ .doc("Wait time before next retry if requestCommitFiles RPC failed.")
+ .version("0.6.3")
+ .timeConf(TimeUnit.MILLISECONDS)
+ .createWithDefaultString("10s")
+
val CLIENT_COMMIT_FILE_REQUEST_MAX_RETRY: ConfigEntry[Int] =
buildConf("celeborn.client.requestCommitFiles.maxRetries")
.categories("client")
diff --git a/docs/configuration/client.md b/docs/configuration/client.md
index fcd28ec2c..ece9503bd 100644
--- a/docs/configuration/client.md
+++ b/docs/configuration/client.md
@@ -84,6 +84,7 @@ license: |
| celeborn.client.registerShuffle.maxRetries | 3 | false | Max retry times for client to register shuffle. | 0.3.0 | celeborn.shuffle.register.maxRetries |
| celeborn.client.registerShuffle.retryWait | 3s | false | Wait time before next retry if register shuffle failed. | 0.3.0 | celeborn.shuffle.register.retryWait |
| celeborn.client.requestCommitFiles.maxRetries | 4 | false | Max retry times for requestCommitFiles RPC. | 0.3.0 | |
+| celeborn.client.requestCommitFiles.retryWait | 10s | false | Wait time before next retry if requestCommitFiles RPC failed. | 0.6.3 | |
| celeborn.client.reserveSlots.maxRetries | 3 | false | Max retry times for client to reserve slots. | 0.3.0 | celeborn.slots.reserve.maxRetries |
| celeborn.client.reserveSlots.rackaware.enabled | false | false | Whether need to place different replicates on different racks when allocating slots. | 0.3.1 | celeborn.client.reserveSlots.rackware.enabled |
| celeborn.client.reserveSlots.retryWait | 3s | false | Wait time before next retry if reserve slots failed. | 0.3.0 | celeborn.slots.reserve.retryWait |