[#2241] feat(server): Introduce option to mark server unhealthy once any storage corrupted (#2245)

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

Introduce option to mark server unhealthy once any storage corrupted.

### Why are the changes needed?

For: #2241
This feature is to reduce the impact while the local directories are corrupted.

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

Yes. 

`rss.server.health.markUnhealthyOnceStorageCorruption` is introduced, the default value is false that will not activate this feature by default.


### How was this patch tested?

Existing unit tests.

Co-authored-by: Junfan Zhang <zhangjunfan@qiyi.com>

Author: zuston

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

@@ -62,6 +62,7 @@ public class LocalStorageChecker extends Checker {
   private boolean isHealthy = true;
   private ExecutorService workers;
   private ReconfigurableConfManager.Reconfigurable<Long> diskCheckerExecutionTimeoutMs;
+  private boolean markUnhealthyOnceDirCorrupted = false;
 
   public LocalStorageChecker(ShuffleServerConf conf, List<LocalStorage> storages) {
     super(conf);
@@ -88,6 +89,9 @@ public LocalStorageChecker(ShuffleServerConf conf, List<LocalStorage> storages)
     this.diskCheckerExecutionTimeoutMs =
         conf.getReconfigurableConf(ShuffleServerConf.HEALTH_CHECKER_LOCAL_STORAGE_EXECUTE_TIMEOUT);
     this.workers = Executors.newFixedThreadPool(basePaths.size());
+
+    this.markUnhealthyOnceDirCorrupted =
+        conf.get(ShuffleServerConf.SERVER_UNHEALTHY_ONCE_STORAGE_CORRUPTION);
   }
 
   @Override
@@ -179,6 +183,15 @@ public boolean checkIsHealthy() {
       return false;
     }
 
+    if (markUnhealthyOnceDirCorrupted && corruptedDirs.get() > 0) {
+      if (isHealthy) {
+        LOG.info(
+            "shuffle server become unhealthy because {} corrupted dirs exist", corruptedDirs.get());
+      }
+      isHealthy = false;
+      return false;
+    }
+
     double availablePercentage = num.get() * 100.0 / storageInfos.size();
     if (Double.compare(availablePercentage, minStorageHealthyPercentage) >= 0) {
       if (!isHealthy) {

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

@@ -267,6 +267,13 @@ public class ShuffleServerConf extends RssBaseConf {
           .defaultValue(2 * 1024L * 1024L)
           .withDescription("The index file size hint");
 
+  public static final ConfigOption<Boolean> SERVER_UNHEALTHY_ONCE_STORAGE_CORRUPTION =
+      ConfigOptions.key("rss.server.health.markUnhealthyOnceStorageCorruption")
+          .booleanType()
+          .defaultValue(false)
+          .withDescription(
+              "Mark server unhealthy once any storage corrupted. Default value is false");
+
   public static final ConfigOption<Double> HEALTH_STORAGE_MAX_USAGE_PERCENTAGE =
       ConfigOptions.key("rss.server.health.max.storage.usage.percentage")
           .doubleType()