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 a7587c3fe [CELEBORN-337] Remove unnecessary StatusCode.message (#1272)
a7587c3fe is described below
commit a7587c3fe7dc1f9cbb1182e5f6f684ef25c94b04
Author: Angerszhuuuu <[email protected]>
AuthorDate: Fri Feb 24 15:11:07 2023 +0800
[CELEBORN-337] Remove unnecessary StatusCode.message (#1272)
* [CELEBORN-337] Remove unnecessary StatusCode.message
---
.../apache/celeborn/client/ShuffleClientImpl.java | 29 ++++---
.../network/client/TransportResponseHandler.java | 7 +-
.../common/protocol/message/StatusCode.java | 46 -----------
.../common/exception/CelebornIOException.scala | 6 ++
.../service/deploy/worker/PushDataHandler.scala | 92 +++++++++++-----------
5 files changed, 69 insertions(+), 111 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 d485e8c49..6c5a1ad01 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
@@ -230,7 +230,7 @@ public class ShuffleClientImpl extends ShuffleClient {
e);
wrappedCallback.onFailure(
new Exception(
- StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER.getMessage()
+ StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER
+ "! "
+ e.getMessage()
+ ". "
@@ -799,7 +799,7 @@ public class ShuffleClientImpl extends ShuffleClient {
if (remainReviveTimes <= 0) {
callback.onFailure(
new Exception(
- cause.getMessage()
+ cause
+ "! Push data to master worker of "
+ loc
+ " failed: "
@@ -873,7 +873,7 @@ public class ShuffleClientImpl extends ShuffleClient {
e);
wrappedCallback.onFailure(
new Exception(
- StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER.getMessage()
+ StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER
+ "! "
+ e.getMessage()
+ ". "
@@ -1189,7 +1189,7 @@ public class ShuffleClientImpl extends ShuffleClient {
if (remainReviveTimes <= 0) {
callback.onFailure(
new Exception(
- cause.getMessage()
+ cause
+ "! Push data to master worker of "
+ hostPort
+ " failed: "
@@ -1248,7 +1248,7 @@ public class ShuffleClientImpl extends ShuffleClient {
e);
wrappedCallback.onFailure(
new Exception(
- StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER.getMessage()
+ StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER
+ "! "
+ e.getMessage()
+ ". "
@@ -1481,22 +1481,21 @@ 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.getMessage())) {
+ if (message.startsWith(StatusCode.PUSH_DATA_FAIL_SLAVE.name())) {
cause = StatusCode.PUSH_DATA_FAIL_SLAVE;
- } else if
(message.startsWith(StatusCode.PUSH_DATA_FAIL_MASTER.getMessage())) {
+ } else if (message.startsWith(StatusCode.PUSH_DATA_FAIL_MASTER.name())) {
cause = StatusCode.PUSH_DATA_FAIL_MASTER;
- } else if (message.startsWith(
- StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER.getMessage())) {
+ } 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.getMessage()))
{
+ } else if
(message.startsWith(StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE.name())) {
cause = StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE;
- } else if
(message.startsWith(StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_MASTER.getMessage()))
{
+ } else if
(message.startsWith(StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_MASTER.name())) {
cause = StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_MASTER;
- } else if
(message.startsWith(StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE.getMessage()))
{
+ } else if
(message.startsWith(StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE.name())) {
cause = StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE;
- } else if
(message.startsWith(StatusCode.PUSH_DATA_TIMEOUT_MASTER.getMessage())) {
+ } else if (message.startsWith(StatusCode.PUSH_DATA_TIMEOUT_MASTER.name()))
{
cause = StatusCode.PUSH_DATA_TIMEOUT_MASTER;
- } else if
(message.startsWith(StatusCode.PUSH_DATA_TIMEOUT_SLAVE.getMessage())) {
+ } else if (message.startsWith(StatusCode.PUSH_DATA_TIMEOUT_SLAVE.name())) {
cause = StatusCode.PUSH_DATA_TIMEOUT_SLAVE;
} else if (connectFail(message)) {
// Throw when push to master worker connection causeException.
@@ -1639,7 +1638,7 @@ public class ShuffleClientImpl extends ShuffleClient {
e);
callback.onFailure(
new Exception(
- StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER.getMessage()
+ StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER
+ "! "
+ e.getMessage()
+ ". "
diff --git
a/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
b/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
index 337b8487f..bebca636d 100644
---
a/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
+++
b/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
@@ -29,6 +29,7 @@ import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.celeborn.common.exception.CelebornIOException;
import org.apache.celeborn.common.network.protocol.*;
import org.apache.celeborn.common.network.server.MessageHandler;
import org.apache.celeborn.common.network.util.NettyUtils;
@@ -92,11 +93,9 @@ public class TransportResponseHandler extends
MessageHandler<ResponseMessage> {
// When module name equals to DATA_MODULE, mean shuffle client push
data, else means
// do data replication.
if
(TransportModuleConstants.DATA_MODULE.equals(conf.getModuleName())) {
- info.callback.onFailure(
- new
IOException(StatusCode.PUSH_DATA_TIMEOUT_MASTER.getMessage()));
+ info.callback.onFailure(new
CelebornIOException(StatusCode.PUSH_DATA_TIMEOUT_MASTER));
} else if
(TransportModuleConstants.PUSH_MODULE.equals(conf.getModuleName())) {
- info.callback.onFailure(
- new
IOException(StatusCode.PUSH_DATA_TIMEOUT_SLAVE.getMessage()));
+ info.callback.onFailure(new
CelebornIOException(StatusCode.PUSH_DATA_TIMEOUT_SLAVE));
}
info.channelFuture = null;
info.callback = null;
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 da5dd0062..a0ec42de7 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
@@ -85,50 +85,4 @@ public enum StatusCode {
public final byte getValue() {
return value;
}
-
- public String getMessage() {
- String msg = "";
- if (value == PUSH_DATA_FAIL_MASTER.getValue()) {
- msg = "PushDataFailMaster";
- } else if (value == PUSH_DATA_FAIL_SLAVE.getValue()) {
- msg = "PushDataFailSlave";
- } else if (value == PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER.getValue()) {
- msg = "PushDataCreateConnectionFailMaster";
- } else if (value == PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE.getValue()) {
- msg = "PushDataCreateConnectionFailSlave";
- } else if (value == PUSH_DATA_CONNECTION_EXCEPTION_MASTER.getValue()) {
- msg = "PushDataConnectionExceptionMaster";
- } else if (value == PUSH_DATA_CONNECTION_EXCEPTION_SLAVE.getValue()) {
- msg = "PushDataConnectionExceptionSlave";
- } else if (value == PUSH_DATA_FAIL_NON_CRITICAL_CAUSE.getValue()) {
- msg = "PushDataFailNonCriticalCause";
- } else if (value == PUSH_DATA_FAIL_PARTITION_NOT_FOUND.getValue()) {
- msg = "PushDataFailPartitionNotFound";
- } else if (value == HARD_SPLIT.getValue()) {
- msg = "PartitionFileSplit";
- } else if (value == PUSH_DATA_HANDSHAKE_FAIL_MASTER.getValue()) {
- msg = "PushDataHandShakeFailMaster";
- } else if (value == PUSH_DATA_HANDSHAKE_FAIL_SLAVE.getValue()) {
- msg = "PushDataHandShakeFailSlave";
- } else if (value == REGION_START_FAIL_MASTER.getValue()) {
- msg = "RegionStartFailMaster";
- } else if (value == REGION_START_FAIL_SLAVE.getValue()) {
- msg = "RegionStartFailSlave";
- } else if (value == REGION_FINISH_FAIL_MASTER.getValue()) {
- msg = "RegionFinishFailMaster";
- } else if (value == REGION_FINISH_FAIL_SLAVE.getValue()) {
- msg = "RegionFinishFailSlave";
- } else if (value == PUSH_DATA_TIMEOUT_MASTER.getValue()) {
- msg = "PushDataTimeoutMaster";
- } else if (value == PUSH_DATA_TIMEOUT_SLAVE.getValue()) {
- msg = "PushDataTimeoutSlave";
- }
-
- return msg;
- }
-
- @Override
- public String toString() {
- return "StatusCode{" + "value=" + this.name() + '}';
- }
}
diff --git
a/common/src/main/scala/org/apache/celeborn/common/exception/CelebornIOException.scala
b/common/src/main/scala/org/apache/celeborn/common/exception/CelebornIOException.scala
index 7f11d3ab7..06012f4f9 100644
---
a/common/src/main/scala/org/apache/celeborn/common/exception/CelebornIOException.scala
+++
b/common/src/main/scala/org/apache/celeborn/common/exception/CelebornIOException.scala
@@ -19,10 +19,16 @@ package org.apache.celeborn.common.exception
import java.io.IOException
+import org.apache.celeborn.common.protocol.message.StatusCode
+
class CelebornIOException(message: String, cause: Throwable)
extends IOException(message, cause) {
def this(message: String) = this(message, null)
def this(cause: Throwable) = this(cause.getMessage, cause)
+
+ def this(statusCode: StatusCode) = this(statusCode.name())
+
+ def this(statusCode: StatusCode, cause: Throwable) = this(statusCode.name(),
cause)
}
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 0eca2c4b2..7d269267f 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
@@ -25,7 +25,7 @@ import com.google.common.base.Throwables
import io.netty.buffer.ByteBuf
import org.apache.celeborn.common.CelebornConf
-import org.apache.celeborn.common.exception.AlreadyClosedException
+import org.apache.celeborn.common.exception.{AlreadyClosedException,
CelebornIOException}
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.meta.{WorkerInfo,
WorkerPartitionLocationInfo}
import org.apache.celeborn.common.metrics.source.{RPCSource, Source}
@@ -201,7 +201,7 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
s"attempt $attemptId, uniqueId ${pushData.partitionUniqueId})."
logWarning(s"[handlePushData] $msg")
callbackWithTimer.onFailure(
- new
Exception(StatusCode.PUSH_DATA_FAIL_PARTITION_NOT_FOUND.getMessage()))
+ new
CelebornIOException(StatusCode.PUSH_DATA_FAIL_PARTITION_NOT_FOUND))
}
}
return
@@ -221,11 +221,11 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
logWarning(s"[handlePushData] fileWriter $fileWriter has Exception
$exception")
val message =
if (isMaster) {
- StatusCode.PUSH_DATA_FAIL_MASTER.getMessage()
+ StatusCode.PUSH_DATA_FAIL_MASTER
} else {
- StatusCode.PUSH_DATA_FAIL_SLAVE.getMessage()
+ StatusCode.PUSH_DATA_FAIL_SLAVE
}
- callbackWithTimer.onFailure(new Exception(s"$message! $location",
exception))
+ callbackWithTimer.onFailure(new CelebornIOException(s"$message!
$location", exception))
return
}
val diskFull =
@@ -263,8 +263,8 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
pushData.body().release()
workerSource.incCounter(WorkerSource.PushDataFailCount)
callbackWithTimer.onFailure(
- new Exception(
-
s"${StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE.getMessage}! Peer
$peerWorker unavailable for $location!"))
+ new CelebornIOException(
+ s"${StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE}! Peer
$peerWorker unavailable for $location!"))
return
}
@@ -306,16 +306,16 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
logError(s"[handlePushData.onFailure] partitionLocation:
$location", e)
workerSource.incCounter(WorkerSource.PushDataFailCount)
// Throw by slave peer worker
- if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_FAIL_SLAVE.getMessage)) {
+ if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_FAIL_SLAVE.name())) {
callbackWithTimer.onFailure(e)
- } else if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_TIMEOUT_SLAVE.getMessage)) {
- callbackWithTimer.onFailure(new Exception(
- s"${StatusCode.PUSH_DATA_TIMEOUT_SLAVE.getMessage}! Push
data to peer of $location timeout: ${e.getMessage}",
+ } else if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_TIMEOUT_SLAVE.name())) {
+ callbackWithTimer.onFailure(new CelebornIOException(
+ s"${StatusCode.PUSH_DATA_TIMEOUT_SLAVE}! Push data to peer
of $location timeout: ${e.getMessage}",
e))
} else {
// Throw by connection
- callbackWithTimer.onFailure(new Exception(
-
s"${StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE.getMessage}! Push data to
peer of $location failed: ${e.getMessage}",
+ callbackWithTimer.onFailure(new CelebornIOException(
+ s"${StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE}! Push
data to peer of $location failed: ${e.getMessage}",
e))
}
}
@@ -335,8 +335,8 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
unavailablePeers.put(peerWorker, System.currentTimeMillis())
workerSource.incCounter(WorkerSource.PushDataFailCount)
callbackWithTimer.onFailure(
- new Exception(
-
s"${StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE.getMessage}! Create
connection to peer $peerWorker failed for $location",
+ new CelebornIOException(
+ s"${StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE}!
Create connection to peer $peerWorker failed for $location",
e))
}
}
@@ -477,7 +477,7 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
s" attempt $attemptId, uniqueId $id)."
logWarning(s"[handlePushMergedData] $msg")
callbackWithTimer.onFailure(
- new
Exception(StatusCode.PUSH_DATA_FAIL_PARTITION_NOT_FOUND.getMessage()))
+ new
CelebornIOException(StatusCode.PUSH_DATA_FAIL_PARTITION_NOT_FOUND))
}
}
return
@@ -500,11 +500,11 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
s" has Exception $exception")
val message =
if (isMaster) {
- StatusCode.PUSH_DATA_FAIL_MASTER.getMessage()
+ StatusCode.PUSH_DATA_FAIL_MASTER
} else {
- StatusCode.PUSH_DATA_FAIL_SLAVE.getMessage()
+ StatusCode.PUSH_DATA_FAIL_SLAVE
}
- callbackWithTimer.onFailure(new Exception(
+ callbackWithTimer.onFailure(new CelebornIOException(
s"$message! ${partitionIdToLocations.head._2}",
exception))
return
@@ -527,8 +527,8 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
if (unavailablePeers.containsKey(peerWorker)) {
pushMergedData.body().release()
workerSource.incCounter(WorkerSource.PushDataFailCount)
- callbackWithTimer.onFailure(new Exception(
-
s"${StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE.getMessage}! Peer
$peerWorker unavailable for $location!"))
+ callbackWithTimer.onFailure(new CelebornIOException(
+ s"${StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE}! Peer
$peerWorker unavailable for $location!"))
return
}
@@ -563,16 +563,16 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
override def onFailure(e: Throwable): Unit = {
workerSource.incCounter(WorkerSource.PushDataFailCount)
// Throw by slave peer worker
- if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_FAIL_SLAVE.getMessage)) {
+ if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_FAIL_SLAVE.name())) {
callbackWithTimer.onFailure(e)
- } else if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_TIMEOUT_SLAVE.getMessage)) {
- callbackWithTimer.onFailure(new Exception(
- s"${StatusCode.PUSH_DATA_TIMEOUT_SLAVE.getMessage}! Push
data to peer of ${partitionIdToLocations.head._2} timeout: ${e.getMessage}",
+ } else if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_TIMEOUT_SLAVE.name())) {
+ callbackWithTimer.onFailure(new CelebornIOException(
+ s"${StatusCode.PUSH_DATA_TIMEOUT_SLAVE}! Push data to peer
of ${partitionIdToLocations.head._2} timeout: ${e.getMessage}",
e))
} else {
// Throw by connection
- callbackWithTimer.onFailure(new Exception(
-
s"${StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE.getMessage}! Push data to
peer of ${partitionIdToLocations.head._2} failed: ${e.getMessage}",
+ callbackWithTimer.onFailure(new CelebornIOException(
+ s"${StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE}! Push
data to peer of ${partitionIdToLocations.head._2} failed: ${e.getMessage}",
e))
}
}
@@ -599,8 +599,8 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
unavailablePeers.put(peerWorker, System.currentTimeMillis())
workerSource.incCounter(WorkerSource.PushDataFailCount)
callbackWithTimer.onFailure(
- new Exception(
-
s"${StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE.getMessage}! Create
connection to peer $peerWorker failed for $location",
+ new CelebornIOException(
+ s"${StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE}!
Create connection to peer $peerWorker failed for $location",
e))
}
}
@@ -929,7 +929,7 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
}
} catch {
case t: Throwable =>
- callback.onFailure(new Exception(s"$messageType failed", t))
+ callback.onFailure(new CelebornIOException(s"$messageType failed", t))
}
}
@@ -967,18 +967,18 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
messageType match {
case Type.PUSH_DATA_HAND_SHAKE =>
workerSource.incCounter(WorkerSource.PushDataHandshakeFailCount)
- callback.onFailure(new Exception(
- StatusCode.PUSH_DATA_HANDSHAKE_FAIL_SLAVE.getMessage,
+ callback.onFailure(new CelebornIOException(
+ StatusCode.PUSH_DATA_HANDSHAKE_FAIL_SLAVE,
e))
case Type.REGION_START =>
workerSource.incCounter(WorkerSource.RegionStartFailCount)
- callback.onFailure(new
Exception(StatusCode.REGION_START_FAIL_SLAVE.getMessage, e))
+ callback.onFailure(new
CelebornIOException(StatusCode.REGION_START_FAIL_SLAVE, e))
case Type.REGION_FINISH =>
workerSource.incCounter(WorkerSource.RegionFinishFailCount)
- callback.onFailure(new
Exception(StatusCode.REGION_FINISH_FAIL_SLAVE.getMessage, e))
+ callback.onFailure(new
CelebornIOException(StatusCode.REGION_FINISH_FAIL_SLAVE, e))
case _ =>
workerSource.incCounter(WorkerSource.PushDataFailCount)
- callback.onFailure(new
Exception(StatusCode.PUSH_DATA_FAIL_SLAVE.getMessage, e))
+ callback.onFailure(new
CelebornIOException(StatusCode.PUSH_DATA_FAIL_SLAVE, e))
}
}
}
@@ -1004,9 +1004,9 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
s"attempt $attemptId, uniqueId $partitionUniqueId)."
logWarning(s"[handle$messageType] $msg")
messageType match {
- case Type.PUSH_MERGED_DATA => callback.onFailure(new Exception(msg))
+ case Type.PUSH_MERGED_DATA => callback.onFailure(new
CelebornIOException(msg))
case _ => callback.onFailure(
- new
Exception(StatusCode.PUSH_DATA_FAIL_PARTITION_NOT_FOUND.getMessage()))
+ new
CelebornIOException(StatusCode.PUSH_DATA_FAIL_PARTITION_NOT_FOUND))
}
}
return true
@@ -1026,19 +1026,19 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
messageType match {
case Type.PUSH_DATA | Type.PUSH_DATA_HAND_SHAKE =>
(
- StatusCode.PUSH_DATA_FAIL_MASTER.getMessage(),
- StatusCode.PUSH_DATA_FAIL_SLAVE.getMessage())
+ StatusCode.PUSH_DATA_FAIL_MASTER,
+ StatusCode.PUSH_DATA_FAIL_SLAVE)
case Type.PUSH_DATA_HAND_SHAKE => (
- StatusCode.PUSH_DATA_HANDSHAKE_FAIL_MASTER.getMessage,
- StatusCode.PUSH_DATA_HANDSHAKE_FAIL_SLAVE.getMessage)
+ StatusCode.PUSH_DATA_HANDSHAKE_FAIL_MASTER,
+ StatusCode.PUSH_DATA_HANDSHAKE_FAIL_SLAVE)
case Type.REGION_START => (
- StatusCode.REGION_START_FAIL_MASTER.getMessage,
- StatusCode.REGION_START_FAIL_SLAVE.getMessage)
+ StatusCode.REGION_START_FAIL_MASTER,
+ StatusCode.REGION_START_FAIL_SLAVE)
case Type.REGION_FINISH => (
- StatusCode.REGION_FINISH_FAIL_MASTER.getMessage,
- StatusCode.REGION_FINISH_FAIL_SLAVE.getMessage)
+ StatusCode.REGION_FINISH_FAIL_MASTER,
+ StatusCode.REGION_FINISH_FAIL_SLAVE)
}
- callback.onFailure(new Exception(
+ callback.onFailure(new CelebornIOException(
if (isMaster) messageMaster else messageSlave,
fileWriter.getException))
}
