This is an automated email from the ASF dual-hosted git repository.
zhouky 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 eda21ead2 [CELEBORN-344] Change PUSH_DATA_FAIL_MASTER/SALVE to
PUSH_DATA_WRITE_FAIL_MASTER/SALVE (#1281)
eda21ead2 is described below
commit eda21ead2486c587dba7dc63c26b4df24e8fc866
Author: Angerszhuuuu <[email protected]>
AuthorDate: Tue Feb 28 11:29:40 2023 +0800
[CELEBORN-344] Change PUSH_DATA_FAIL_MASTER/SALVE to
PUSH_DATA_WRITE_FAIL_MASTER/SALVE (#1281)
---
.../org/apache/celeborn/client/ShuffleClientImpl.java | 10 ++++++----
.../org/apache/celeborn/client/LifecycleManager.scala | 8 ++++----
.../celeborn/common/protocol/message/StatusCode.java | 6 +++---
.../scala/org/apache/celeborn/common/util/Utils.scala | 6 +++---
.../service/deploy/worker/PushDataHandler.scala | 18 +++++++++---------
5 files changed, 25 insertions(+), 23 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 f0c6c9476..fc278bda2 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
@@ -1465,10 +1465,10 @@ public class ShuffleClientImpl extends ShuffleClient {
private StatusCode getPushDataFailCause(String message) {
logger.debug("Push data failed cause message: " + message);
StatusCode cause;
- if (message.startsWith(StatusCode.PUSH_DATA_FAIL_SLAVE.name())) {
- cause = StatusCode.PUSH_DATA_FAIL_SLAVE;
- } else if (message.startsWith(StatusCode.PUSH_DATA_FAIL_MASTER.name())) {
- cause = StatusCode.PUSH_DATA_FAIL_MASTER;
+ if (message.startsWith(StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE.name())) {
+ cause = StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE;
+ } else if
(message.startsWith(StatusCode.PUSH_DATA_WRITE_FAIL_MASTER.name())) {
+ cause = StatusCode.PUSH_DATA_WRITE_FAIL_MASTER;
} else if
(message.startsWith(StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER.name()))
{
cause = StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER;
} else if
(message.startsWith(StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE.name())) {
@@ -1481,6 +1481,8 @@ public class ShuffleClientImpl extends ShuffleClient {
cause = StatusCode.PUSH_DATA_TIMEOUT_MASTER;
} else if (message.startsWith(StatusCode.PUSH_DATA_TIMEOUT_SLAVE.name())) {
cause = StatusCode.PUSH_DATA_TIMEOUT_SLAVE;
+ } else if (message.startsWith(StatusCode.REPLICATE_DATA_FAILED.name())) {
+ cause = StatusCode.REPLICATE_DATA_FAILED;
} else if (connectFail(message)) {
// Throw when push to master worker connection causeException.
cause = StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_MASTER;
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 6754d1278..df08f06c0 100644
--- a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
@@ -533,11 +533,11 @@ class LifecycleManager(appId: String, val conf:
CelebornConf) extends RpcEndpoin
if (oldPartition != null) {
cause match {
- case StatusCode.PUSH_DATA_FAIL_MASTER =>
- blacklistPartitionWorker(oldPartition,
StatusCode.PUSH_DATA_FAIL_MASTER)
- case StatusCode.PUSH_DATA_FAIL_SLAVE
+ case StatusCode.PUSH_DATA_WRITE_FAIL_MASTER =>
+ blacklistPartitionWorker(oldPartition,
StatusCode.PUSH_DATA_WRITE_FAIL_MASTER)
+ case StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE
if oldPartition.getPeer != null && conf.blacklistSlaveEnabled =>
- blacklistPartitionWorker(oldPartition.getPeer,
StatusCode.PUSH_DATA_FAIL_SLAVE)
+ blacklistPartitionWorker(oldPartition.getPeer,
StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE)
case StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER =>
blacklistPartitionWorker(oldPartition,
StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER)
case StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE
diff --git
a/common/src/main/java/org/apache/celeborn/common/protocol/message/StatusCode.java
b/common/src/main/java/org/apache/celeborn/common/protocol/message/StatusCode.java
index a0ec42de7..407db99f3 100644
---
a/common/src/main/java/org/apache/celeborn/common/protocol/message/StatusCode.java
+++
b/common/src/main/java/org/apache/celeborn/common/protocol/message/StatusCode.java
@@ -34,15 +34,15 @@ public enum StatusCode {
DELETE_FILES_FAILED(10),
PARTITION_EXISTS(11),
REVIVE_FAILED(12),
- PUSH_DATA_FAILED(13),
+ REPLICATE_DATA_FAILED(13),
NUM_MAPPER_ZERO(14),
MAP_ENDED(15),
STAGE_ENDED(16),
// push data fail causes
PUSH_DATA_FAIL_NON_CRITICAL_CAUSE(17),
- PUSH_DATA_FAIL_SLAVE(18),
- PUSH_DATA_FAIL_MASTER(19),
+ PUSH_DATA_WRITE_FAIL_SLAVE(18),
+ PUSH_DATA_WRITE_FAIL_MASTER(19),
PUSH_DATA_FAIL_PARTITION_NOT_FOUND(20),
HARD_SPLIT(21),
diff --git a/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
b/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
index 5e697e225..e4abc6bd8 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
@@ -862,7 +862,7 @@ object Utils extends Logging {
case 12 =>
StatusCode.REVIVE_FAILED
case 13 =>
- StatusCode.PUSH_DATA_FAILED
+ StatusCode.REPLICATE_DATA_FAILED
case 14 =>
StatusCode.NUM_MAPPER_ZERO
case 15 =>
@@ -872,9 +872,9 @@ object Utils extends Logging {
case 17 =>
StatusCode.PUSH_DATA_FAIL_NON_CRITICAL_CAUSE
case 18 =>
- StatusCode.PUSH_DATA_FAIL_SLAVE
+ StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE
case 19 =>
- StatusCode.PUSH_DATA_FAIL_MASTER
+ StatusCode.PUSH_DATA_WRITE_FAIL_MASTER
case 20 =>
StatusCode.PUSH_DATA_FAIL_PARTITION_NOT_FOUND
case 21 =>
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
index f2f37da23..273cce6e9 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
@@ -214,9 +214,9 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
if (exception != null) {
val cause =
if (isMaster) {
- StatusCode.PUSH_DATA_FAIL_MASTER
+ StatusCode.PUSH_DATA_WRITE_FAIL_MASTER
} else {
- StatusCode.PUSH_DATA_FAIL_SLAVE
+ StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE
}
logError(
s"While handling PushData, throw $cause, fileWriter $fileWriter has
exception.",
@@ -305,7 +305,7 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
// 1. Throw PUSH_DATA_FAIL_SLAVE by slave peer worker
// 2. Throw PUSH_DATA_TIMEOUT_SLAVE by TransportResponseHandler
// 3. Throw IOException by channel, convert to
PUSH_DATA_CONNECTION_EXCEPTION_SLAVE
- if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_FAIL_SLAVE.name()) ||
+ if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE.name()) ||
e.getMessage.startsWith(StatusCode.PUSH_DATA_TIMEOUT_SLAVE.name())) {
callbackWithTimer.onFailure(e)
} else {
@@ -488,9 +488,9 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
val exception = fileWriterWithException.get.getException
val cause =
if (isMaster) {
- StatusCode.PUSH_DATA_FAIL_MASTER
+ StatusCode.PUSH_DATA_WRITE_FAIL_MASTER
} else {
- StatusCode.PUSH_DATA_FAIL_SLAVE
+ StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE
}
logError(
s"While handling PushMergedData, throw $cause, fileWriter
$fileWriterWithException has exception.",
@@ -557,7 +557,7 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
// 1. Throw PUSH_DATA_FAIL_SLAVE by slave peer worker
// 2. Throw PUSH_DATA_TIMEOUT_SLAVE by TransportResponseHandler
// 3. Throw IOException by channel, convert to
PUSH_DATA_CONNECTION_EXCEPTION_SLAVE
- if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_FAIL_SLAVE.name()) ||
+ if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE.name()) ||
e.getMessage.startsWith(StatusCode.PUSH_DATA_TIMEOUT_SLAVE.name())) {
callbackWithTimer.onFailure(e)
} else {
@@ -961,7 +961,7 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
if (e.isInstanceOf[CelebornIOException]) {
callback.onFailure(e)
} else {
- callback.onFailure(new
CelebornIOException(StatusCode.PUSH_DATA_FAIL_SLAVE, e))
+ callback.onFailure(new
CelebornIOException(StatusCode.REPLICATE_DATA_FAILED, e))
}
}
}
@@ -1010,8 +1010,8 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
messageType match {
case Type.PUSH_DATA | Type.PUSH_DATA_HAND_SHAKE =>
(
- StatusCode.PUSH_DATA_FAIL_MASTER,
- StatusCode.PUSH_DATA_FAIL_SLAVE)
+ StatusCode.PUSH_DATA_WRITE_FAIL_MASTER,
+ StatusCode.PUSH_DATA_WRITE_FAIL_SLAVE)
case Type.PUSH_DATA_HAND_SHAKE => (
StatusCode.PUSH_DATA_HANDSHAKE_FAIL_MASTER,
StatusCode.PUSH_DATA_HANDSHAKE_FAIL_SLAVE)
