You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by an...@apache.org on 2023/06/02 06:04:44 UTC
[incubator-celeborn] branch main updated: [CELEBORN-621][BUG] Push merged data task timeout and mapended should also remove push states (#1526)
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 4f1ca8c96 [CELEBORN-621][BUG] Push merged data task timeout and mapended should also remove push states (#1526)
4f1ca8c96 is described below
commit 4f1ca8c960eba453d6abd64a1a2beaeb28a8ecf2
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Fri Jun 2 14:04:39 2023 +0800
[CELEBORN-621][BUG] Push merged data task timeout and mapended should also remove push states (#1526)
---
.../org/apache/celeborn/client/ShuffleClientImpl.java | 16 ++++++++++++++--
1 file changed, 14 insertions(+), 2 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 253b727a3..5bf9ea392 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
@@ -817,6 +817,7 @@ public class ShuffleClientImpl extends ShuffleClient {
pushState.onCongestControl(loc.hostAndPushPort());
callback.onSuccess(response);
} else {
+ // StageEnd.
response.rewind();
pushState.onSuccess(loc.hostAndPushPort());
callback.onSuccess(response);
@@ -1203,9 +1204,8 @@ public class ShuffleClientImpl extends ShuffleClient {
pushState.onCongestControl(hostPort);
callback.onSuccess(response);
} else {
- // Should not happen in current architecture.
+ // StageEnd.
response.rewind();
- logger.error("Push merged data should not receive this response.");
pushState.onSuccess(hostPort);
callback.onSuccess(response);
}
@@ -1260,6 +1260,18 @@ public class ShuffleClientImpl extends ShuffleClient {
cause,
groupedBatchId,
finalRemainReviveTimes));
+ } else {
+ pushState.removeBatch(groupedBatchId, hostPort);
+ logger.info(
+ "Push merged data to {} failed but mapper already ended for shuffle {} map {} attempt {} partition {} groupedBatch {} batch {}, remain revive times {}.",
+ hostPort,
+ shuffleId,
+ mapId,
+ attemptId,
+ Arrays.toString(partitionIds),
+ groupedBatchId,
+ Arrays.toString(batchIds),
+ remainReviveTimes);
}
}
};