You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by an...@apache.org on 2022/12/02 14:53:54 UTC
[incubator-celeborn] branch main updated: [CELEBORN-90][REFACTOR] GetReducerFileGroup should support separated timeout configuration (#1045)
This is an automated email from the ASF dual-hosted git repository.
angerszhuuuu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 44d45c2a [CELEBORN-90][REFACTOR] GetReducerFileGroup should support separated timeout configuration (#1045)
44d45c2a is described below
commit 44d45c2a27d2ab74469665ab44e096d9dfc4ed41
Author: nafiy <30...@users.noreply.github.com>
AuthorDate: Fri Dec 2 22:53:51 2022 +0800
[CELEBORN-90][REFACTOR] GetReducerFileGroup should support separated timeout configuration (#1045)
---
.../java/org/apache/celeborn/client/ShuffleClientImpl.java | 8 ++++----
.../main/scala/org/apache/celeborn/common/CelebornConf.scala | 12 ++++++++++++
docs/configuration/network.md | 1 +
3 files changed, 17 insertions(+), 4 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 2df00e91..aefb2966 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
@@ -26,7 +26,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import scala.reflect.ClassTag;
import scala.reflect.ClassTag$;
import com.google.common.annotations.VisibleForTesting;
@@ -1109,11 +1108,12 @@ public class ShuffleClientImpl extends ShuffleClient {
GetReducerFileGroup getReducerFileGroup =
new GetReducerFileGroup(applicationId, shuffleId);
- ClassTag<GetReducerFileGroupResponse> classTag =
- ClassTag$.MODULE$.apply(GetReducerFileGroupResponse.class);
GetReducerFileGroupResponse response =
- driverRssMetaService.askSync(getReducerFileGroup, classTag);
+ driverRssMetaService.askSync(
+ getReducerFileGroup,
+ conf.getReducerFileGroupRpcAskTimeout(),
+ ClassTag$.MODULE$.apply(GetReducerFileGroupResponse.class));
if (response.status() == StatusCode.SUCCESS) {
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 d3c1a814..4da854a2 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -384,6 +384,10 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
new RpcTimeout(
get(REGISTER_SHUFFLE_RPC_ASK_TIMEOUT).milli,
REGISTER_SHUFFLE_RPC_ASK_TIMEOUT.key)
+ def getReducerFileGroupRpcAskTimeout: RpcTimeout =
+ new RpcTimeout(
+ get(GET_REDUCER_FILE_GROUP_RPC_ASK_TIMEOUT).milli,
+ GET_REDUCER_FILE_GROUP_RPC_ASK_TIMEOUT.key)
def networkIoMode(module: String): String = {
val key = NETWORK_IO_MODE.key.replace("<module>", module)
@@ -1032,6 +1036,14 @@ object CelebornConf extends Logging {
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("600s")
+ val GET_REDUCER_FILE_GROUP_RPC_ASK_TIMEOUT: ConfigEntry[Long] =
+ buildConf("celeborn.rpc.getReducerFileGroup.askTimeout")
+ .categories("network")
+ .version("0.2.0")
+ .doc("Timeout for ask operations during get reducer file group.")
+ .timeConf(TimeUnit.MILLISECONDS)
+ .createWithDefaultString("600s")
+
val NETWORK_IO_MODE: ConfigEntry[String] =
buildConf("celeborn.<module>.io.mode")
.categories("network")
diff --git a/docs/configuration/network.md b/docs/configuration/network.md
index 83688e17..569429c6 100644
--- a/docs/configuration/network.md
+++ b/docs/configuration/network.md
@@ -39,6 +39,7 @@ license: |
| celeborn.port.maxRetries | 1 | When port is occupied, we will retry for max retry times. | 0.2.0 |
| celeborn.rpc.askTimeout | <value of celeborn.network.timeout> | Timeout for RPC ask operations. | 0.2.0 |
| celeborn.rpc.connect.threads | 64 | | 0.2.0 |
+| celeborn.rpc.getReducerFileGroup.askTimeout | 600s | Timeout for ask operations during get reducer file group. | 0.2.0 |
| celeborn.rpc.haClient.askTimeout | <value of celeborn.network.timeout> | Timeout for HA client RPC ask operations. | 0.2.0 |
| celeborn.rpc.lookupTimeout | 30s | Timeout for RPC lookup operations. | 0.2.0 |
| celeborn.rpc.registerShuffle.askTimeout | 600s | Timeout for ask operations during register shuffle. | 0.2.0 |