[#477] feat(spark): support getShuffleResult throws FetchFailedException. #1004

leixm · 2023-07-11T09:34:07Z

What changes were proposed in this pull request?

feat #477

Why are the changes needed?

support getShuffleResult throws FetchFailedException.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Exiting UTs, Already added getShuffleResult failed, see MockedShuffleServerGrpcService.

leixm · 2023-07-11T09:34:23Z

cc @advancedxy @jerqi PTAL.

leixm · 2023-07-11T09:40:58Z

I try to shutdown a ShuffleServer during app running, it will throws FetchFailed, you can see here Fetch Failed image.

codecov-commenter · 2023-07-11T09:42:19Z

Codecov Report

Merging #1004 (99c6a16) into master (d6b43f0) will increase coverage by 1.09%.
The diff coverage is 0.00%.

@@             Coverage Diff              @@
##             master    #1004      +/-   ##
============================================
+ Coverage     53.66%   54.76%   +1.09%     
- Complexity     2523     2526       +3     
============================================
  Files           382      362      -20     
  Lines         21672    19333    -2339     
  Branches       1795     1799       +4     
============================================
- Hits          11630    10587    -1043     
+ Misses         9338     8115    -1223     
+ Partials        704      631      -73

Impacted Files	Coverage Δ
...org/apache/spark/shuffle/RssSparkShuffleUtils.java	`51.49% <0.00%> (-7.01%)`	⬇️
...he/uniffle/client/impl/ShuffleWriteClientImpl.java	`35.98% <0.00%> (-0.08%)`	⬇️

... and 27 files with indirect coverage changes

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

leixm · 2023-07-11T10:00:00Z

Seems flaky test.

advancedxy

Thanks for submitting this PR, left some minor comments.

However maybe we need some refactor to:

move getShuffleResult into ShuffleReadClient
the get result part should also in the RssShuffleReader, that might make the whole code a bit clearer.
But let's do that in follow-up prs.

advancedxy · 2023-07-11T14:10:29Z

client-spark/spark2/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java

+      RssConf rssConf = RssSparkConfig.toRssConf(sparkConf);
+      if (rssConf.getBoolean(RssClientConfig.RSS_RESUBMIT_STAGE, false)
+              && RssSparkShuffleUtils.isStageResubmitSupported()) {
+        String driver = rssConf.getString("driver.host", "");


this literal driver.host should be extracted as a constant field. This also applies to RssShuffleReader.

advancedxy · 2023-07-11T14:11:43Z

client-spark/spark2/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java

+        int port = rssConf.get(RssClientConf.SHUFFLE_MANAGER_GRPC_PORT);
+        try (ShuffleManagerClient client = ShuffleManagerClientFactory
+                .getInstance().createShuffleManagerClient(ClientType.GRPC, driver, port)) {
+          RssReportShuffleFetchFailureRequest req = new RssReportShuffleFetchFailureRequest(
+                  appId, shuffleId, stageAttemptId, partitionId, e.getMessage());
+          RssReportShuffleFetchFailureResponse response = client.reportShuffleFetchFailure(req);
+          if (response.getReSubmitWholeStage()) {
+            // since we are going to roll out the whole stage, mapIndex shouldn't matter, hence -1 is provided.
+            FetchFailedException ffe =
+                    RssSparkShuffleUtils.createFetchFailedException(shuffleId, -1, partitionId, e);
+            throw new RssException(ffe);
+          }
+        } catch (IOException ioe) {
+          LOG.info("Error closing shuffle manager client with error:", ioe);
+        }
+      }


could we extract this into a utility method and reuse code for both spark 2 and spark 3?

advancedxy · 2023-07-11T14:14:53Z

client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java

+            RssReportShuffleFetchFailureRequest req = new RssReportShuffleFetchFailureRequest(
+                    appId, shuffleId, stageAttemptId, partitionId, e.getMessage());
+            RssReportShuffleFetchFailureResponse response = client.reportShuffleFetchFailure(req);
+            if (response.getReSubmitWholeStage()) {


For this part, I think we should try to report all the shuffle fetch failures then throw FetchFailedException if any of response indicates the whole stage should be re-submitted.

P.S: one todo is create a new rpc interface to report failures in batch.

server/src/test/java/org/apache/uniffle/server/MockedShuffleServerGrpcService.java

leixm · 2023-07-12T06:31:52Z

I will raise sub PRs to address the following problem:

Create a new rpc interface to report failures in batch
move getShuffleResult into ShuffleReadClient
get result part move to RssShuffleReader

leixm · 2023-07-12T09:07:35Z

Flaky test.

common/src/main/java/org/apache/uniffle/common/util/Constants.java

advancedxy

LGTM, except one minor comment. Thanks for your work.

leixm · 2023-07-13T06:22:59Z

LGTM, except one minor comment. Thanks for your work.

cc @smallzhongfeng , Can you help review plz?

smallzhongfeng

Thanks for ping me. LTGM!

smallzhongfeng · 2023-07-13T11:28:30Z

client-spark/common/src/main/java/org/apache/spark/shuffle/RssSparkShuffleUtils.java

+      int port = rssConf.get(RssClientConf.SHUFFLE_MANAGER_GRPC_PORT);
+      try (ShuffleManagerClient client = ShuffleManagerClientFactory
+              .getInstance().createShuffleManagerClient(ClientType.GRPC, driver, port)) {
+        // todo: Create a new rpc interface to report failures in batch.


nit: remember create a new issue for this.

smallzhongfeng · 2023-07-13T11:29:12Z

client-spark/spark2/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java

@@ -581,4 +582,15 @@ public int getPartitionNum(int shuffleId) {
  public int getNumMaps(int shuffleId) {
    return shuffleIdToNumMapTasks.getOrDefault(shuffleId, 0);
  }
+
+  private Roaring64NavigableMap getShuffleResult(String clientType, Set<ShuffleServerInfo> shuffleServerInfoSet,
+                                                 String appId, int shuffleId, int partitionId, int stageAttemptId) {


nit: code indentation

smallzhongfeng · 2023-07-13T11:29:40Z

server/src/test/java/org/apache/uniffle/server/MockedShuffleServerGrpcService.java

+      int currentFailedReadRequest = failedReadRequest.getAndIncrement();
+      if (currentFailedReadRequest < numOfFailedReadRequest) {
+        LOG.info("This request is failed as mocked failure, current/firstN: {}/{}",
+                currentFailedReadRequest, numOfFailedReadRequest);


smallzhongfeng · 2023-07-13T11:29:54Z

server/src/test/java/org/apache/uniffle/server/MockedShuffleServerGrpcService.java

+      int currentFailedReadRequest = failedReadRequest.getAndIncrement();
+      if (currentFailedReadRequest < numOfFailedReadRequest) {
+        LOG.info("This request is failed as mocked failure, current/firstN: {}/{}",
+                currentFailedReadRequest, numOfFailedReadRequest);


…Exception.

jerqi · 2023-07-20T07:44:56Z

you can run this command to fix spotless issues.

mvn spotless:apply -Pspark3 -Pspark2 -Ptez -Pmr -Phadoop2.8

leixm · 2023-07-20T07:48:08Z

mvn spotless:apply -Pspark3 -Pspark2 -Ptez -Pmr -Phadoop2.8

Thank you. Already fixed.

jerqi

LGTM, thanks @leixm @smallzhongfeng @advancedxy , wait for CI.

jerqi · 2023-07-20T08:09:43Z

Merged to master.

jerqi requested a review from advancedxy July 11, 2023 09:46

advancedxy reviewed Jul 11, 2023

View reviewed changes

jerqi changed the title ~~[#477] feat: support getShuffleResult throws FetchFailedException.~~ [#477] feat(spark): support getShuffleResult throws FetchFailedException. Jul 13, 2023

advancedxy reviewed Jul 13, 2023

View reviewed changes

common/src/main/java/org/apache/uniffle/common/util/Constants.java Show resolved Hide resolved

advancedxy reviewed Jul 13, 2023

View reviewed changes

smallzhongfeng previously approved these changes Jul 13, 2023

View reviewed changes

smallzhongfeng reviewed Jul 13, 2023

View reviewed changes

[apache#477] feat(spark): support getShuffleResult throws FetchFailed…

99c6a16

…Exception.

leixm dismissed smallzhongfeng’s stale review via 99c6a16 July 20, 2023 07:15

leixm force-pushed the issue_1003 branch from ed7f7ee to 99c6a16 Compare July 20, 2023 07:15

Fix CheckStyle.

3a57086

Fix CheckStyle.

7cc7717

jerqi approved these changes Jul 20, 2023

View reviewed changes

jerqi merged commit d0f0b57 into apache:master Jul 20, 2023
30 checks passed

jerqi mentioned this pull request Aug 19, 2023

Improve the code of getShuffleResult #1159

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[#477] feat(spark): support getShuffleResult throws FetchFailedException. #1004

[#477] feat(spark): support getShuffleResult throws FetchFailedException. #1004

leixm commented Jul 11, 2023

leixm commented Jul 11, 2023

leixm commented Jul 11, 2023

codecov-commenter commented Jul 11, 2023 •

edited

Loading

leixm commented Jul 11, 2023

advancedxy left a comment

advancedxy Jul 11, 2023

advancedxy Jul 11, 2023

advancedxy Jul 11, 2023

leixm commented Jul 12, 2023

leixm commented Jul 12, 2023

advancedxy left a comment

leixm commented Jul 13, 2023

smallzhongfeng left a comment

smallzhongfeng Jul 13, 2023

smallzhongfeng Jul 13, 2023

smallzhongfeng Jul 13, 2023

smallzhongfeng Jul 13, 2023

jerqi commented Jul 20, 2023

leixm commented Jul 20, 2023

jerqi left a comment

jerqi commented Jul 20, 2023

[#477] feat(spark): support getShuffleResult throws FetchFailedException. #1004

[#477] feat(spark): support getShuffleResult throws FetchFailedException. #1004

Conversation

leixm commented Jul 11, 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?

leixm commented Jul 11, 2023

leixm commented Jul 11, 2023

codecov-commenter commented Jul 11, 2023 • edited Loading

Codecov Report

leixm commented Jul 11, 2023

advancedxy left a comment

Choose a reason for hiding this comment

advancedxy Jul 11, 2023

Choose a reason for hiding this comment

advancedxy Jul 11, 2023

Choose a reason for hiding this comment

advancedxy Jul 11, 2023

Choose a reason for hiding this comment

leixm commented Jul 12, 2023

leixm commented Jul 12, 2023

advancedxy left a comment

Choose a reason for hiding this comment

leixm commented Jul 13, 2023

smallzhongfeng left a comment

Choose a reason for hiding this comment

smallzhongfeng Jul 13, 2023

Choose a reason for hiding this comment

smallzhongfeng Jul 13, 2023

Choose a reason for hiding this comment

smallzhongfeng Jul 13, 2023

Choose a reason for hiding this comment

smallzhongfeng Jul 13, 2023

Choose a reason for hiding this comment

jerqi commented Jul 20, 2023

leixm commented Jul 20, 2023

jerqi left a comment

Choose a reason for hiding this comment

jerqi commented Jul 20, 2023

codecov-commenter commented Jul 11, 2023 •

edited

Loading