This is an automated email from the ASF dual-hosted git repository.
ethanfeng 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 61065230 [CELEBORN-311] not retry when register for map partition
occurs exception (#1246)
61065230 is described below
commit 61065230bd3e1bfa42b81b54282db83e4a30a7d4
Author: Shuang <[email protected]>
AuthorDate: Tue Feb 21 10:16:10 2023 +0800
[CELEBORN-311] not retry when register for map partition occurs exception
(#1246)
---
.../plugin/flink/RemoteShuffleOutputGate.java | 2 +-
.../org/apache/celeborn/client/ShuffleClient.java | 2 +-
.../apache/celeborn/client/ShuffleClientImpl.java | 8 ++++++-
.../apache/celeborn/client/LifecycleManager.scala | 25 ++++++++++++++--------
.../celeborn/client/WithShuffleClientSuite.scala | 4 ++--
.../celeborn/tests/client/ShuffleClientSuite.scala | 24 ++++++++++++++++++++-
6 files changed, 50 insertions(+), 15 deletions(-)
diff --git
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleOutputGate.java
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleOutputGate.java
index 3d7adef3..52a9a249 100644
---
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleOutputGate.java
+++
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleOutputGate.java
@@ -233,7 +233,7 @@ public class RemoteShuffleOutputGate {
}
}
- public void handshake(boolean isFirstHandShake) {
+ public void handshake(boolean isFirstHandShake) throws IOException {
if (partitionLocation == null) {
partitionLocation =
shuffleWriteClient.registerMapPartitionTask(
diff --git a/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java
b/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java
index 4d3d65c6..3d34fcf2 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java
@@ -235,7 +235,7 @@ public abstract class ShuffleClient {
throws IOException;
public abstract PartitionLocation registerMapPartitionTask(
- String appId, int shuffleId, int numMappers, int mapId, int attemptId);
+ String appId, int shuffleId, int numMappers, int mapId, int attemptId)
throws IOException;
public abstract ConcurrentHashMap<Integer, PartitionLocation>
getPartitionLocation(
String applicationId, int shuffleId, int numMappers, int numPartitions);
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 c44f3492..238f19c1 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
@@ -320,7 +320,7 @@ public class ShuffleClientImpl extends ShuffleClient {
@VisibleForTesting
public PartitionLocation registerMapPartitionTask(
- String appId, int shuffleId, int numMappers, int mapId, int attemptId) {
+ String appId, int shuffleId, int numMappers, int mapId, int attemptId)
throws IOException {
int partitionId = PackedPartitionId.packedPartitionId(mapId, attemptId);
logger.info(
"register mapPartitionTask, mapId: {}, attemptId: {}, partitionId: {}",
@@ -338,6 +338,12 @@ public class ShuffleClientImpl extends ShuffleClient {
appId, shuffleId, numMappers, mapId, attemptId,
partitionId),
conf.registerShuffleRpcAskTimeout(),
ClassTag$.MODULE$.apply(PbRegisterShuffleResponse.class)));
+
+ if (partitionLocationMap == null) {
+ String shuffleKey = Utils.makeShuffleKey(appId, shuffleId);
+ throw new IOException("Register shuffle failed for shuffle " +
shuffleKey);
+ }
+
return partitionLocationMap.get(partitionId);
}
diff --git
a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
index 77b06750..22ae9dc1 100644
--- a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
@@ -381,15 +381,22 @@ class LifecycleManager(appId: String, val conf:
CelebornConf) extends RpcEndpoin
.foreach(_.asScala.foreach(context => {
partitionType match {
case PartitionType.MAP =>
- val partitionLocations =
- response.getPartitionLocationsList.asScala.filter(_.getId ==
partitionId).map(r =>
- PbSerDeUtils.fromPbPartitionLocation(r)).toArray
- processMapTaskReply(
- applicationId,
- shuffleId,
- context.context,
- partitionId,
- partitionLocations)
+ if (response.getStatus == 0) {
+ val partitionLocations =
+ response.getPartitionLocationsList.asScala.filter(
+ _.getId == context.partitionId).map(r =>
+ PbSerDeUtils.fromPbPartitionLocation(r)).toArray
+ processMapTaskReply(
+ applicationId,
+ shuffleId,
+ context.context,
+ partitionId,
+ partitionLocations)
+ } else {
+ // when register not success, need reply origin response,
+ // otherwise will lost original exception message
+ context.reply(response)
+ }
case PartitionType.REDUCE => context.reply(response)
}
}))
diff --git
a/client/src/test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala
b/client/src/test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala
index a8dcb4e7..70e80275 100644
---
a/client/src/test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala
+++
b/client/src/test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala
@@ -44,7 +44,7 @@ trait WithShuffleClientSuite extends CelebornFunSuite {
private val mapId = 1
private val attemptId = 0
- test(s"test register map partition task") {
+ test("test register map partition task") {
Assert.assertNotNull(lifecycleManager)
Assert.assertNotNull(shuffleClient)
val shuffleId = 1
@@ -82,7 +82,7 @@ trait WithShuffleClientSuite extends CelebornFunSuite {
Assert.assertEquals(count, numMappers + 1)
}
- test(s"test map end & get reducer file group") {
+ test("test map end & get reducer file group") {
val shuffleId = 2
shuffleClient.registerMapPartitionTask(APP, shuffleId, numMappers, mapId,
attemptId)
shuffleClient.registerMapPartitionTask(APP, shuffleId, numMappers, mapId +
1, attemptId)
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ShuffleClientSuite.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ShuffleClientSuite.scala
index ce50fdcc..efc74b42 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ShuffleClientSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ShuffleClientSuite.scala
@@ -17,7 +17,10 @@
package org.apache.celeborn.tests.client
-import org.apache.celeborn.client.WithShuffleClientSuite
+import java.io.IOException
+
+import org.apache.celeborn.client.{LifecycleManager, ShuffleClientImpl,
WithShuffleClientSuite}
+import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.service.deploy.MiniClusterFeature
class ShuffleClientSuite extends WithShuffleClientSuite with
MiniClusterFeature {
@@ -37,6 +40,25 @@ class ShuffleClientSuite extends WithShuffleClientSuite with
MiniClusterFeature
setUpMiniCluster(masterConf, workerConf)
}
+ test("test register when master not available") {
+ val celebornConf: CelebornConf = new CelebornConf()
+ celebornConf.set("celeborn.master.endpoints", "localhost:19098")
+ celebornConf.set("celeborn.client.maxRetries", "0")
+
+ val lifecycleManager: LifecycleManager = new LifecycleManager(APP,
celebornConf)
+ val shuffleClient: ShuffleClientImpl = {
+ val client = new ShuffleClientImpl(celebornConf, userIdentifier)
+ client.setupMetaServiceRef(lifecycleManager.self)
+ client
+ }
+
+ assertThrows[IOException] {
+ () -> shuffleClient.registerMapPartitionTask(APP, 1, 1, 0, 0)
+ }
+
+ lifecycleManager.stop()
+ }
+
override def afterAll(): Unit = {
logInfo("all test complete , stop rss mini cluster")
shutdownMiniCluster()
