Put sparkConf as extra properties while client request accessCluster

Author: maobaolong

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

@@ -17,6 +17,8 @@
 
 package org.apache.spark.shuffle;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Set;
 
 import scala.Tuple2;
@@ -512,4 +514,15 @@ public static RssConf toRssConf(SparkConf sparkConf) {
     }
     return rssConf;
   }
+
+  public static Map<String, String> sparkConfToMap(SparkConf sparkConf) {
+    Map<String, String> map = new HashMap<>();
+
+    for (Tuple2<String, String> tuple : sparkConf.getAll()) {
+      String key = tuple._1;
+      map.put(key, tuple._2);
+    }
+
+    return map;
+  }
 }

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

@@ -34,8 +34,6 @@
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
-import scala.Tuple2;
-
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
@@ -1064,7 +1062,7 @@ protected void registerShuffleServers(
     }
     LOG.info("Start to register shuffleId {}", shuffleId);
     long start = System.currentTimeMillis();
-    Map<String, String> sparkConfMap = sparkConfToMap(getSparkConf());
+    Map<String, String> sparkConfMap = RssSparkConfig.sparkConfToMap(getSparkConf());
     serverToPartitionRanges.entrySet().stream()
         .forEach(
             entry -> {
@@ -1095,7 +1093,7 @@ protected void registerShuffleServers(
     }
     LOG.info("Start to register shuffleId[{}]", shuffleId);
     long start = System.currentTimeMillis();
-    Map<String, String> sparkConfMap = sparkConfToMap(getSparkConf());
+    Map<String, String> sparkConfMap = RssSparkConfig.sparkConfToMap(getSparkConf());
     Set<Map.Entry<ShuffleServerInfo, List<PartitionRange>>> entries =
         serverToPartitionRanges.entrySet();
     entries.stream()
@@ -1141,15 +1139,4 @@ public boolean isRssStageRetryForFetchFailureEnabled() {
   public SparkConf getSparkConf() {
     return sparkConf;
   }
-
-  public Map<String, String> sparkConfToMap(SparkConf sparkConf) {
-    Map<String, String> map = new HashMap<>();
-
-    for (Tuple2<String, String> tuple : sparkConf.getAll()) {
-      String key = tuple._1;
-      map.put(key, tuple._2);
-    }
-
-    return map;
-  }
 }

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

@@ -123,6 +123,7 @@ private boolean tryAccessCluster() {
     Map<String, String> extraProperties = Maps.newHashMap();
     extraProperties.put(
         ACCESS_INFO_REQUIRED_SHUFFLE_NODES_NUM, String.valueOf(assignmentShuffleNodesNum));
+    extraProperties.putAll(RssSparkConfig.sparkConfToMap(sparkConf));
 
     Set<String> assignmentTags = RssSparkShuffleUtils.getAssignmentTags(sparkConf);
     try {