yl09099 commented on code in PR #1845:
URL:
https://github.com/apache/incubator-uniffle/pull/1845#discussion_r1862989697
##########
client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/ShuffleManagerGrpcService.java:
##########
@@ -75,43 +79,62 @@ public void reportShuffleWriteFailure(
code = RssProtos.StatusCode.INVALID_REQUEST;
reSubmitWholeStage = false;
} else {
- Map<String, AtomicInteger> shuffleServerInfoIntegerMap =
JavaUtils.newConcurrentMap();
+ Map<String, AtomicInteger> initServerFailures =
JavaUtils.newConcurrentMap();
List<ShuffleServerInfo> shuffleServerInfos =
ShuffleServerInfo.fromProto(shuffleServerIdsList);
shuffleServerInfos.forEach(
- shuffleServerInfo -> {
- shuffleServerInfoIntegerMap.put(shuffleServerInfo.getId(), new
AtomicInteger(0));
- });
- ShuffleServerFailureRecord shuffleServerFailureRecord =
+ shuffleServerInfo ->
+ initServerFailures.computeIfAbsent(
+ shuffleServerInfo.getId(), key -> new AtomicInteger(0)));
+ ShuffleServerWriterFailureRecord shuffleServerWriterFailureRecord =
shuffleWriteStatus.computeIfAbsent(
shuffleId,
- key ->
- new ShuffleServerFailureRecord(shuffleServerInfoIntegerMap,
stageAttemptNumber));
+ key -> new ShuffleServerWriterFailureRecord(stageAttemptNumber,
initServerFailures));
boolean resetflag =
-
shuffleServerFailureRecord.resetStageAttemptIfNecessary(stageAttemptNumber);
+
shuffleServerWriterFailureRecord.resetStageAttemptIfNecessary(stageAttemptNumber);
if (resetflag) {
msg =
String.format(
- "got an old stage(%d vs %d) shuffle write failure report,
which should be impossible.",
- shuffleServerFailureRecord.getStageAttempt(),
stageAttemptNumber);
+ "got an old stage(%d_%d) shuffle write failure report, which
should be impossible.",
+ stageAttemptId, stageAttemptNumber);
LOG.warn(msg);
code = RssProtos.StatusCode.INVALID_REQUEST;
reSubmitWholeStage = false;
} else {
code = RssProtos.StatusCode.SUCCESS;
// update the stage shuffleServer write failed count
- boolean fetchFailureflag =
- shuffleServerFailureRecord.incPartitionWriteFailure(
+ boolean isFetchFailed =
+ shuffleServerWriterFailureRecord.incWriteFailureForShuffleServer(
stageAttemptNumber, shuffleServerInfos, shuffleManager);
- if (fetchFailureflag) {
+ if (isFetchFailed) {
reSubmitWholeStage = true;
msg =
String.format(
- "report shuffle write failure as maximum number(%d) of
shuffle write is occurred",
+ "Report shuffle write failure as maximum number(%d) of
shuffle write is occurred.",
shuffleManager.getMaxFetchFailures());
+ if (!shuffleServerWriterFailureRecord.isClearedMapTrackerBlock()) {
+ try {
+ // Clear the metadata of the completed task, otherwise some of
the stage's data will
+ // be lost.
+ shuffleManager.unregisterAllMapOutput(shuffleId);
Review Comment:
> Why the `unregisterAllMapOutput` is executed in the report phase rather
than in the `getPartitionToShufflerServerWithStageRetry` ?
Before assigning a Writer, clear corresponding Map tasks. Otherwise,
successful Map tasks will be missed, resulting in data loss.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
