[#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer

[yl09099]

What changes were proposed in this pull request?

We need to provide an RPC interface for reassigning ShuffleServer, with support for excluding some ShuffleServer lists.

Ⅰ. 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?

Such an interface is required when dynamically allocating a shuffleServer list.

Fix: #825

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Add UT.

[codecov-commenter]

Codecov Report

Merging #1137 (2cff5ca) into master (81844b0) will increase coverage by 1.11%.
The diff coverage is 41.17%.

@@             Coverage Diff              @@
##             master    #1137      +/-   ##
============================================
+ Coverage     53.76%   54.88%   +1.11%     
- Complexity     2604     2613       +9     
============================================
  Files           392      373      -19     
  Lines         22507    20174    -2333     
  Branches       1889     1893       +4     
============================================
- Hits          12102    11072    -1030     
+ Misses         9695     8461    -1234     
+ Partials        710      641      -69     

Files	Coverage Δ
...r/strategy/assignment/BasicAssignmentStrategy.java	95.83% <100.00%> (+0.59%)	⬆️
...assignment/PartitionBalanceAssignmentStrategy.java	95.71% <100.00%> (+0.12%)	⬆️
.../apache/uniffle/client/api/ShuffleWriteClient.java	0.00% <0.00%> (ø)
...he/uniffle/client/impl/ShuffleWriteClientImpl.java	36.08% <0.00%> (-0.16%)	⬇️
...he/uniffle/coordinator/CoordinatorGrpcService.java	1.88% <0.00%> (-0.02%)	⬇️
...niffle/client/impl/grpc/CoordinatorGrpcClient.java	0.00% <0.00%> (ø)
...ache/uniffle/coordinator/SimpleClusterManager.java	76.34% <54.54%> (-1.38%)	⬇️
...lient/request/RssGetShuffleAssignmentsRequest.java	0.00% <0.00%> (ø)

... and 22 files with indirect coverage changes

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

[yl09099]

@jerqi @advancedxy I have run mvn spotless:apply locally, why is there still a format error.

[roryqi]

@jerqi @advancedxy I have run mvn spotless:apply locally, why is there still a format error.

You don't enable profile hadoop2.8.

[advancedxy]

I will take a look at this today or tomorrow.

The follow chart looks awesome on the surface.

Also cc @zuston if you have interest to review this.

[advancedxy]

What changes were proposed in this pull request?

We need to provide an RPC interface for reassigning ShuffleServer, with support for excluding some ShuffleServer lists.

Ⅰ. 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?

Such an interface is required when dynamically allocating a shuffleServer list.

Fix: #825

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Add UT.

I did a quick overview of this diagram, seems like that it only handles shuffle write failures, how about the shuffle read failure? In which case, some of the shuffle servers are done or unable to serving shuffle data, the shuffle read client would report FetchFailedException to trigger a parent stage recompute. Would you mind to elaborate a bit more on how that would be handled?

And by the way, I don't think it's a good idea to throw an FetchFailedException when writing to uniffle server has been failed for multiple times.

[advancedxy]

I believe some refactor is needed to reduce code duplication.

[zuston]

And by the way, I don't think it's a good idea to throw an FetchFailedException when writing to uniffle server has been failed for multiple times.

For writing phase, I think the mechanism of task retry is enough. If the server is bad or out-of-service, the dynamic assignment is necessary when the spark task is retrying.

[roryqi]

@advancedxy @yl09099 @zuston Let us discuss this feature in the meeting next week.

[advancedxy]

This is in good shape. Just some minor comments.

@zuston Please take another look, when you have time.

[advancedxy]

I did a quick overview of this diagram, seems like that it only handles shuffle write failures, how about the shuffle read failure? In which case, some of the shuffle servers are done or unable to serving shuffle data, the shuffle read client would report FetchFailedException to trigger a parent stage recompute. Would you mind to elaborate a bit more on how that would be handled?

And by the way, I don't think it's a good idea to throw an FetchFailedException when writing to uniffle server has been failed for multiple times.

Also kindly remind of these two high level questions.

[yl09099]

I did a quick overview of this diagram, seems like that it only handles shuffle write failures, how about the shuffle read failure? In which case, some of the shuffle servers are done or unable to serving shuffle data, the shuffle read client would report FetchFailedException to trigger a parent stage recompute. Would you mind to elaborate a bit more on how that would be handled?
And by the way, I don't think it's a good idea to throw an FetchFailedException when writing to uniffle server has been failed for multiple times.

Also kindly remind of these two high level questions.

1、Read failure Someone has submitted the relevant PR earlier#787,There is a situation that is indeed not implemented, and there is no way to notify the upstream rewrite when the read fails, which would like to be implemented later.
2、FetchFailedException is a way to handle a read failure without intruding into SPARK code. I can't think of any other better way. Do you have any good suggestions?

[zuston]

This is in good shape. Just some minor comments.

@zuston Please take another look, when you have time.

Sorry for the late reply, I have to say sorry again that don't have much time to review for this feature.

[yl09099]

This is in good shape. Just some minor comments.
@zuston Please take another look, when you have time.

Sorry for the late reply, I have to say sorry again that don't have much time to review for this feature.

Please look at it for me when you have time.Please trigger the compilation for me again, I didn't change anything in Tez.

[zuston]

LGTM. Let's merge this firstly

[zuston]

Oh, please take a look for these requested changes. The resolve operation by me is not valid for this merging. @advancedxy .

And I think we could push forward quickly for this feature, If I understand correctly, this has been applied in DIDI . cc @yl09099

[advancedxy]

LGTM. Let's merge this first

[yl09099]

Oh, please take a look for these requested changes. The resolve operation by me is not valid for this merging. @advancedxy .

And I think we could push forward quickly for this feature, If I understand correctly, this has been applied in DIDI . cc @yl09099

OK,I'll merge as soon as possible.