[#825][part-2] feat(spark): Report failed blocks and a list of ShuffleServer by yl09099 · Pull Request #1138 · apache/uniffle

yl09099 · 2023-08-12T17:08:01Z

What changes were proposed in this pull request?

The ShuffleServer corresponding to the block that failed to be sent needs to be reported.

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

Reports the ShuffleServer corresponding to the block that failed to be sent

Fix: #825

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing UT

codecov-commenter · 2023-08-12T17:16:27Z

Codecov Report

Merging #1138 (0500a29) into master (98b0d12) will increase coverage by 0.02%.
Report is 2 commits behind head on master.
The diff coverage is 42.10%.

@@             Coverage Diff              @@
##             master    #1138      +/-   ##
============================================
+ Coverage     54.55%   54.58%   +0.02%     
  Complexity     2613     2613              
============================================
  Files           375      375              
  Lines         20224    20307      +83     
  Branches       1930     1932       +2     
============================================
+ Hits          11033    11084      +51     
- Misses         8526     8554      +28     
- Partials        665      669       +4

Files	Coverage Δ
...uniffle/client/response/SendShuffleDataResult.java	`46.15% <57.14%> (-37.18%)`	⬇️
...va/org/apache/spark/shuffle/writer/DataPusher.java	`62.26% <44.44%> (-3.65%)`	⬇️
...he/uniffle/client/impl/ShuffleWriteClientImpl.java	`33.97% <36.36%> (-1.82%)`	⬇️

... and 4 files with indirect coverage changes

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

roryqi · 2023-10-09T02:24:43Z

+      return;
+    }
+    taskToFailedBlockIdsAndServer
+        .computeIfAbsent(taskAttemptId, x -> Maps.newConcurrentMap())


JavaUtils.newConcurrentMap()?

roryqi · 2023-10-09T02:25:07Z

