[#825][part-5] feat(spark): Adds the RPC interface to reassign the ShuffleServer list. by yl09099 · Pull Request #1146 · apache/uniffle

yl09099 · 2023-08-14T05:31:47Z

What changes were proposed in this pull request?

Adds the RPC interface to reassign the ShuffleServer list.

Ⅰ. Overall objective:

During the shuffle write phase, the ShuffleServer reports faulty nodes and reallocates the ShuffleServer list;
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:

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;
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.
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;
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 · 2023-08-14T06:12:29Z

Codecov Report

Merging #1146 (f26a569) into master (5a89609) will increase coverage by 0.86%.
The diff coverage is 0.00%.

@@             Coverage Diff              @@
##             master    #1146      +/-   ##
============================================
+ Coverage     53.74%   54.61%   +0.86%     
+ Complexity     2691     2679      -12     
============================================
  Files           401      383      -18     
  Lines         23406    21084    -2322     
  Branches       1992     1992              
============================================
- Hits          12580    11515    -1065     
+ Misses        10051     8868    -1183     
+ Partials        775      701      -74

Files	Coverage Δ
...fle/client/impl/grpc/ShuffleManagerGrpcClient.java	`26.31% <0.00%> (-3.10%)`	⬇️
...fle/client/response/RssReassignServersReponse.java	`0.00% <0.00%> (ø)`
...ffle/client/request/RssReassignServersRequest.java	`0.00% <0.00%> (ø)`
...fle/shuffle/manager/ShuffleManagerGrpcService.java	`31.01% <0.00%> (-2.15%)`	⬇️

... and 23 files with indirect coverage changes

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

yl09099 · 2023-10-31T10:01:59Z

@jerqi @leixm Please look at it for me.

roryqi · 2023-10-31T10:58:05Z

   */
  void addFailuresShuffleServerInfos(String shuffleServerId);
+
+  boolean reShuffleAssignments(int stageId, int stageAttemptNumber, int shuffleId, int numMaps);


reShuffleAssignments -> reassignShuffleServers ?

roryqi · 2023-10-31T10:58:36Z

  }

