File tree Expand file tree Collapse file tree 3 files changed +16
-2
lines changed
common/src/main/java/org/apache
spark3/src/main/java/org/apache/spark/shuffle Expand file tree Collapse file tree 3 files changed +16
-2
lines changed Original file line number Diff line number Diff line change 1717
1818package org .apache .spark .shuffle ;
1919
20+ import java .util .HashMap ;
21+ import java .util .Map ;
2022import java .util .Set ;
2123
2224import scala .Tuple2 ;
@@ -512,4 +514,15 @@ public static RssConf toRssConf(SparkConf sparkConf) {
512514 }
513515 return rssConf ;
514516 }
517+
518+ public static Map <String , String > sparkConfToMap (SparkConf sparkConf ) {
519+ Map <String , String > map = new HashMap <>();
520+
521+ for (Tuple2 <String , String > tuple : sparkConf .getAll ()) {
522+ String key = tuple ._1 ;
523+ map .put (key , tuple ._2 );
524+ }
525+
526+ return map ;
527+ }
515528}
Original file line number Diff line number Diff line change @@ -1064,7 +1064,7 @@ protected void registerShuffleServers(
10641064 }
10651065 LOG .info ("Start to register shuffleId {}" , shuffleId );
10661066 long start = System .currentTimeMillis ();
1067- Map <String , String > sparkConfMap = sparkConfToMap (getSparkConf ());
1067+ Map <String , String > sparkConfMap = RssSparkConfig . sparkConfToMap (getSparkConf ());
10681068 serverToPartitionRanges .entrySet ().stream ()
10691069 .forEach (
10701070 entry -> {
@@ -1095,7 +1095,7 @@ protected void registerShuffleServers(
10951095 }
10961096 LOG .info ("Start to register shuffleId[{}]" , shuffleId );
10971097 long start = System .currentTimeMillis ();
1098- Map <String , String > sparkConfMap = sparkConfToMap (getSparkConf ());
1098+ Map <String , String > sparkConfMap = RssSparkConfig . sparkConfToMap (getSparkConf ());
10991099 Set <Map .Entry <ShuffleServerInfo , List <PartitionRange >>> entries =
11001100 serverToPartitionRanges .entrySet ();
11011101 entries .stream ()
Original file line number Diff line number Diff line change @@ -123,6 +123,7 @@ private boolean tryAccessCluster() {
123123 Map <String , String > extraProperties = Maps .newHashMap ();
124124 extraProperties .put (
125125 ACCESS_INFO_REQUIRED_SHUFFLE_NODES_NUM , String .valueOf (assignmentShuffleNodesNum ));
126+ extraProperties .putAll (RssSparkConfig .sparkConfToMap (sparkConf ));
126127
127128 Set <String > assignmentTags = RssSparkShuffleUtils .getAssignmentTags (sparkConf );
128129 try {
You can’t perform that action at this time.
0 commit comments