Merge remote-tracking branch 'github/master' into fixMetricSystem

Author: kuszz

README.md

@@ -122,6 +122,8 @@ Build with dashboard
 
     ./mvnw -DskipTests clean package -Pdashboard
 
+> note: currently Uniffle build the project against Java 8. If you want to compile it against other Java versions, you can build the code with `-Dmaven.compiler.release=${release-version}`.
+
 To package the Uniffle, run:
 
     ./build_distribution.sh

client-mr/core/src/main/java/org/apache/hadoop/mapreduce/RssMRConfig.java

@@ -64,7 +64,7 @@ public class RssMRConfig {
   public static final String RSS_DATA_TRANSFER_POOL_SIZE =
       MR_CONFIG_PREFIX + RssClientConfig.RSS_DATA_TRANSFER_POOL_SIZE;
   public static final int RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE =
-      RssClientConfig.RSS_DATA_TRANFER_POOL_SIZE_DEFAULT_VALUE;
+      RssClientConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE;
   public static final String RSS_DATA_COMMIT_POOL_SIZE =
       MR_CONFIG_PREFIX + RssClientConfig.RSS_DATA_COMMIT_POOL_SIZE;
   public static final int RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE =

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

@@ -296,7 +296,7 @@ public class RssSparkConfig {
       createIntegerBuilder(
               new ConfigBuilder(
                   SPARK_RSS_CONFIG_PREFIX + RssClientConfig.RSS_DATA_TRANSFER_POOL_SIZE))
-          .createWithDefault(RssClientConfig.RSS_DATA_TRANFER_POOL_SIZE_DEFAULT_VALUE);
+          .createWithDefault(RssClientConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE);
 
   public static final ConfigEntry<Integer> RSS_DATA_COMMIT_POOL_SIZE =
       createIntegerBuilder(

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

@@ -183,7 +183,7 @@ public RssShuffleManager(SparkConf conf, boolean isDriver) {
     taskToSuccessBlockIds = JavaUtils.newConcurrentMap();
     taskToFailedBlockSendTracker = JavaUtils.newConcurrentMap();
 
-    this.rssStageRetryEnabled = rssConf.get(RssClientConf.RSS_CLIENT_REASSIGN_ENABLED);
+    this.rssStageRetryEnabled = rssConf.get(RssSparkConfig.RSS_RESUBMIT_STAGE_ENABLED);
     this.partitionReassignEnabled = rssConf.get(RssClientConf.RSS_CLIENT_REASSIGN_ENABLED);
 
     // stage retry for write/fetch failure

client-tez/src/main/java/org/apache/tez/common/RssTezConfig.java

@@ -63,7 +63,7 @@ public class RssTezConfig {
   public static final String RSS_DATA_TRANSFER_POOL_SIZE =
       TEZ_RSS_CONFIG_PREFIX + RssClientConfig.RSS_DATA_TRANSFER_POOL_SIZE;
   public static final int RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE =
-      RssClientConfig.RSS_DATA_TRANFER_POOL_SIZE_DEFAULT_VALUE;
+      RssClientConfig.RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE;
   public static final String RSS_DATA_COMMIT_POOL_SIZE =
       TEZ_RSS_CONFIG_PREFIX + RssClientConfig.RSS_DATA_COMMIT_POOL_SIZE;
   public static final int RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE =

client-tez/src/main/java/org/apache/tez/dag/app/RssDAGAppMaster.java

@@ -457,7 +457,7 @@ public static void main(String[] args) {
               TEZ_AM_NODE_UNHEALTHY_RESCHEDULE_TASKS,
               TEZ_AM_NODE_UNHEALTHY_RESCHEDULE_TASKS_DEFAULT)) {
         LOG.info(
-            "When rss.avoid.recompute.succeeded.task is enable, "
+            "When rss.avoid.recompute.succeeded.task is enabled, "
                 + "we can not rescheduler succeeded task on unhealthy node");
         conf.setBoolean(TEZ_AM_NODE_UNHEALTHY_RESCHEDULE_TASKS, false);
       }

client/src/main/java/org/apache/uniffle/client/util/RssClientConfig.java

@@ -35,7 +35,7 @@ public class RssClientConfig {
   public static final String RSS_DATA_REPLICA_SKIP_ENABLED = "rss.data.replica.skip.enabled";
   public static final boolean RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE = true;
   public static final String RSS_DATA_TRANSFER_POOL_SIZE = "rss.client.data.transfer.pool.size";
-  public static final int RSS_DATA_TRANFER_POOL_SIZE_DEFAULT_VALUE =
+  public static final int RSS_DATA_TRANSFER_POOL_SIZE_DEFAULT_VALUE =
       Runtime.getRuntime().availableProcessors();
   public static final String RSS_DATA_COMMIT_POOL_SIZE = "rss.client.data.commit.pool.size";
   public static final int RSS_DATA_COMMIT_POOL_SIZE_DEFAULT_VALUE = -1;

common/src/main/java/org/apache/uniffle/common/ShuffleIndexResult.java

@@ -28,6 +28,7 @@
 public class ShuffleIndexResult {
   private final ManagedBuffer buffer;
   private long dataFileLen;
+  private String dataFileName;
 
   public ShuffleIndexResult() {
     this(ByteBuffer.wrap(new byte[0]), -1);
@@ -43,9 +44,10 @@ public ShuffleIndexResult(ByteBuffer data, long dataFileLen) {
     this.dataFileLen = dataFileLen;
   }
 
-  public ShuffleIndexResult(ManagedBuffer buffer, long dataFileLen) {
+  public ShuffleIndexResult(ManagedBuffer buffer, long dataFileLen, String dataFileName) {
     this.buffer = buffer;
     this.dataFileLen = dataFileLen;
+    this.dataFileName = dataFileName;
   }
 
   public byte[] getData() {
@@ -79,4 +81,8 @@ public void release() {
   public ManagedBuffer getManagedBuffer() {
     return buffer;
   }
+
+  public String getDataFileName() {
+    return dataFileName;
+  }
 }

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

@@ -168,7 +168,7 @@ public Iterable<FallbackKey> fallbackKeys() {
   /**
    * Returns the class of value.
    *
-   * @return The option' value class
+   * @return The option value class
    */
   public Class<?> getClazz() {
     return clazz;

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

@@ -208,7 +208,7 @@ public ConfigOption<T> noDefaultValue() {
    * @param <E> list element type of the option
    */
   public static class ListConfigOptionBuilder<E> {
-    private static final String LIST_SPILTTER = ",";
+    private static final String LIST_SPLITTER = ",";
 
     private final String key;
     private final Class<E> clazz;
@@ -230,7 +230,7 @@ public ListConfigOptionBuilder(
               if (StringUtils.isEmpty(trimmedVal)) {
                 return Collections.emptyList();
               }
-              return Arrays.stream(trimmedVal.split(LIST_SPILTTER))
+              return Arrays.stream(trimmedVal.split(LIST_SPLITTER))
                   .map(atomicConverter::apply)
                   .collect(Collectors.toList());
             }