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 24f5478ad [CELEBORN-338] Clean duplicated exception message of
handling push data (#1274)
24f5478ad is described below
commit 24f5478adc703f8b3ac3445c54f801efe6e42f31
Author: Angerszhuuuu <[email protected]>
AuthorDate: Tue Feb 28 10:35:18 2023 +0800
[CELEBORN-338] Clean duplicated exception message of handling push data
(#1274)
---
.../apache/celeborn/client/ShuffleClientImpl.java | 187 ++++++++-------------
.../common/network/client/TransportClient.java | 3 +-
.../network/client/TransportResponseHandler.java | 4 +-
.../service/deploy/worker/PushDataHandler.scala | 138 +++++++--------
4 files changed, 141 insertions(+), 191 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 185f76cd6..f0c6c9476 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
@@ -194,16 +194,16 @@ public class ShuffleClientImpl extends ShuffleClient {
int partitionId = loc.getId();
if (!revive(
applicationId, shuffleId, mapId, attemptId, partitionId,
loc.getEpoch(), loc, cause)) {
- wrappedCallback.onFailure(
- new CelebornIOException("Revive Failed, remain revive times " +
remainReviveTimes));
+ wrappedCallback.onFailure(new
CelebornIOException(StatusCode.REVIVE_FAILED));
} else if (mapperEnded(shuffleId, mapId, attemptId)) {
logger.debug(
- "Revive for push data success, but the mapper already ended for
shuffle {} map {} attempt {} partition {} batch {}.",
+ "Revive for push data success, but the mapper already ended for
shuffle {} map {} attempt {} partition {} batch {} location {}.",
shuffleId,
mapId,
attemptId,
partitionId,
- batchId);
+ batchId,
+ loc);
pushState.removeBatch(batchId, loc.hostAndPushPort());
} else {
PartitionLocation newLoc =
reducePartitionMap.get(shuffleId).get(partitionId);
@@ -240,12 +240,7 @@ public class ShuffleClientImpl extends ShuffleClient {
newLoc,
e);
wrappedCallback.onFailure(
- new Exception(
- StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER
- + "! "
- + e.getMessage()
- + ". "
- + newLoc));
+ new
CelebornIOException(StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER, e));
}
}
}
@@ -277,10 +272,11 @@ public class ShuffleClientImpl extends ShuffleClient {
if (remainReviveTimes > 0) {
reviveFailedBatchesMap.add(batch);
} else {
- pushState.exception.compareAndSet(
- null,
- new CelebornIOException(
- "Revive Failed in retry push merged data for location: " +
batch.loc));
+ String errorMsg =
+ String.format(
+ "Revive failed while pushing merged for shuffle %d map %d
attempt %d partition %d batch %d location %s.",
+ shuffleId, mapId, attemptId, partitionId, oldGroupedBatchId,
batch.loc);
+ pushState.exception.compareAndSet(null, new
CelebornIOException(errorMsg));
return;
}
} else if (mapperEnded(shuffleId, mapId, attemptId)) {
@@ -310,9 +306,8 @@ public class ShuffleClientImpl extends ShuffleClient {
for (Map.Entry<String, DataBatches> entry : newDataBatchesMap.entrySet()) {
String addressPair = entry.getKey();
DataBatches newDataBatches = entry.getValue();
- String[] tokens = addressPair.split("-");
doPushMergedData(
- tokens[0],
+ addressPair,
applicationId,
shuffleId,
mapId,
@@ -387,7 +382,7 @@ public class ShuffleClientImpl extends ShuffleClient {
if (partitionLocationMap == null) {
String shuffleKey = Utils.makeShuffleKey(appId, shuffleId);
- throw new IOException("Register shuffle failed for shuffle " +
shuffleKey);
+ throw new CelebornIOException("Register shuffle failed for shuffle " +
shuffleKey);
}
return partitionLocationMap.get(partitionId);
@@ -626,7 +621,7 @@ public class ShuffleClientImpl extends ShuffleClient {
null,
StatusCode.PUSH_DATA_FAIL_NON_CRITICAL_CAUSE)) {
throw new CelebornIOException(
- "Revive for shuffle " + shuffleKey + " partition " + partitionId +
" failed.");
+ String.format("Revive for shuffle %d partition %d failed.",
shuffleKey, partitionId));
}
}
@@ -647,11 +642,8 @@ public class ShuffleClientImpl extends ShuffleClient {
final PartitionLocation loc = map.get(partitionId);
if (loc == null) {
throw new CelebornIOException(
- "Partition location for shuffle "
- + shuffleKey
- + " partitionId "
- + partitionId
- + " is NULL!");
+ String.format(
+ "Partition location for shuffle %d partition %d is NULL!",
shuffleKey, partitionId));
}
PushState pushState = getPushState(mapKey);
@@ -709,17 +701,11 @@ public class ShuffleClientImpl extends ShuffleClient {
@Override
public void onFailure(Throwable e) {
- pushState.exception.compareAndSet(
- null, new CelebornIOException("Revived PushData failed!",
e));
- logger.error(
- "Push data to {} failed for shuffle {} map {} attempt {}
partition {} batch {}.",
- loc.hostAndPushPort(),
- shuffleId,
- mapId,
- attemptId,
- partitionId,
- nextBatchId,
- e);
+ String errorMsg =
+ String.format(
+ "Push data to %s failed for shuffle $d map %d attempt %d
partition %d batch %d.",
+ loc, shuffleId, mapId, attemptId, partitionId,
nextBatchId);
+ pushState.exception.compareAndSet(null, new
CelebornIOException(errorMsg, e));
}
};
@@ -808,26 +794,23 @@ public class ShuffleClientImpl extends ShuffleClient {
}
if (remainReviveTimes <= 0) {
- callback.onFailure(
- new Exception(
- cause
- + "! Push data to master worker of "
- + loc
- + " failed: "
- + e.getMessage(),
- e));
-
+ if (e instanceof CelebornIOException) {
+ callback.onFailure(e);
+ } else {
+ callback.onFailure(new CelebornIOException(cause, e));
+ }
return;
}
logger.error(
- "Push data to {} failed for shuffle {} map {} attempt {}
partition {} batch {}.",
+ "Push data to {} failed for shuffle {} map {} attempt {}
partition {} batch {}, remain revive times {}.",
loc.hostAndPushPort(),
shuffleId,
mapId,
attemptId,
partitionId,
nextBatchId,
+ remainReviveTimes,
e);
// async retry push data
if (!mapperEnded(shuffleId, mapId, attemptId)) {
@@ -849,13 +832,14 @@ public class ShuffleClientImpl extends ShuffleClient {
} else {
pushState.removeBatch(nextBatchId, loc.hostAndPushPort());
logger.info(
- "Push data to {} failed but mapper already ended for
shuffle {} map {} attempt {} partition {} batch {}.",
+ "Push data to {} failed but mapper already ended for
shuffle {} map {} attempt {} partition {} batch {}, remain revive times {}.",
loc.hostAndPushPort(),
shuffleId,
mapId,
attemptId,
partitionId,
- nextBatchId);
+ nextBatchId,
+ remainReviveTimes);
}
}
};
@@ -868,8 +852,8 @@ public class ShuffleClientImpl extends ShuffleClient {
client.pushData(pushData, pushDataTimeout, wrappedCallback);
} else {
wrappedCallback.onFailure(
- new Exception(
- StatusCode.PUSH_DATA_FAIL_NON_CRITICAL_CAUSE.toString(),
+ new CelebornIOException(
+ StatusCode.PUSH_DATA_FAIL_NON_CRITICAL_CAUSE,
new RuntimeException("Mock push data first time failed.")));
}
} catch (Exception e) {
@@ -883,12 +867,7 @@ public class ShuffleClientImpl extends ShuffleClient {
loc,
e);
wrappedCallback.onFailure(
- new Exception(
- StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER
- + "! "
- + e.getMessage()
- + ". "
- + loc));
+ new
CelebornIOException(StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER, e));
}
} else {
// add batch data
@@ -899,7 +878,7 @@ public class ShuffleClientImpl extends ShuffleClient {
limitMaxInFlight(mapKey, pushState, loc.hostAndPushPort());
DataBatches dataBatches = pushState.takeDataBatches(addressPair);
doPushMergedData(
- addressPair.split("-")[0],
+ addressPair,
applicationId,
shuffleId,
mapId,
@@ -1022,7 +1001,7 @@ public class ShuffleClientImpl extends ShuffleClient {
batchesArr.remove(entry);
}
doPushMergedData(
- tokens[0],
+ entry.getKey(),
applicationId,
shuffleId,
mapId,
@@ -1034,7 +1013,7 @@ public class ShuffleClientImpl extends ShuffleClient {
}
private void doPushMergedData(
- String hostPort,
+ String addressPair,
String applicationId,
int shuffleId,
int mapId,
@@ -1042,6 +1021,7 @@ public class ShuffleClientImpl extends ShuffleClient {
ArrayList<DataBatches.DataBatch> batches,
PushState pushState,
int remainReviveTimes) {
+ String hostPort = addressPair.split("-")[0];
final String[] splits = hostPort.split(":");
final String host = splits[0];
final int port = Integer.parseInt(splits[1]);
@@ -1076,7 +1056,7 @@ public class ShuffleClientImpl extends ShuffleClient {
public void onSuccess(ByteBuffer response) {
logger.debug(
"Push merged data to {} success for shuffle {} map {} attempt
{} partition {} groupedBatch {} batch {}.",
- hostPort,
+ addressPair,
shuffleId,
mapId,
attemptId,
@@ -1095,32 +1075,29 @@ public class ShuffleClientImpl extends ShuffleClient {
@Override
public void onFailure(Throwable e) {
String errorMsg =
- (remainReviveTimes < maxReviveTimes ? "Revived push" : "Push")
- + " merged data to "
- + hostPort
- + " failed for map "
- + mapId
- + " attempt "
- + attemptId
- + " partition "
- + Arrays.toString(partitionIds)
- + " groupedBatch "
- + groupedBatchId
- + " batch "
- + Arrays.toString(batchIds)
- + ".";
+ String.format(
+ "Push merged data to %s failed for shuffle %d map %d
attempt %d partition %s groupedBatch %d batch %s, remain revive times %d.",
+ addressPair,
+ shuffleId,
+ mapId,
+ attemptId,
+ Arrays.toString(partitionIds),
+ groupedBatchId,
+ Arrays.toString(batchIds),
+ remainReviveTimes);
pushState.exception.compareAndSet(null, new
CelebornIOException(errorMsg, e));
if (logger.isDebugEnabled()) {
for (int i = 0; i < numBatches; i++) {
logger.debug(
- "Push merged data to {} failed for shuffle {} map {}
attempt {} partition {} groupedBatch {} batch {}.",
- hostPort,
+ "Push merged data to {} failed for shuffle {} map {}
attempt {} partition {} groupedBatch {} batch {}, remain revive times {}.",
+ addressPair,
shuffleId,
mapId,
attemptId,
partitionIds[i],
groupedBatchId,
- batchIds[i]);
+ batchIds[i],
+ remainReviveTimes);
}
}
}
@@ -1135,7 +1112,7 @@ public class ShuffleClientImpl extends ShuffleClient {
if (reason == StatusCode.HARD_SPLIT.getValue()) {
logger.info(
"Push merged data to {} hard split required for shuffle {}
map {} attempt {} partition {} groupedBatch {} batch {}.",
- hostPort,
+ addressPair,
shuffleId,
mapId,
attemptId,
@@ -1157,7 +1134,7 @@ public class ShuffleClientImpl extends ShuffleClient {
} else if (reason ==
StatusCode.PUSH_DATA_SUCCESS_MASTER_CONGESTED.getValue()) {
logger.debug(
"Push merged data to {} master congestion required for
shuffle {} map {} attempt {} partition {} groupedBatch {} batch {}.",
- hostPort,
+ addressPair,
shuffleId,
mapId,
attemptId,
@@ -1169,7 +1146,7 @@ public class ShuffleClientImpl extends ShuffleClient {
} else if (reason ==
StatusCode.PUSH_DATA_SUCCESS_SLAVE_CONGESTED.getValue()) {
logger.debug(
"Push merged data to {} slave congestion required for
shuffle {} map {} attempt {} partition {} groupedBatch {} batch {}.",
- hostPort,
+ addressPair,
shuffleId,
mapId,
attemptId,
@@ -1194,29 +1171,28 @@ public class ShuffleClientImpl extends ShuffleClient {
@Override
public void onFailure(Throwable e) {
StatusCode cause = getPushDataFailCause(e.getMessage());
+
if (pushState.exception.get() != null) {
return;
}
if (remainReviveTimes <= 0) {
- callback.onFailure(
- new Exception(
- cause
- + "! Push data to master worker of "
- + hostPort
- + " failed: "
- + e.getMessage(),
- e));
+ if (e instanceof CelebornIOException) {
+ callback.onFailure(e);
+ } else {
+ callback.onFailure(new CelebornIOException(cause, e));
+ }
return;
}
logger.error(
- "Push merged data to {} failed for shuffle {} map {} attempt
{} partition {} groupedBatch {} batch {}.",
- hostPort,
+ "Push merged data to {} failed for shuffle {} map {} attempt
{} partition {} groupedBatch {} batch {}, remain revive times {}.",
+ addressPair,
shuffleId,
mapId,
attemptId,
Arrays.toString(partitionIds),
groupedBatchId,
Arrays.toString(batchIds),
+ remainReviveTimes,
e);
if (!mapperEnded(shuffleId, mapId, attemptId)) {
pushDataRetryPool.submit(
@@ -1242,8 +1218,8 @@ public class ShuffleClientImpl extends ShuffleClient {
client.pushMergedData(mergedData, pushDataTimeout, wrappedCallback);
} else {
wrappedCallback.onFailure(
- new Exception(
- StatusCode.PUSH_DATA_FAIL_NON_CRITICAL_CAUSE.toString(),
+ new CelebornIOException(
+ StatusCode.PUSH_DATA_FAIL_NON_CRITICAL_CAUSE,
new RuntimeException("Mock push merge data failed.")));
}
} catch (Exception e) {
@@ -1255,15 +1231,10 @@ public class ShuffleClientImpl extends ShuffleClient {
Arrays.toString(partitionIds),
groupedBatchId,
Arrays.toString(batchIds),
- hostPort,
+ addressPair,
e);
wrappedCallback.onFailure(
- new Exception(
- StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER
- + "! "
- + e.getMessage()
- + ". "
- + hostPort));
+ new
CelebornIOException(StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER, e));
}
}
@@ -1614,16 +1585,11 @@ public class ShuffleClientImpl extends ShuffleClient {
return;
}
if (!mapperEnded(shuffleId, mapId, attemptId)) {
- pushState.exception.compareAndSet(
- null, new CelebornIOException("PushData byteBuf failed!",
e));
- logger.error(
- "Push data byteBuf to {} failed for shuffle {} map {}
attempt {} batch {}.",
- location.hostAndPushPort(),
- shuffleId,
- mapId,
- attemptId,
- nextBatchId,
- e);
+ String errorMsg =
+ String.format(
+ "Push data byteBuf to %s failed for shuffle %d map %d
attempt %d batch %d.",
+ location.hostAndPushPort(), shuffleId, mapId, attemptId,
nextBatchId);
+ pushState.exception.compareAndSet(null, new
CelebornIOException(errorMsg, e));
} else {
logger.warn(
"Push data to {} failed but mapper already ended for shuffle
{} map {} attempt {} batch {}.",
@@ -1650,12 +1616,7 @@ public class ShuffleClientImpl extends ShuffleClient {
location,
e);
callback.onFailure(
- new Exception(
- StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER
- + "! "
- + e.getMessage()
- + ". "
- + location));
+ new
CelebornIOException(StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_MASTER, e));
}
return totalLength;
}
diff --git
a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java
b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java
index 77acf7948..4bf6b93d7 100644
---
a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java
+++
b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java
@@ -35,6 +35,7 @@ import io.netty.util.concurrent.GenericFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.celeborn.common.exception.CelebornIOException;
import org.apache.celeborn.common.network.buffer.NioManagedBuffer;
import org.apache.celeborn.common.network.protocol.*;
import org.apache.celeborn.common.network.util.NettyUtils;
@@ -366,7 +367,7 @@ public class TransportClient implements Closeable {
@Override
protected void handleFailure(String errorMsg, Throwable cause) {
handler.removePushRequest(pushRequestId);
- callback.onFailure(new IOException(errorMsg, cause));
+ callback.onFailure(new CelebornIOException(errorMsg, cause));
}
}
}
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 bebca636d..da9e32385 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
@@ -269,10 +269,10 @@ public class TransportResponseHandler extends
MessageHandler<ResponseMessage> {
NettyUtils.getRemoteAddress(channel),
resp.errorString);
} else {
- listener.onFailure(new RuntimeException(resp.errorString));
+ listener.onFailure(new IOException(resp.errorString));
}
} else {
- info.callback.onFailure(new RuntimeException(resp.errorString));
+ info.callback.onFailure(new CelebornIOException(resp.errorString));
}
} else {
throw new IllegalStateException("Unknown response type: " +
message.type());
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 7d269267f..6ca9c6afe 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
@@ -92,11 +92,9 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
pushData.requestId,
() => {
val callback = new SimpleRpcResponseCallback(
- Type.PUSH_DATA,
client,
pushData.requestId,
- pushData.shuffleKey,
- pushData.partitionUniqueId)
+ pushData.shuffleKey)
shufflePartitionType.getOrDefault(pushData.shuffleKey,
PartitionType.REDUCE) match {
case PartitionType.REDUCE => handlePushData(
pushData,
@@ -115,11 +113,9 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
handlePushMergedData(
pushMergedData,
new SimpleRpcResponseCallback(
- Type.PUSH_MERGED_DATA,
client,
pushMergedData.requestId,
- pushMergedData.shuffleKey,
- pushMergedData.partitionUniqueIds.mkString(","))))
+ pushMergedData.shuffleKey)))
}
case rpcRequest: RpcRequest => handleRpcRequest(client, rpcRequest)
}
@@ -192,14 +188,12 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
// If there is no shuffle key in shuffleMapperAttempts but there is
shuffle key
// in StorageManager. This partition should be HARD_SPLIT partition
and
// after worker restart, some task still push data to this
HARD_SPLIT partition.
- logInfo(
- s"Receive push data for committed hard split partition of (shuffle
$shuffleKey, " +
- s"map $mapId attempt $attemptId)")
+ logInfo(s"Receive push data for committed hard split partition of " +
+ s"(shuffle $shuffleKey, map $mapId attempt $attemptId)")
callbackWithTimer.onSuccess(ByteBuffer.wrap(Array[Byte](StatusCode.HARD_SPLIT.getValue)))
} else {
- val msg = s"Partition location wasn't found for task(shuffle
$shuffleKey, map $mapId, " +
- s"attempt $attemptId, uniqueId ${pushData.partitionUniqueId})."
- logWarning(s"[handlePushData] $msg")
+ logWarning(s"While handle PushData, Partition location wasn't found
for " +
+ s"task(shuffle $shuffleKey, map $mapId, attempt $attemptId,
uniqueId ${pushData.partitionUniqueId}).")
callbackWithTimer.onFailure(
new
CelebornIOException(StatusCode.PUSH_DATA_FAIL_PARTITION_NOT_FOUND))
}
@@ -218,14 +212,16 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
val fileWriter = location.asInstanceOf[WorkingPartition].getFileWriter
val exception = fileWriter.getException
if (exception != null) {
- logWarning(s"[handlePushData] fileWriter $fileWriter has Exception
$exception")
- val message =
+ val cause =
if (isMaster) {
StatusCode.PUSH_DATA_FAIL_MASTER
} else {
StatusCode.PUSH_DATA_FAIL_SLAVE
}
- callbackWithTimer.onFailure(new CelebornIOException(s"$message!
$location", exception))
+ logError(
+ s"While handling PushData, throw $cause, fileWriter $fileWriter has
exception.",
+ exception)
+ callbackWithTimer.onFailure(new CelebornIOException(cause))
return
}
val diskFull =
@@ -262,9 +258,10 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
if (unavailablePeers.containsKey(peerWorker)) {
pushData.body().release()
workerSource.incCounter(WorkerSource.PushDataFailCount)
+ logError(
+ s"PushData replication failed caused by unavailable peer for
partitionLocation: $location")
callbackWithTimer.onFailure(
- new CelebornIOException(
- s"${StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE}! Peer
$peerWorker unavailable for $location!"))
+ new
CelebornIOException(StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE))
return
}
@@ -303,20 +300,17 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
}
override def onFailure(e: Throwable): Unit = {
- logError(s"[handlePushData.onFailure] partitionLocation:
$location", e)
+ logError(s"PushData replication failed for partitionLocation:
$location", e)
workerSource.incCounter(WorkerSource.PushDataFailCount)
- // Throw by slave peer worker
- if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_FAIL_SLAVE.name())) {
+ // 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()) ||
+
e.getMessage.startsWith(StatusCode.PUSH_DATA_TIMEOUT_SLAVE.name())) {
callbackWithTimer.onFailure(e)
- } 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 CelebornIOException(
- s"${StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE}! Push
data to peer of $location failed: ${e.getMessage}",
- e))
+ callbackWithTimer.onFailure(
+ new
CelebornIOException(StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE))
}
}
}
@@ -334,10 +328,11 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
pushData.body().release()
unavailablePeers.put(peerWorker, System.currentTimeMillis())
workerSource.incCounter(WorkerSource.PushDataFailCount)
+ logError(
+ s"PushData replication failed during connecting peer for
partitionLocation: $location",
+ e)
callbackWithTimer.onFailure(
- new CelebornIOException(
- s"${StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE}!
Create connection to peer $peerWorker failed for $location",
- e))
+ new
CelebornIOException(StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE))
}
}
})
@@ -449,16 +444,14 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
// it's probably because commitFiles for Had-Split happens.
if (shuffleMapperAttempts.containsKey(shuffleKey)) {
if (-1 != shuffleMapperAttempts.get(shuffleKey).get(mapId)) {
- val msg =
- s"Receive push merged data from speculative task(shuffle
$shuffleKey, map $mapId," +
- s" attempt $attemptId), but this mapper has already been
ended."
- logInfo(msg)
+ logInfo(s"Receive push merged data from speculative " +
+ s"task(shuffle $shuffleKey, map $mapId, attempt $attemptId), " +
+ s"but this mapper has already been ended.")
callbackWithTimer.onSuccess(
ByteBuffer.wrap(Array[Byte](StatusCode.STAGE_ENDED.getValue)))
} else {
- logInfo(
- s"Receive push merged data for committed hard split partition of
(shuffle $shuffleKey, " +
- s"map $mapId attempt $attemptId)")
+ logInfo(s"Receive push merged data for committed hard split
partition of " +
+ s"(shuffle $shuffleKey, map $mapId attempt $attemptId)")
callbackWithTimer.onSuccess(
ByteBuffer.wrap(Array[Byte](StatusCode.HARD_SPLIT.getValue)))
}
@@ -467,15 +460,13 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
// If there is no shuffle key in shuffleMapperAttempts but there
is shuffle key
// in StorageManager. This partition should be HARD_SPLIT
partition and
// after worker restart, some task still push data to this
HARD_SPLIT partition.
- logInfo(
- s"Receive push merged data for committed hard split partition of
(shuffle $shuffleKey, " +
- s"map $mapId attempt $attemptId)")
+ logInfo(s"Receive push merged data for committed hard split
partition of " +
+ s"(shuffle $shuffleKey, map $mapId attempt $attemptId)")
callbackWithTimer.onSuccess(
ByteBuffer.wrap(Array[Byte](StatusCode.HARD_SPLIT.getValue)))
} else {
- val msg = s"Partition location wasn't found for task(shuffle
$shuffleKey, map $mapId," +
- s" attempt $attemptId, uniqueId $id)."
- logWarning(s"[handlePushMergedData] $msg")
+ logWarning(s"While handling PushMergedData, Partition location
wasn't found for " +
+ s"task(shuffle $shuffleKey, map $mapId, attempt $attemptId,
uniqueId $id).")
callbackWithTimer.onFailure(
new
CelebornIOException(StatusCode.PUSH_DATA_FAIL_PARTITION_NOT_FOUND))
}
@@ -496,17 +487,16 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
val fileWriterWithException = fileWriters.find(_.getException != null)
if (fileWriterWithException.nonEmpty) {
val exception = fileWriterWithException.get.getException
- logDebug(s"[handlePushMergedData] fileWriter ${fileWriterWithException}"
+
- s" has Exception $exception")
- val message =
+ val cause =
if (isMaster) {
StatusCode.PUSH_DATA_FAIL_MASTER
} else {
StatusCode.PUSH_DATA_FAIL_SLAVE
}
- callbackWithTimer.onFailure(new CelebornIOException(
- s"$message! ${partitionIdToLocations.head._2}",
- exception))
+ logError(
+ s"While handling PushMergedData, throw $cause, fileWriter
$fileWriterWithException has exception.",
+ exception)
+ callbackWithTimer.onFailure(new CelebornIOException(cause))
return
}
fileWriters.foreach(_.incrementPendingWrites())
@@ -527,8 +517,10 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
if (unavailablePeers.containsKey(peerWorker)) {
pushMergedData.body().release()
workerSource.incCounter(WorkerSource.PushDataFailCount)
- callbackWithTimer.onFailure(new CelebornIOException(
- s"${StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE}! Peer
$peerWorker unavailable for $location!"))
+ logError(
+ s"PushMergedData replication failed caused by unavailable peer
for partitionLocation: $location")
+ callbackWithTimer.onFailure(
+ new
CelebornIOException(StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE))
return
}
@@ -561,19 +553,17 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
}
override def onFailure(e: Throwable): Unit = {
+ logError(s"PushMergedData replicate failed for
partitionLocation: $location", e)
workerSource.incCounter(WorkerSource.PushDataFailCount)
- // Throw by slave peer worker
- if
(e.getMessage.startsWith(StatusCode.PUSH_DATA_FAIL_SLAVE.name())) {
+ // 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()) ||
+
e.getMessage.startsWith(StatusCode.PUSH_DATA_TIMEOUT_SLAVE.name())) {
callbackWithTimer.onFailure(e)
- } 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 CelebornIOException(
- s"${StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE}! Push
data to peer of ${partitionIdToLocations.head._2} failed: ${e.getMessage}",
- e))
+ callbackWithTimer.onFailure(
+ new
CelebornIOException(StatusCode.PUSH_DATA_CONNECTION_EXCEPTION_SLAVE))
}
}
}
@@ -598,10 +588,11 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
pushMergedData.body().release()
unavailablePeers.put(peerWorker, System.currentTimeMillis())
workerSource.incCounter(WorkerSource.PushDataFailCount)
+ logError(
+ s"PushMergedData replication failed during connecting peer for
partitionLocation: $location",
+ e)
callbackWithTimer.onFailure(
- new CelebornIOException(
- s"${StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE}!
Create connection to peer $peerWorker failed for $location",
- e))
+ new
CelebornIOException(StatusCode.PUSH_DATA_CREATE_CONNECTION_FAIL_SLAVE))
}
}
})
@@ -699,11 +690,9 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
}
class SimpleRpcResponseCallback(
- messageType: Message.Type,
client: TransportClient,
requestId: Long,
- shuffleKey: String,
- partitionUniqueId: String)
+ shuffleKey: String)
extends RpcResponseCallback {
override def onSuccess(response: ByteBuffer): Unit = {
client.getChannel.writeAndFlush(new RpcResponse(
@@ -712,9 +701,6 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
}
override def onFailure(e: Throwable): Unit = {
- logError(
- s"[process$messageType] Process $messageType onFailure!
ShuffleKey:$shuffleKey , partitionUniqueId: $partitionUniqueId",
- e)
client.getChannel.writeAndFlush(new RpcFailure(requestId, e.getMessage))
}
}
@@ -843,11 +829,9 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
requestId,
isCheckSplit,
new SimpleRpcResponseCallback(
- msg.`type`(),
client,
requestId,
- shuffleKey,
- partitionUniqueId)))
+ shuffleKey)))
}
@@ -978,7 +962,11 @@ class PushDataHandler extends BaseMessageHandler with
Logging {
callback.onFailure(new
CelebornIOException(StatusCode.REGION_FINISH_FAIL_SLAVE, e))
case _ =>
workerSource.incCounter(WorkerSource.PushDataFailCount)
- callback.onFailure(new
CelebornIOException(StatusCode.PUSH_DATA_FAIL_SLAVE, e))
+ if (e.isInstanceOf[CelebornIOException]) {
+ callback.onFailure(e)
+ } else {
+ callback.onFailure(new
CelebornIOException(StatusCode.PUSH_DATA_FAIL_SLAVE, e))
+ }
}
}
}
