Enable spotbugs and fix high priority bugs (#38)

### What changes were proposed in this pull request?

1. Enable (previously experimental) spotbugs check in CI.
2. Set spotbugs to only check high priority bugs. 
    Because currently there are too many medium (default) priority bugs,
    and many of them are difficult to fix.
3. Fix bugs reported by spotbugs.

### Why are the changes needed?

To improve code quality.

### Does this PR introduce _any_ user-facing change?

One notable change is we are comparing content instead of reference of the array in `ShufflePartitionedBlock#equals()`, see: a994cff. We can revert it if it's not desired.

Another is `toString()` methods will call `Arrays.toString()` instead of printing array reference as string now.

### How was this patch tested?

1. `mvn test-compile spotbugs:check -Pspark3`
2. `mvn test-compile spotbugs:check -Pspark2`
3. `mvn test-compile spotbugs:check -Pmr`

Also by CI: https://github.com/kaijchen/incubator-uniffle/actions/runs/2634089750

.github/workflows/build.yml

@@ -37,7 +37,6 @@ jobs:
     uses: ./.github/workflows/sequential.yml
     with:
       maven-args: test-compile spotbugs:check
-      experimental: true
 
   build:
     uses: ./.github/workflows/parallel.yml

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

@@ -61,7 +61,7 @@ public class RssMRConfig {
       RssClientConfig.RSS_CLIENT_DEFAULT_SEND_NUM;
   public static final String RSS_CLIENT_SEND_THRESHOLD = MR_RSS_CONFIG_PREFIX + "rss.client.send.threshold";
   public static final double RSS_CLIENT_DEFAULT_SEND_THRESHOLD = 0.2f;
-  public static boolean RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE =
+  public static final boolean RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE =
       RssClientConfig.RSS_DATA_REPLICA_SKIP_ENABLED_DEFAULT_VALUE;
   public static final String RSS_HEARTBEAT_INTERVAL =
       MR_RSS_CONFIG_PREFIX + RssClientConfig.RSS_HEARTBEAT_INTERVAL;
@@ -109,26 +109,26 @@ public class RssMRConfig {
       MR_RSS_CONFIG_PREFIX + RssClientConfig.RSS_INDEX_READ_LIMIT;
   public static final int RSS_INDEX_READ_LIMIT_DEFAULT_VALUE =
       RssClientConfig.RSS_INDEX_READ_LIMIT_DEFAULT_VALUE;
-  public static String RSS_CLIENT_READ_BUFFER_SIZE =
+  public static final String RSS_CLIENT_READ_BUFFER_SIZE =
       MR_RSS_CONFIG_PREFIX + RssClientConfig.RSS_CLIENT_READ_BUFFER_SIZE;
 
   // When the size of read buffer reaches the half of JVM region (i.e., 32m),
   // it will incur humongous allocation, so we set it to 14m.
-  public static String RSS_CLIENT_READ_BUFFER_SIZE_DEFAULT_VALUE =
+  public static final String RSS_CLIENT_READ_BUFFER_SIZE_DEFAULT_VALUE =
       RssClientConfig.RSS_CLIENT_READ_BUFFER_SIZE_DEFAULT_VALUE;
 
-  public static String RSS_DYNAMIC_CLIENT_CONF_ENABLED =
+  public static final String RSS_DYNAMIC_CLIENT_CONF_ENABLED =
       MR_RSS_CONFIG_PREFIX + RssClientConfig.RSS_DYNAMIC_CLIENT_CONF_ENABLED;
-  public static boolean RSS_DYNAMIC_CLIENT_CONF_ENABLED_DEFAULT_VALUE =
+  public static final boolean RSS_DYNAMIC_CLIENT_CONF_ENABLED_DEFAULT_VALUE =
       RssClientConfig.RSS_DYNAMIC_CLIENT_CONF_ENABLED_DEFAULT_VALUE;
-  public static String RSS_ACCESS_TIMEOUT_MS = MR_RSS_CONFIG_PREFIX + RssClientConfig.RSS_ACCESS_TIMEOUT_MS;
-  public static int RSS_ACCESS_TIMEOUT_MS_DEFAULT_VALUE = RssClientConfig.RSS_ACCESS_TIMEOUT_MS_DEFAULT_VALUE;
+  public static final String RSS_ACCESS_TIMEOUT_MS = MR_RSS_CONFIG_PREFIX + RssClientConfig.RSS_ACCESS_TIMEOUT_MS;
+  public static final int RSS_ACCESS_TIMEOUT_MS_DEFAULT_VALUE = RssClientConfig.RSS_ACCESS_TIMEOUT_MS_DEFAULT_VALUE;
 
   public static final String RSS_CLIENT_ASSIGNMENT_TAGS =
           MR_RSS_CONFIG_PREFIX + RssClientConfig.RSS_CLIENT_ASSIGNMENT_TAGS;
 
-  public static String RSS_CONF_FILE = "rss_conf.xml";
+  public static final String RSS_CONF_FILE = "rss_conf.xml";
 
-  public static Set<String> RSS_MANDATORY_CLUSTER_CONF = Sets.newHashSet(
+  public static final Set<String> RSS_MANDATORY_CLUSTER_CONF = Sets.newHashSet(
       RSS_STORAGE_TYPE, RSS_REMOTE_STORAGE_PATH);
 }

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

@@ -17,6 +17,8 @@
 
 package org.apache.uniffle.common;
 
+import java.util.Objects;
+
 public class BufferSegment {
 
   private long blockId;
@@ -48,6 +50,11 @@ public boolean equals(Object obj) {
     return false;
   }
 
+  @Override
+  public int hashCode() {
+    return Objects.hash(blockId, offset, length, uncompressLength, crc, taskAttemptId);
+  }
+
   @Override
   public String toString() {
     return "BufferSegment{blockId[" + blockId + "], taskAttemptId[" + taskAttemptId

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

@@ -17,6 +17,9 @@
 
 package org.apache.uniffle.common;
 
+import java.util.Arrays;
+import java.util.Objects;
+
 public class ShufflePartitionedBlock {
 
   private int length;
@@ -59,7 +62,12 @@ public boolean equals(Object o) {
     return length == that.length
         && crc == that.crc
         && blockId == that.blockId
-        && data.equals(that.data);
+        && Arrays.equals(data, that.data);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(length, crc, blockId, Arrays.hashCode(data));
   }
 
   public int getLength() {

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

@@ -17,6 +17,8 @@
 
 package org.apache.uniffle.common;
 
+import java.util.Arrays;
+
 public class ShufflePartitionedData {
 
   private int partitionId;
@@ -29,7 +31,7 @@ public ShufflePartitionedData(int partitionId, ShufflePartitionedBlock[] blockLi
 
   @Override
   public String toString() {
-    return "ShufflePartitionedData{partitionId=" + partitionId + ", blockList=" + blockList + '}';
+    return "ShufflePartitionedData{partitionId=" + partitionId + ", blockList=" + Arrays.toString(blockList) + '}';
   }
 
   public int getPartitionId() {

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

@@ -162,6 +162,11 @@ public boolean equals(Object o) {
     }
   }
 
+  @Override
+  public int hashCode() {
+    return Objects.hash(key, defaultValue);
+  }
+
   @Override
   public String toString() {
     return String.format("Key: '%s' , default: %s", key, defaultValue);

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

@@ -177,17 +177,17 @@ && isFinal(modifiers) && field.getType().isAssignableFrom(ConfigOption.class)) {
     return configOptionList;
   }
 
-  public static Function<Long, Boolean> positiveLongValidator = value -> value > 0;
+  public static final Function<Long, Boolean> POSITIVE_LONG_VALIDATOR = value -> value > 0;
 
-  public static Function<Long, Boolean> non_negativeLongValidator = value -> value >= 0;
+  public static final Function<Long, Boolean> NON_NEGATIVE_LONG_VALIDATOR = value -> value >= 0;
 
-  public static Function<Long, Boolean> positiveIntegerValidator =
+  public static final Function<Long, Boolean> POSITIVE_INTEGER_VALIDATOR =
       value -> value > 0L && value <= Integer.MAX_VALUE;
 
-  public static Function<Integer, Boolean> positiveIntegerValidator2 =
+  public static final Function<Integer, Boolean> POSITIVE_INTEGER_VALIDATOR_2 =
       value -> value > 0;
 
-  public static Function<Double, Boolean> percentageDoubleValidator =
+  public static final Function<Double, Boolean> PERCENTAGE_DOUBLE_VALIDATOR =
       (Function<Double, Boolean>) value -> Double.compare(value, 100.0) <= 0 && Double.compare(value, 0.0) >= 0;
 
 }

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

@@ -116,7 +116,7 @@ public class RssBaseConf extends RssConf {
   public static final ConfigOption<Long> RPC_MESSAGE_MAX_SIZE = ConfigOptions
       .key("rss.rpc.message.max.size")
       .longType()
-      .checkValue(ConfigUtils.positiveIntegerValidator,
+      .checkValue(ConfigUtils.POSITIVE_INTEGER_VALIDATOR,
         "The value must be positive integer")
       .defaultValue(1024L * 1024L * 1024L)
       .withDescription("Max size of rpc message (byte)");

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

@@ -27,12 +27,12 @@
 
 import org.apache.uniffle.common.util.UnitConverter;
 
-public class RssConf {
+public class RssConf implements Cloneable {
 
   /**
    * Stores the concrete key/value pairs of this configuration object.
    */
-  private final ConcurrentHashMap<String, Object> settings;
+  private ConcurrentHashMap<String, Object> settings;
 
   /**
    * Creates a new empty configuration.
@@ -532,9 +532,9 @@ public String getValue(ConfigOption<?> configOption) {
   // --------------------------------------------------------------------------------------------
 
   @Override
-  public RssConf clone() {
-    RssConf config = new RssConf();
-    config.addAll(this);
+  public RssConf clone() throws CloneNotSupportedException {
+    RssConf config = (RssConf) super.clone();
+    config.settings = new ConcurrentHashMap<>(settings);
     return config;
   }
 

coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorConf.java

@@ -71,7 +71,7 @@ public class CoordinatorConf extends RssBaseConf {
   public static final ConfigOption<Integer> COORDINATOR_ACCESS_CANDIDATES_UPDATE_INTERVAL_SEC = ConfigOptions
       .key("rss.coordinator.access.candidates.updateIntervalSec")
       .intType()
-      .checkValue(ConfigUtils.positiveIntegerValidator2, "access candidates update interval must be positive")
+      .checkValue(ConfigUtils.POSITIVE_INTEGER_VALIDATOR_2, "access candidates update interval must be positive")
       .defaultValue(120)
       .withDescription("Accessed candidates update interval in seconds");
   public static final ConfigOption<String> COORDINATOR_ACCESS_CANDIDATES_PATH = ConfigOptions
@@ -82,14 +82,14 @@ public class CoordinatorConf extends RssBaseConf {
   public static final ConfigOption<Double> COORDINATOR_ACCESS_LOADCHECKER_MEMORY_PERCENTAGE = ConfigOptions
       .key("rss.coordinator.access.loadChecker.memory.percentage")
       .doubleType()
-      .checkValue(ConfigUtils.percentageDoubleValidator,
+      .checkValue(ConfigUtils.PERCENTAGE_DOUBLE_VALIDATOR,
           "The recovery usage percentage must be between 0.0 and 100.0")
       .defaultValue(15.0)
       .withDescription("The minimal percentage of available memory percentage of a server");
   public static final ConfigOption<Integer> COORDINATOR_ACCESS_LOADCHECKER_SERVER_NUM_THRESHOLD = ConfigOptions
       .key("rss.coordinator.access.loadChecker.serverNum.threshold")
       .intType()
-      .checkValue(ConfigUtils.positiveIntegerValidator2, "load checker serverNum threshold must be positive")
+      .checkValue(ConfigUtils.POSITIVE_INTEGER_VALIDATOR_2, "load checker serverNum threshold must be positive")
       .noDefaultValue()
       .withDescription("The minimal required number of healthy shuffle servers when being accessed by client");
   public static final ConfigOption<Boolean> COORDINATOR_DYNAMIC_CLIENT_CONF_ENABLED = ConfigOptions
@@ -110,7 +110,7 @@ public class CoordinatorConf extends RssBaseConf {
   public static final ConfigOption<Integer> COORDINATOR_DYNAMIC_CLIENT_CONF_UPDATE_INTERVAL_SEC = ConfigOptions
       .key("rss.coordinator.dynamicClientConf.updateIntervalSec")
       .intType()
-      .checkValue(ConfigUtils.positiveIntegerValidator2, "dynamic client conf update interval in seconds")
+      .checkValue(ConfigUtils.POSITIVE_INTEGER_VALIDATOR_2, "dynamic client conf update interval in seconds")
       .defaultValue(120)
       .withDescription("Accessed candidates update interval in seconds");
   public static final ConfigOption<String> COORDINATOR_REMOTE_STORAGE_CLUSTER_CONF = ConfigOptions

coordinator/src/main/java/org/apache/uniffle/coordinator/SimpleClusterManager.java

@@ -22,6 +22,7 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -127,7 +128,7 @@ private void updateExcludeNodes(String path) {
 
   private void parseExcludeNodesFile(DataInputStream fsDataInputStream) throws IOException {
     Set<String> nodes = Sets.newConcurrentHashSet();
-    try (BufferedReader br = new BufferedReader(new InputStreamReader(fsDataInputStream))) {
+    try (BufferedReader br = new BufferedReader(new InputStreamReader(fsDataInputStream, StandardCharsets.UTF_8))) {
       String line;
       while ((line = br.readLine()) != null) {
         if (!StringUtils.isEmpty(line)) {

internal-client/src/main/java/org/apache/uniffle/client/factory/CoordinatorClientFactory.java

@@ -17,6 +17,7 @@
 
 package org.apache.uniffle.client.factory;
 
+import java.util.Arrays;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -58,7 +59,7 @@ public List<CoordinatorClient> createCoordinatorClient(String coordinators) {
     for (String coordinator: coordinatorList) {
       String[] ipPort = coordinator.trim().split(":");
       if (ipPort.length != 2) {
-        String msg = "Invalid coordinator format " + ipPort;
+        String msg = "Invalid coordinator format " + Arrays.toString(ipPort);
         LOG.error(msg);
         throw new RuntimeException(msg);
       }

pom.xml

@@ -820,6 +820,7 @@
           </dependency>
         </dependencies>
         <configuration>
+          <threshold>high</threshold>
           <excludeFilterFile>spotbugs-exclude.xml</excludeFilterFile>
         </configuration>
       </plugin>

server/src/main/java/org/apache/uniffle/server/LocalStorageChecker.java

@@ -22,11 +22,11 @@
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.util.List;
-import java.util.Random;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
 import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.RandomUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -156,8 +156,7 @@ boolean checkStorageReadAndWrite() {
         if (!writeFile.createNewFile()) {
           return false;
         }
-        byte[] data = new byte[1024];
-        new Random().nextBytes(data);
+        byte[] data = RandomUtils.nextBytes(1024);
         try (FileOutputStream fos = new FileOutputStream(writeFile)) {
           fos.write(data);
           fos.flush();

server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java

@@ -46,7 +46,7 @@
 public class ShuffleFlushManager {
 
   private static final Logger LOG = LoggerFactory.getLogger(ShuffleFlushManager.class);
-  public static AtomicLong ATOMIC_EVENT_ID = new AtomicLong(0);
+  public static final AtomicLong ATOMIC_EVENT_ID = new AtomicLong(0);
   private final ShuffleServer shuffleServer;
   private final BlockingQueue<ShuffleDataFlushEvent> flushQueue = Queues.newLinkedBlockingQueue();
   private final ThreadPoolExecutor threadPoolExecutor;