apache/celeborn
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[CELEBORN-155] Wrong TimeUnit for registerShuffleRetryWait in Shuffle… (#1099)

Browse Source
This commit is contained in:
Keyong Zhou 2022-12-19 17:32:18 +08:00 committed by GitHub
parent 13769f0f0a
commit a2dd72f20c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 4 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
View File

@ -82,7 +82,7 @@ public class ShuffleClientImpl extends ShuffleClient {
private final UserIdentifier userIdentifier;
private final int registerShuffleMaxRetries;
private final long registerShuffleRetryWait;
private final long registerShuffleRetryWaitMs;
private int maxInFlight;
private Integer currentMaxReqsInFlight = 1;
private int congestionAvoidanceFlag = 0;
@ -136,7 +136,7 @@ public class ShuffleClientImpl extends ShuffleClient {
this.conf = conf;
this.userIdentifier = userIdentifier;
registerShuffleMaxRetries = conf.registerShuffleMaxRetry();
registerShuffleRetryWait = conf.registerShuffleRetryWait();
registerShuffleRetryWaitMs = conf.registerShuffleRetryWaitMs();
maxInFlight = conf.pushMaxReqsInFlight();
pushBufferMaxSize = conf.pushBufferMaxSize();
@ -345,7 +345,7 @@ public class ShuffleClientImpl extends ShuffleClient {
}
try {
TimeUnit.SECONDS.sleep(registerShuffleRetryWait);
TimeUnit.MILLISECONDS.sleep(registerShuffleRetryWaitMs);
} catch (InterruptedException e) {
break;
}

2
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
View File

@ -520,7 +520,7 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
def shuffleManagerPort: Int = get(SHUFFLE_MANAGER_PORT)
def shuffleChunkSize: Long = get(SHUFFLE_CHUCK_SIZE)
def registerShuffleMaxRetry: Int = get(SHUFFLE_REGISTER_MAX_RETRIES)
def registerShuffleRetryWait: Long = get(SHUFFLE_REGISTER_RETRY_WAIT)
def registerShuffleRetryWaitMs: Long = get(SHUFFLE_REGISTER_RETRY_WAIT)
def reserveSlotsMaxRetries: Int = get(RESERVE_SLOTS_MAX_RETRIES)
def reserveSlotsRetryWait: Long = get(RESERVE_SLOTS_RETRY_WAIT)
def rpcMaxParallelism: Int = get(CLIENT_RPC_MAX_PARALLELISM)