+  @Override
+  public void reallocationShuffleServers(


reallocationShuffleServers -> reassignShuffleServers?

roryqi · 2023-10-31T10:59:30Z

+     * If the shuffleServer is reassigned, we need to clear the metadata of the completed Task,
+     * otherwise some of the Stage's data will be lost
+     */
+    if (reallocationFlag) {


reallocation -> needReassign ?

roryqi · 2023-10-31T11:01:37Z

  public void addFailuresShuffleServerInfos(String shuffleServerId) {}
+
+  @Override
+  public boolean reShuffleAssignments(


reShuffleAssignments -> reassignShuffleServers

roryqi · 2023-10-31T11:01:52Z

+   * @param numPartitions
+   */
+  @Override
+  public synchronized boolean reShuffleAssignments(


roryqi · 2023-10-31T11:03:34Z

+   * Prevent multiple tasks from reporting FetchFailed, resulting in multiple ShuffleServer
+   * assignments, stageID, Attemptnumber Whether to reallocate the combination flag;
+   */
+  private Map<String, Boolean> stageAttemptTag;


Could we give this variable a better name?

roryqi · 2023-10-31T11:04:04Z

@leixm Could you help me review this pr?

roryqi · 2023-11-01T02:38:19Z

+   * Prevent multiple tasks from reporting FetchFailed, resulting in multiple ShuffleServer
+   * assignments, stageID, Attemptnumber Whether to reallocate the combination flag;
+   */
+  private Map<String, Boolean> preventDoubleAllocationFlag = JavaUtils.newConcurrentMap();


serversAssignments?

roryqi · 2023-11-01T02:39:11Z

+  public synchronized boolean reassignShuffleServers(
+      int stageId, int stageAttemptNumber, int shuffleId, int numPartitions) {
+    String stageIdAndAttempt = stageId + "_" + stageAttemptNumber;
+    Boolean reassignFlag =


reassignFlag -> needReassgin

roryqi · 2023-11-01T02:39:38Z

+   * Prevent multiple tasks from reporting FetchFailed, resulting in multiple ShuffleServer
+   * assignments, stageID, Attemptnumber Whether to reallocate the combination flag;
+   */
+  private Map<String, Boolean> preventDoubleAllocationFlag;


serverAssignedInfos?

roryqi · 2023-11-01T02:40:24Z

+    String stageIdAndAttempt = stageId + "_" + stageAttemptNumber;
+    Boolean reallocationFlag =
+        preventDoubleAllocationFlag.computeIfAbsent(stageIdAndAttempt, id -> false);
+    if (!reallocationFlag) {


reallocationFlag -> needReassign?

roryqi · 2023-11-01T02:41:21Z

+
+public class RssReassignServersReponse extends ClientResponse {
+
+  private boolean reallocationFlag;


rellactionFlag -> needReassign?

roryqi · 2023-11-01T02:41:53Z

  rpc getPartitionToShufflerServer(PartitionToShuffleServerRequest) returns (PartitionToShuffleServerResponse);
  // Report write failures to ShuffleManager
  rpc reportShuffleWriteFailure (ReportShuffleWriteFailureRequest) returns (ReportShuffleWriteFailureResponse);
+  // Reallocate the RPC interface of the ShuffleServer list


Reallocate -> Reassign?

roryqi · 2023-11-01T02:42:37Z

  }
+
+  @Override
+  public RssReassignServersReponse reallocationShuffleServers(RssReassignServersRequest req) {


realloctionShuffleServers -> reassignShuffleServers?

roryqi · 2023-11-01T02:45:29Z

+        throw new RssException("registerShuffle failed!", throwable);
+      }
+      /**
+       * we need to clear the metadata of the completed Task, otherwise some of the Stage's data


completed Task -> completed tasks

Stage -> stage.

roryqi

Two questions:
If one shuffle server is allocated by two stages, one stage trigger the reassignment, what will it happen?
Do we remove the shuffle server and then won't send the heartbeat after we trigger the reassignment?

roryqi · 2023-11-01T02:51:10Z

+
+message ReassignServersReponse{
+  StatusCode status = 1;
+  bool reallocationFlag = 2;


needReassign?

yl09099 · 2023-11-01T06:10:47Z

Two questions: If one shuffle server is allocated by two stages, one stage trigger the reassignment, what will it happen? Do we remove the shuffle server and then won't send the heartbeat after we trigger the reassignment?

If the same ShuffleServer triggers a retry on the first stage, the task in the second Stage will be sensed once the failure is triggered and the Stage retry will be triggered.
Heartbeat problem resolved.

…the ShuffleServer list.

roryqi · 2023-11-02T01:49:34Z

LGTM, let @leixm take an another look.

leixm · 2023-11-03T03:44:47Z

+      int retryTimes = sparkConf.get(RssSparkConfig.RSS_CLIENT_ASSIGNMENT_RETRY_TIMES);
+      int estimateTaskConcurrency = RssSparkShuffleUtils.estimateTaskConcurrency(sparkConf);
+      /** Before reassigning ShuffleServer, clear the ShuffleServer list in ShuffleWriteClient. */
+      shuffleWriteClient.unregisterShuffle(appId, shuffleId);


When cleaning up at this time, some tasks are reading this data. Will this have any impact?

When cleaning up at this time, some tasks are reading this data. Will this have any impact?

Write failures are handled here, and read cases should not be involved.

leixm · 2023-11-03T03:48:37Z

+       * will be lost
+       */
+      try {
+        unregisterAllMapOutput(shuffleId);


What happens if there are not enough ShuffleServers to assign at this time?

What happens if there are not enough ShuffleServers to assign at this time?

The ShuffleServer allocation was done before the metadata was cleaned, and the amount was insufficient to throw an exception directly, a few lines of code before this code.

leixm · 2023-11-03T06:10:28Z

LGTM.

leixm · 2023-11-03T06:12:24Z

Merged, thanks @jerqi @yl09099

yl09099 changed the title ~~[#825][part-4] feat(spark): Adds the RPC interface to reassign the ShuffleServer list~~ [#825][part-5] feat(spark): Adds the RPC interface to reassign the ShuffleServer list Aug 14, 2023

yl09099 force-pushed the uniffle-825-5 branch from 4e6aaab to d5d391b Compare August 14, 2023 06:29

yl09099 closed this Oct 10, 2023

yl09099 force-pushed the uniffle-825-5 branch from d5d391b to 64c4a68 Compare October 10, 2023 07:12

yl09099 reopened this Oct 10, 2023

yl09099 force-pushed the uniffle-825-5 branch from 59ce524 to 021924f Compare October 10, 2023 07:15

yl09099 force-pushed the uniffle-825-5 branch from 021924f to 18782cc Compare October 31, 2023 09:54

yl09099 changed the title ~~[#825][part-5] feat(spark): Adds the RPC interface to reassign the ShuffleServer list~~ [#825][part-5] feat(spark): Adds the RPC interface to reassign the ShuffleServer list. Oct 31, 2023

yl09099 force-pushed the uniffle-825-5 branch from 18782cc to f6d41c6 Compare October 31, 2023 09:57

roryqi reviewed Oct 31, 2023

View reviewed changes

roryqi requested a review from leixm October 31, 2023 11:03

yl09099 force-pushed the uniffle-825-5 branch 5 times, most recently from 5883c28 to 78b115e Compare November 1, 2023 02:33

roryqi reviewed Nov 1, 2023

View reviewed changes

yl09099 force-pushed the uniffle-825-5 branch 2 times, most recently from 633502d to bcf10a1 Compare November 1, 2023 06:07

[apache#825][part-5] feat(spark): Adds the RPC interface to reassign …

f26a569

…the ShuffleServer list.

yl09099 force-pushed the uniffle-825-5 branch from bcf10a1 to f26a569 Compare November 1, 2023 06:18

leixm reviewed Nov 3, 2023

View reviewed changes

leixm approved these changes Nov 3, 2023

View reviewed changes

leixm merged commit adafac4 into apache:master Nov 3, 2023


		public class RssReassignServersReponse extends ClientResponse {

		private boolean reallocationFlag;

Uh oh!

Conversation

yl09099 commented Aug 14, 2023

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

codecov-commenter commented Aug 14, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

yl09099 commented Oct 31, 2023

Uh oh!

roryqi Oct 31, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

roryqi Oct 31, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

roryqi commented Oct 31, 2023

Uh oh!

roryqi Nov 1, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

roryqi left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yl09099 commented Nov 1, 2023

Uh oh!

roryqi commented Nov 2, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leixm commented Nov 3, 2023

Uh oh!

leixm commented Nov 3, 2023

Uh oh!

Reviewers

Assignees

Labels

codecov-commenter commented Aug 14, 2023 •

edited

Loading

roryqi Oct 31, 2023 •

edited

Loading

roryqi Oct 31, 2023 •

edited

Loading

roryqi Nov 1, 2023 •

edited

Loading