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 2024/05/06 09:48:17 UTC
(celeborn) branch branch-0.4 updated: [CELEBORN-1409] CommitHandler commitFiles RPC supports separate timeout configuration
This is an automated email from the ASF dual-hosted git repository.
nicholasjiang pushed a commit to branch branch-0.4
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-0.4 by this push:
new 3d9152a30 [CELEBORN-1409] CommitHandler commitFiles RPC supports separate timeout configuration
3d9152a30 is described below
commit 3d9152a3030f6427d7c43fbc4b2fdd3c1ca8714b
Author: sychen <sy...@ctrip.com>
AuthorDate: Mon May 6 17:42:52 2024 +0800
[CELEBORN-1409] CommitHandler commitFiles RPC supports separate timeout configuration
This PR aims to supports separate timeout configuration at CommitHandler commitFiles RPC.
The default value of `celeborn.worker.commitFiles.timeout` is 120s, and the default value of Client's RPC is 60s.
No
GA
Closes #2488 from cxzl25/CELEBORN-1409.
Authored-by: sychen <sy...@ctrip.com>
Signed-off-by: SteNicholas <pr...@163.com>
(cherry picked from commit dc5219216331be1ab0b949f681542452a9c465b1)
Signed-off-by: SteNicholas <pr...@163.com>
---
.../org/apache/celeborn/client/commit/CommitHandler.scala | 3 ++-
.../main/scala/org/apache/celeborn/common/CelebornConf.scala | 12 ++++++++++++
2 files changed, 14 insertions(+), 1 deletion(-)
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 af4bd99f4..8532599bd 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
@@ -68,6 +68,7 @@ abstract class CommitHandler(
val sharedRpcPool: ThreadPoolExecutor) extends Logging {
private val pushReplicateEnabled = conf.clientPushReplicateEnabled
+ private val clientRpcCommitFilesAskTimeout = conf.clientRpcCommitFilesAskTimeout
private val commitEpoch = new AtomicLong()
private val totalWritten = new LongAdder
@@ -452,7 +453,7 @@ abstract class CommitHandler(
message.replicaIds)
}(ec)
} else {
- worker.endpoint.ask[CommitFilesResponse](message)
+ worker.endpoint.ask[CommitFilesResponse](message, clientRpcCommitFilesAskTimeout)
}
}
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 261e97478..6ff18b5a6 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -797,6 +797,11 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
get(CLIENT_RPC_GET_REDUCER_FILE_GROUP_RPC_ASK_TIMEOUT).milli,
CLIENT_RPC_GET_REDUCER_FILE_GROUP_RPC_ASK_TIMEOUT.key)
+ def clientRpcCommitFilesAskTimeout: RpcTimeout =
+ new RpcTimeout(
+ get(CLIENT_RPC_COMMIT_FILES_ASK_TIMEOUT).milli,
+ CLIENT_RPC_COMMIT_FILES_ASK_TIMEOUT.key)
+
// //////////////////////////////////////////////////////
// Shuffle Client Fetch //
// //////////////////////////////////////////////////////
@@ -3795,6 +3800,13 @@ object CelebornConf extends Logging {
s"By default, the value is the max timeout value `${NETWORK_IO_CONNECTION_TIMEOUT.key}`.")
.fallbackConf(NETWORK_IO_CONNECTION_TIMEOUT)
+ val CLIENT_RPC_COMMIT_FILES_ASK_TIMEOUT: ConfigEntry[Long] =
+ buildConf("celeborn.client.rpc.commitFiles.askTimeout")
+ .categories("client")
+ .version("0.4.1")
+ .doc("Timeout for CommitHandler commit files.")
+ .fallbackConf(RPC_ASK_TIMEOUT)
+
val CLIENT_RPC_CACHE_SIZE: ConfigEntry[Int] =
buildConf("celeborn.client.rpc.cache.size")
.withAlternative("celeborn.rpc.cache.size")