[#825][part-4] feat(spark): Report write failures to ShuffleManager

[yl09099]

What changes were proposed in this pull request?

Send the write exception to the ShuffleServer to the ShuffleManager.

Ⅰ. Overall objective:

1. During the shuffle write phase, the ShuffleServer reports faulty nodes and reallocates the ShuffleServer list;
2. Triggers a Stage level retry of SPARK. The shuffleServer node is excluded and reallocated before the retry.

Ⅱ. Implementation logic diagram:

Ⅲ. As shown in the picture above:

1. During Shuffle registration, obtain the ShuffleServer list to be written through the RPC interface of a Coordinator Client by following the solid blue line step. The list is bound using ShuffleID.
   2, the Task of Stage starts, solid steps, in accordance with the green by ShuffleManager Client RPC interface gets to be written for shuffleIdToShuffleHandleInfo ShuffleServer list;
2. In the Stage, if Task fails to write blocks to the ShuffleServer, press the steps in red to report ShuffleServer to FailedShuffleServerList in RSSShuffleManager through the RPC interface.
3. FailedShuffleServerList records the number of ShuffleServer failures. After the number of failures reaches the maximum number of retries of the Task level, follow the steps in dotted orange lines. Through the RPC interface of a Coordinator Client, obtain the list of ShuffleServer files to be written (the ShuffleServer files that fail to be written are excluded). After obtaining the list, go to Step 5 of the dotted orange line. Throwing a FetchFailed Exception triggers a stage-level retry for SPARK;
4. Attempt 1 is generated by the SPARK Stage level again. Pull the corresponding ShuffleServer list according to the green dotted line.

Why are the changes needed?

Fix: #825

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing UT.

[codecov-commenter]

Codecov Report

Merging #1258 (59b58e1) into master (73ae342) will increase coverage by 0.72%.
The diff coverage is 1.47%.

@@             Coverage Diff              @@
##             master    #1258      +/-   ##
============================================
+ Coverage     53.70%   54.43%   +0.72%     
  Complexity     2655     2655              
============================================
  Files           398      380      -18     
  Lines         23157    20933    -2224     
  Branches       1976     1985       +9     
============================================
- Hits          12437    11395    -1042     
+ Misses         9954     8843    -1111     
+ Partials        766      695      -71     

Files	Coverage Δ
...fle/client/impl/grpc/ShuffleManagerGrpcClient.java	29.41% <0.00%> (-9.05%)	⬇️
...response/RssReportShuffleWriteFailureResponse.java	0.00% <0.00%> (ø)
...t/request/RssReportShuffleWriteFailureRequest.java	0.00% <0.00%> (ø)
...fle/shuffle/manager/ShuffleManagerGrpcService.java	33.16% <2.24%> (-24.36%)	⬇️

... and 20 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[yl09099]

@jerqi @leixm Please look at it for me.

[leixm]

LGTM, @jerqi Please take a look.

[roryqi]

Could we give this variable a better name?

[roryqi]

      if (this.stageAttemptNumber != stageAttemptNumber) {
              // do nothing here
              return false;
      }
      shuffleServerInfos.forEach(
                  shuffleServerInfo -> {
                    shuffleServerFailureRecordCount
                        .computeIfAbsent(shuffleServerInfo.getId(), k -> new AtomicInteger())
                        .incrementAndGet();
                  });
              List<Map.Entry<String, AtomicInteger>> list =
                  new ArrayList(shuffleServerFailureRecordCount.entrySet());
              Collections.sort(list, (o1, o2) -> (o1.getValue().get() - o2.getValue().get()));
              Map.Entry<String, AtomicInteger> shuffleServerInfoIntegerEntry = list.get(0);
     if (shuffleServerInfoIntegerEntry.getValue().get()
                  > shuffleManager.getMaxFetchFailures()) {
                shuffleManager.addFailuresShuffleServerInfos(
                    shuffleServerInfoIntegerEntry.getKey());
                return true;
      }
      return false;

More clear.

[roryqi]

Could we return bool type here?

[yl09099]

Could we return bool type here?

Have been changed.

[roryqi]

These methods seems useful. Could we extract some methods for other classes to use?

[yl09099]

These methods seems useful. Could we extract some methods for other classes to use?

Later PR will use this method.

[roryqi]

LGTM, thanks @leixm @yl09099

[leixm]

Waiting for UTs.

[leixm]

Merged, thanks @jerqi @yl09099