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/01 10:40:01 UTC
[incubator-celeborn] branch main updated: [CELEBORN-86][REFATCOR] Register shuffle should have separated timeout configuration (#1031)
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 13e1e240 [CELEBORN-86][REFATCOR] Register shuffle should have separated timeout configuration (#1031)
13e1e240 is described below
commit 13e1e240350690792e85814b5de48b21c2b70443
Author: nafiy <30...@users.noreply.github.com>
AuthorDate: Thu Dec 1 18:39:56 2022 +0800
[CELEBORN-86][REFATCOR] Register shuffle should have separated timeout configuration (#1031)
* [CELEBORN-86][REFATCOR] Register shuffle should have separated timeout configuration
---
.../java/org/apache/celeborn/client/ShuffleClientImpl.java | 2 ++
.../java/org/apache/celeborn/client/ShuffleClientSuiteJ.java | 7 +------
.../main/scala/org/apache/celeborn/common/CelebornConf.scala | 12 ++++++++++++
docs/configuration/network.md | 1 +
4 files changed, 16 insertions(+), 6 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 ad7d4832..2df00e91 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
@@ -273,6 +273,7 @@ public class ShuffleClientImpl extends ShuffleClient {
() ->
driverRssMetaService.askSync(
RegisterShuffle$.MODULE$.apply(appId, shuffleId, numMappers, numPartitions),
+ conf.registerShuffleRpcAskTimeout(),
ClassTag$.MODULE$.apply(PbRegisterShuffleResponse.class)));
}
@@ -294,6 +295,7 @@ public class ShuffleClientImpl extends ShuffleClient {
driverRssMetaService.askSync(
RegisterMapPartitionTask$.MODULE$.apply(
appId, shuffleId, numMappers, mapId, attemptId, partitionId),
+ conf.registerShuffleRpcAskTimeout(),
ClassTag$.MODULE$.apply(PbRegisterShuffleResponse.class)));
return partitionLocationMap.get(partitionId);
}
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 e2be945d..bb5f5aa4 100644
--- a/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java
+++ b/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java
@@ -25,8 +25,6 @@ import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
-import scala.reflect.ClassTag$;
-
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.util.concurrent.Future;
@@ -41,7 +39,6 @@ import org.apache.celeborn.common.network.client.TransportClient;
import org.apache.celeborn.common.network.client.TransportClientFactory;
import org.apache.celeborn.common.protocol.CompressionCodec;
import org.apache.celeborn.common.protocol.PartitionLocation;
-import org.apache.celeborn.common.protocol.PbRegisterShuffleResponse;
import org.apache.celeborn.common.protocol.message.ControlMessages.*;
import org.apache.celeborn.common.protocol.message.StatusCode;
import org.apache.celeborn.common.rpc.RpcEndpointRef;
@@ -182,9 +179,7 @@ public class ShuffleClientSuiteJ {
shuffleClient = new ShuffleClientImpl(conf, new UserIdentifier("mock", "mock"));
masterLocation.setPeer(slaveLocation);
- when(endpointRef.askSync(
- RegisterShuffle$.MODULE$.apply(TEST_APPLICATION_ID, TEST_SHUFFLE_ID, 1, 1),
- ClassTag$.MODULE$.apply(PbRegisterShuffleResponse.class)))
+ when(endpointRef.askSync(any(), any(), any()))
.thenAnswer(
t ->
RegisterShuffleResponse$.MODULE$.apply(
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 83364634..d3c1a814 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -380,6 +380,10 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
new RpcTimeout(get(RPC_ASK_TIMEOUT).milli, RPC_ASK_TIMEOUT.key)
def haClientRpcAskTimeout: RpcTimeout =
new RpcTimeout(get(HA_CLIENT_RPC_ASK_TIMEOUT).milli, HA_CLIENT_RPC_ASK_TIMEOUT.key)
+ def registerShuffleRpcAskTimeout: RpcTimeout =
+ new RpcTimeout(
+ get(REGISTER_SHUFFLE_RPC_ASK_TIMEOUT).milli,
+ REGISTER_SHUFFLE_RPC_ASK_TIMEOUT.key)
def networkIoMode(module: String): String = {
val key = NETWORK_IO_MODE.key.replace("<module>", module)
@@ -1020,6 +1024,14 @@ object CelebornConf extends Logging {
.doc("Timeout for HA client RPC ask operations.")
.fallbackConf(NETWORK_TIMEOUT)
+ val REGISTER_SHUFFLE_RPC_ASK_TIMEOUT: ConfigEntry[Long] =
+ buildConf("celeborn.rpc.registerShuffle.askTimeout")
+ .categories("network")
+ .version("0.2.0")
+ .doc("Timeout for ask operations during register shuffle.")
+ .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 d6b01c62..83688e17 100644
--- a/docs/configuration/network.md
+++ b/docs/configuration/network.md
@@ -41,5 +41,6 @@ license: |
| celeborn.rpc.connect.threads | 64 | | 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 |
| celeborn.shuffle.maxChunksBeingTransferred | 9223372036854775807 | The max number of chunks allowed to be transferred at the same time on shuffle service. Note that new incoming connections will be closed when the max number is hit. The client will retry according to the shuffle retry configs (see `celeborn.shuffle.io.maxRetries` and `celeborn.shuffle.io.retryWait`), if those limits are reached the task will fail with fetch failure. | 0.2.0 |
<!--end-include-->