+  public Map<Long, List<ShuffleServerInfo>> getFailedBlockIdsWithShuffleServer(String taskId) {
+    Map<Long, List<ShuffleServerInfo>> result = taskToFailedBlockIdsAndServer.get(taskId);
+    if (result == null) {
+      result = Maps.newHashMap();


JavaUtils.newConcurrentMap?

roryqi · 2023-10-09T02:25:33Z

-    // if send block failed, the task will fail
-    // todo: better to have fallback solution when send to multiple servers
+    /** Records the ShuffleServer that successfully or failed to send blocks */
+    Map<Long, List<ShuffleServerInfo>> blockIdSendSuccessTracker = Maps.newConcurrentMap();


JavaUtils.newConcurrentMap?

roryqi · 2023-10-09T10:38:22Z

@leixm Could you help me review this pr?

roryqi · 2023-10-09T12:18:38Z

        .addAll(blockIds);
  }

+  private synchronized void putBlockIdAndShuffle(


Could we give this method a better name? I can't get the point of put shuffle.

Will it influence the performance of pusher if we use the lock here?

Could we give this method a better name? I can't get the point of put shuffle.

Will it influence the performance of pusher if we use the lock here?

The name has been changed.

The lock here is the same logic as the putBlockId method above and should not affect it.

roryqi

LGTM, let @leixm take an another look.

advancedxy

Generally lgtm, except one minor comment.

advancedxy · 2023-10-16T02:29:46Z

+        return new SendShuffleDataResult(
+            Sets.newHashSet(2L), Sets.newHashSet(1L), Maps.newHashMap());


I don't think this line is needed? Since the constructor of SentShuffleDataResult is kept for compatibility.

I don't think this line is needed? Since the constructor of SentShuffleDataResult is kept for compatibility.

Have been modified.

advancedxy · 2023-10-16T02:29:52Z

          successBlockIds.add(blockInfo.getBlockId());
        }
-        return new SendShuffleDataResult(successBlockIds, Sets.newHashSet());
+        return new SendShuffleDataResult(successBlockIds, Sets.newHashSet(), Maps.newHashMap());


advancedxy · 2023-10-16T02:38:44Z

+   * @param taskId Shuffle taskId
+   * @return List of failed ShuffleServer blocks
+   */
+  public Map<Long, List<ShuffleServerInfo>> getFailedBlockIdsWithShuffleServer(String taskId) {


when is this method called?

And it seems a bit unusual to return an empty map without putting it back to the taskToFailedBlockIdsAndServer.

when is this method called?

And it seems a bit unusual to return an empty map without putting it back to the taskToFailedBlockIdsAndServer.

Later PR will call this method.

advancedxy · 2023-10-16T02:39:45Z

-    // todo: better to have fallback solution when send to multiple servers
+    /** Records the ShuffleServer that successfully or failed to send blocks */
+    Map<Long, List<ShuffleServerInfo>> blockIdSendSuccessTracker = JavaUtils.newConcurrentMap();
+    Map<Long, List<ShuffleServerInfo>> blockIdsSendFailTracker = JavaUtils.newConcurrentMap();


Reminder: let's review the memory usage for this change after all the code has been finished.

I'm a bit of worried about the memory overhead.

It seems not. The ShuffleServerInfo object in the map is reused.

leixm · 2023-10-16T06:49:16Z

+1 LGTM.

…ShuffleServer.

roryqi

LGTM, thanks @advancedxy @yl09099 @leixm

roryqi requested a review from advancedxy August 12, 2023 17:08

yl09099 force-pushed the uniffle-825-2 branch 4 times, most recently from bf311fa to 065a7ba Compare August 13, 2023 04:01

roryqi changed the title ~~[#825][part-2] feat:Report failed blocks and a list of ShuffleServer~~ [#825][part-2] feat(spark): Report failed blocks and a list of ShuffleServer Aug 13, 2023

yl09099 closed this Oct 9, 2023

yl09099 force-pushed the uniffle-825-2 branch from 065a7ba to 1406e1f Compare October 9, 2023 00:15

yl09099 reopened this Oct 9, 2023

yl09099 closed this Oct 9, 2023

yl09099 force-pushed the uniffle-825-2 branch from 7fe3059 to fa6d162 Compare October 9, 2023 02:20

yl09099 reopened this Oct 9, 2023

roryqi reviewed Oct 9, 2023

View reviewed changes

yl09099 force-pushed the uniffle-825-2 branch 4 times, most recently from 6fdbd28 to 406c285 Compare October 9, 2023 06:47

roryqi requested a review from leixm October 9, 2023 10:38

roryqi reviewed Oct 9, 2023

View reviewed changes

yl09099 force-pushed the uniffle-825-2 branch 2 times, most recently from cf24af4 to 0500a29 Compare October 11, 2023 06:17

roryqi reviewed Oct 11, 2023

View reviewed changes

advancedxy reviewed Oct 16, 2023

View reviewed changes

leixm previously approved these changes Oct 16, 2023

View reviewed changes

yl09099 closed this Oct 20, 2023

yl09099 force-pushed the uniffle-825-2 branch from 9c4a069 to 98b0d12 Compare October 20, 2023 11:17

[apache#825][part-2] feat(spark): Report failed blocks and a list of …

760cb23

…ShuffleServer.

yl09099 reopened this Oct 20, 2023

yl09099 dismissed leixm’s stale review via 760cb23 October 20, 2023 11:37

roryqi approved these changes Oct 20, 2023

View reviewed changes

roryqi merged commit 748acbf into apache:master Oct 20, 2023

		return new SendShuffleDataResult(
		Sets.newHashSet(2L), Sets.newHashSet(1L), Maps.newHashMap());

Uh oh!

Conversation

yl09099 commented Aug 12, 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 12, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

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 9, 2023

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!

advancedxy left a comment

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!

leixm commented Oct 16, 2023

Uh oh!

roryqi left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

codecov-commenter commented Aug 12, 2023 •

edited

Loading