[#1373][FOLLOWUP] fix(spark): shuffle manager rpc service invalid when partition data reassign is enabled #1583

dingshun3016 · 2024-03-14T13:31:32Z

What changes were proposed in this pull request?

Fix task fail retry parameter not work and add task fail retry parameter in other locations

Why are the changes needed?

Fix: (#1373)

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Not necessary.

… work

codecov-commenter · 2024-03-14T13:39:40Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 54.85%. Comparing base (f6569b0) to head (c79ef03).
Report is 11 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #1583      +/-   ##
============================================
+ Coverage     53.99%   54.85%   +0.85%     
- Complexity     2862     2864       +2     
============================================
  Files           438      418      -20     
  Lines         24819    22551    -2268     
  Branches       2111     2123      +12     
============================================
- Hits          13402    12371    -1031     
+ Misses        10578     9408    -1170     
+ Partials        839      772      -67

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

github-actions · 2024-03-14T14:04:07Z

Test Results

2 109 files - 231 2 109 suites - 231 3h 16m 19s ⏱️ - 1h 12m 4s
893 tests - 15 892 ✅ - 15 1 💤 ±0 0 ❌ ±0
10 009 runs - 532 9 995 ✅ - 532 14 💤 ±0 0 ❌ ±0

Results for commit f28ca7d. ± Comparison against base commit cbf4f6f.

This pull request removes 15 tests.

org.apache.spark.shuffle.RssShuffleManagerTest ‑ testCreateShuffleManagerServer
org.apache.spark.shuffle.RssShuffleManagerTest ‑ testGetDataDistributionType
org.apache.spark.shuffle.RssShuffleManagerTest ‑ testRssShuffleManagerInterface
org.apache.spark.shuffle.RssShuffleManagerTest ‑ testRssShuffleManagerRegisterShuffle{int}[1]
org.apache.spark.shuffle.RssShuffleManagerTest ‑ testRssShuffleManagerRegisterShuffle{int}[2]
org.apache.spark.shuffle.RssShuffleManagerTest ‑ testRssShuffleManagerRegisterShuffle{int}[3]
org.apache.spark.shuffle.RssSpark3ShuffleUtilsTest ‑ testCreateFetchFailedException
org.apache.spark.shuffle.RssSpark3ShuffleUtilsTest ‑ testIsStageResubmitSupported
org.apache.spark.shuffle.SparkVersionUtilsTest ‑ testSpark3Version
org.apache.spark.shuffle.writer.RssShuffleWriterTest ‑ dataConsistencyWhenSpillTriggeredTest
…

♻️ This comment has been updated with latest results.

zuston

Task write fail retry is not related with stage recomputing, right?

dingshun3016 · 2024-03-15T02:35:17Z

Task write fail retry is not related with stage recomputing, right?

Yes, they can run independently

zuston

Please also fix the spark2

rickyma · 2024-03-15T03:51:08Z

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

@@ -137,6 +137,8 @@ public class RssShuffleManager extends RssShuffleManagerBase {
  private Map<Integer, ShuffleHandleInfo> shuffleIdToShuffleHandleInfo;
  /** Whether to enable the dynamic shuffleServer function rewrite and reread functions */
  private boolean rssResubmitStage;
+
+  private boolean taskFailRetry;


taskFailRetry -> taskReattemptOnFailures

Is is better?

I don't think this is only valid for task write failure, in the future, it could be applied on the not-enough-capacity disk servers

taskFailureRetryEnabled

jerqi · 2024-03-15T06:37:24Z

Could you add a ut for this case?

zuston

LGTM overall. But after reviewing this part code again, I think we could extra this logic into a general method and put this into RssShuffleManagerBase, which could be shared by spark2 + 3.

Could you help optimize this?

dingshun3016 · 2024-03-19T09:20:22Z

LGTM overall. But after reviewing this part code again, I think we could extra this logic into a general method and put this into RssShuffleManagerBase, which could be shared by spark2 + 3.

Could you help optimize this?

I will do it

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

zuston · 2024-04-01T07:01:40Z

common/src/main/java/org/apache/uniffle/common/config/RssClientConf.java

-  public static final ConfigOption<Boolean> RSS_TASK_FAILED_RETRY_ENABLED =
-      ConfigOptions.key("rss.task.failed.retry.enabled")
+  public static final ConfigOption<Boolean> RSS_TASK_FAILURE_RETRY_ENABLED =
+      ConfigOptions.key("rss.task.failure.retry.enabled")


rss.task.failure.retry.enabled -> rss.client.blockSendFailureRetry.enabled

zuston · 2024-04-01T07:02:17Z

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

@@ -112,7 +112,7 @@ public class RssShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
  private final Set<Long> blockIds = Sets.newConcurrentHashSet();
  private TaskContext taskContext;
  private SparkConf sparkConf;
-  private boolean taskFailRetry;
+  private boolean taskFailureRetryEnabled;


taskFailureRetryEnabled -> blockSendFailureRetryEnabled

zuston · 2024-04-01T07:03:00Z

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

@@ -137,6 +137,8 @@ public class RssShuffleManager extends RssShuffleManagerBase {
  private Map<Integer, ShuffleHandleInfo> shuffleIdToShuffleHandleInfo;
  /** Whether to enable the dynamic shuffleServer function rewrite and reread functions */
  private boolean rssResubmitStage;
+
+  private boolean taskFailureRetryEnabled;


taskFailureRetryEnabled -> taskBlockSendFailureRetry

zuston · 2024-04-01T07:03:31Z

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

@@ -239,10 +241,15 @@ public RssShuffleManager(SparkConf conf, boolean isDriver) {
    this.rssResubmitStage =
        rssConf.getBoolean(RssClientConfig.RSS_RESUBMIT_STAGE, false)
            && RssSparkShuffleUtils.isStageResubmitSupported();
+    this.taskFailureRetryEnabled =


this.taskBlockSendFailureRetry = rssConf.getBoolean(RssClientConf.RSS_TASK_FAILURE_RETRY_ENABLED);

zuston · 2024-04-01T07:04:55Z

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

+        sparkConf.getBoolean(
+            RssSparkConfig.SPARK_RSS_CONFIG_PREFIX
+                + RssClientConf.RSS_TASK_FAILURE_RETRY_ENABLED.key(),
+            RssClientConf.RSS_TASK_FAILURE_RETRY_ENABLED.defaultValue());


And here you can introduce the new var to indicate whether we need to start shuffle manager rpc service.Like this:

this.shuffleManagerRpcServiceEnabled = taskBlockSendFailureRetry || rssResubmitStage

zuston · 2024-04-01T07:05:12Z

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

    if (isDriver) {
      heartBeatScheduledExecutorService =
          ThreadUtils.getDaemonSingleThreadScheduledExecutor("rss-heartbeat");
-      if (rssResubmitStage) {
+      if (rssResubmitStage || taskFailureRetryEnabled) {


replace with if (shuffleManagerRpcServiceEnabled)

zuston · 2024-04-01T07:05:22Z

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

@@ -471,7 +478,7 @@ public <K, V, C> ShuffleHandle registerShuffle(

    shuffleIdToPartitionNum.putIfAbsent(shuffleId, dependency.partitioner().numPartitions());
    shuffleIdToNumMapTasks.putIfAbsent(shuffleId, dependency.rdd().partitions().length);
-    if (rssResubmitStage) {
+    if (rssResubmitStage || taskFailureRetryEnabled) {


zuston · 2024-04-01T07:05:29Z

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

-          new ShuffleHandleInfo(
-              shuffleId, rssHandle.getPartitionToServers(), rssHandle.getRemoteStorage());
-    }
+    ShuffleHandleInfo shuffleHandleInfo =


revert this part

zuston · 2024-04-01T07:05:49Z

client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/RssShuffleManagerBase.java

@@ -488,4 +490,20 @@ protected static RemoteStorageInfo getDefaultRemoteStorageInfo(SparkConf sparkCo
    return new RemoteStorageInfo(
        sparkConf.get(RssSparkConfig.RSS_REMOTE_STORAGE_PATH.key(), ""), confItems);
  }
+
+  protected static ShuffleHandleInfo getShuffleHandleInfo(


Please revert this

zuston · 2024-04-01T07:54:25Z

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

@@ -222,11 +226,17 @@ public RssShuffleManager(SparkConf sparkConf, boolean isDriver) {
    this.rssResubmitStage =
        rssConf.getBoolean(RssClientConfig.RSS_RESUBMIT_STAGE, false)
            && RssSparkShuffleUtils.isStageResubmitSupported();
+    this.taskBlockFailureRetryEnabled =


`this.taskBlockSendFailureRetry = rssConf.getBoolean(RssClientConf.RSS_TASK_FAILURE_RETRY_ENABLED);

`

zuston · 2024-04-01T07:54:54Z

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

@@ -137,6 +137,10 @@ public class RssShuffleManager extends RssShuffleManagerBase {
  private Map<Integer, ShuffleHandleInfo> shuffleIdToShuffleHandleInfo;
  /** Whether to enable the dynamic shuffleServer function rewrite and reread functions */
  private boolean rssResubmitStage;
+
+  private boolean taskBlockSendFailureRetry;


taskBlockSendFailureRetryEnabled

zuston · 2024-04-01T07:55:06Z

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

@@ -239,10 +243,16 @@ public RssShuffleManager(SparkConf conf, boolean isDriver) {
    this.rssResubmitStage =
        rssConf.getBoolean(RssClientConfig.RSS_RESUBMIT_STAGE, false)
            && RssSparkShuffleUtils.isStageResubmitSupported();
+    this.taskBlockSendFailureRetry =


this.taskBlockSendFailureRetryEnabled = rssConf.getBoolean(RssClientConf.RSS_TASK_FAILURE_RETRY_ENABLED);

zuston · 2024-04-01T07:55:53Z

common/src/main/java/org/apache/uniffle/common/config/RssClientConf.java

@@ -203,10 +203,10 @@ public class RssClientConf {
              "This option is only valid when the remote storage path is specified. If ture, "
                  + "the remote storage conf will use the client side hadoop configuration loaded from the classpath.");

-  public static final ConfigOption<Boolean> RSS_TASK_FAILED_RETRY_ENABLED =
-      ConfigOptions.key("rss.task.failed.retry.enabled")
+  public static final ConfigOption<Boolean> RSS_CLIENT_BLOCK_SEND_FAILURE_RETRY_ENABLED =


rss.client.blockSendFailureRetry.enabled

rss.client.blockSendFailureRetry.enabled -> rss.client.blockSendFailureRetryEnabled.

zuston · 2024-04-01T07:56:47Z

And the spotless failure could be solved by the mvn spotless:apply -Pspark3 -Pspark2 -Ptez -Pmr -Phadoop2.8 @dingshun3016

zuston

LGTM. Thanks for your contribution @dingshun3016

[apache#1373][followup] fix(client):Fix task fail retry parameter not…

81e129c

… work

zuston reviewed Mar 15, 2024

View reviewed changes

fix code style

c0a09ba

zuston reviewed Mar 15, 2024

View reviewed changes

rickyma reviewed Mar 15, 2024

View reviewed changes

fix the spark2 and rename

d795246

zuston reviewed Mar 18, 2024

View reviewed changes

zuston changed the title ~~[#1373][followup] fix(client):Fix task fail retry parameter not work~~ [#1373][FOLLOWUP] fix(client): partition data reassign is invalid Apr 1, 2024

zuston changed the title ~~[#1373][FOLLOWUP] fix(client): partition data reassign is invalid~~ [#1373][FOLLOWUP] fix(spark): shuffle manager rpc service invalid when partition data reassign is enabled Apr 1, 2024

zuston mentioned this pull request Apr 1, 2024

Support client partition data reassign #1608

Open

9 tasks

zuston reviewed Apr 1, 2024

View reviewed changes

refact code structure

f28ca7d

dingshun3016 force-pushed the followup-1373-2 branch from fa3ca15 to f28ca7d Compare April 1, 2024 07:48

zuston reviewed Apr 1, 2024

View reviewed changes

shun01.ding added 2 commits April 1, 2024 16:17

fix config

c79ef03

fix code style

f488c59

zuston approved these changes Apr 1, 2024

View reviewed changes

zuston merged commit 05ff6db into apache:master Apr 1, 2024
36 of 37 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[#1373][FOLLOWUP] fix(spark): shuffle manager rpc service invalid when partition data reassign is enabled #1583

[#1373][FOLLOWUP] fix(spark): shuffle manager rpc service invalid when partition data reassign is enabled #1583

dingshun3016 commented Mar 14, 2024 •

edited

codecov-commenter commented Mar 14, 2024 •

edited

github-actions bot commented Mar 14, 2024 •

edited

zuston left a comment

dingshun3016 commented Mar 15, 2024

zuston left a comment

rickyma Mar 15, 2024

zuston Mar 15, 2024

jerqi Mar 15, 2024

jerqi commented Mar 15, 2024

zuston left a comment

dingshun3016 commented Mar 19, 2024

zuston Apr 1, 2024

zuston Apr 1, 2024

zuston Apr 1, 2024

zuston Apr 1, 2024

zuston Apr 1, 2024

zuston Apr 1, 2024

zuston Apr 1, 2024

zuston Apr 1, 2024

zuston Apr 1, 2024

zuston Apr 1, 2024

zuston Apr 1, 2024

zuston Apr 1, 2024

zuston Apr 1, 2024

jerqi Apr 1, 2024

zuston commented Apr 1, 2024

zuston left a comment

[#1373][FOLLOWUP] fix(spark): shuffle manager rpc service invalid when partition data reassign is enabled #1583

[#1373][FOLLOWUP] fix(spark): shuffle manager rpc service invalid when partition data reassign is enabled #1583

Conversation

dingshun3016 commented Mar 14, 2024 • edited

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?

codecov-commenter commented Mar 14, 2024 • edited

Codecov Report

github-actions bot commented Mar 14, 2024 • edited

Test Results

zuston left a comment

Choose a reason for hiding this comment

dingshun3016 commented Mar 15, 2024

zuston left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerqi commented Mar 15, 2024

zuston left a comment

Choose a reason for hiding this comment

dingshun3016 commented Mar 19, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zuston commented Apr 1, 2024

zuston left a comment

Choose a reason for hiding this comment

dingshun3016 commented Mar 14, 2024 •

edited

codecov-commenter commented Mar 14, 2024 •

edited

github-actions bot commented Mar 14, 2024 •

edited