HDDS-12136. Allow StorageVolumeChecker to be disabled

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeConfiguration.java

@@ -309,7 +309,9 @@ public class DatanodeConfiguration extends ReconfigurableConfig {
       defaultValue = "60",
       type = ConfigType.LONG,
       tags = { DATANODE },
-      description = "Periodic disk check run interval in minutes."
+      description = "Periodic disk check run interval in minutes. " +
+          "A negative value disables periodic disk checks. " +
+          "A value of 0 is invalid and defaults to 60."
   )
   private long periodicDiskCheckIntervalMinutes =
       PERIODIC_DISK_CHECK_INTERVAL_MINUTES_DEFAULT;
@@ -602,13 +604,11 @@ public void validate() {
       containerCloseThreads = CONTAINER_CLOSE_THREADS_DEFAULT;
     }
 
-    if (periodicDiskCheckIntervalMinutes < 1) {
-      LOG.warn(PERIODIC_DISK_CHECK_INTERVAL_MINUTES_KEY +
-              " must be greater than zero and was set to {}. Defaulting to {}",
-          periodicDiskCheckIntervalMinutes,
+    if (periodicDiskCheckIntervalMinutes == 0) {
+      LOG.warn("{} must not be zero. Defaulting to {}",
+          PERIODIC_DISK_CHECK_INTERVAL_MINUTES_KEY,
           PERIODIC_DISK_CHECK_INTERVAL_MINUTES_DEFAULT);
-      periodicDiskCheckIntervalMinutes =
-          PERIODIC_DISK_CHECK_INTERVAL_MINUTES_DEFAULT;
+      periodicDiskCheckIntervalMinutes = PERIODIC_DISK_CHECK_INTERVAL_MINUTES_DEFAULT;
     }
 
     if (failedDataVolumesTolerated < -1) {

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/StorageVolumeChecker.java

@@ -150,6 +150,11 @@ public void start() {
     if (started.compareAndSet(false, true)) {
       long periodicDiskCheckIntervalMinutes =
           dnConf.getPeriodicDiskCheckIntervalMinutes();
+      if (periodicDiskCheckIntervalMinutes < 0) {
+        LOG.warn("StorageVolumeChecker is DISABLED because " +
+            "hdds.datanode.periodic.disk.check.interval.minutes = {} (negative).", periodicDiskCheckIntervalMinutes);
+        return;
+      }
       periodicDiskChecker =
           diskCheckerservice.scheduleWithFixedDelay(this::checkAllVolumeSets,
               periodicDiskCheckIntervalMinutes,
@@ -415,7 +420,9 @@ private void invokeCallback() {
    */
   public void shutdownAndWait(int gracePeriod, TimeUnit timeUnit) {
     if (started.compareAndSet(true, false)) {
-      periodicDiskChecker.cancel(true);
+      if (periodicDiskChecker != null) {
+        periodicDiskChecker.cancel(true);
+      }
       diskCheckerservice.shutdownNow();
       checkVolumeResultHandlerExecutorService.shutdownNow();
       metrics.unregister();

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestDatanodeConfiguration.java

@@ -108,7 +108,7 @@ public void acceptsValidValues() {
   public void overridesInvalidValues() {
     // GIVEN
     int invalidDeleteThreads = 0;
-    long invalidDiskCheckIntervalMinutes = -1;
+    long invalidDiskCheckIntervalMinutes = 0;
     int invalidFailedVolumesTolerated = -2;
     long invalidDiskCheckMinGap = -1;
     long invalidDiskCheckTimeout = -1;

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestStorageVolumeChecker.java

@@ -31,6 +31,7 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import java.io.File;
+import java.lang.reflect.Field;
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -299,6 +300,28 @@ public void testNumScansSkipped() throws Exception {
     checker.shutdownAndWait(0, TimeUnit.SECONDS);
   }
 
+  /**
+   * Verify that negative values disable the periodic disk checker.
+   */
+  @Test
+  public void testNegativeIntervalDisablesPeriodicDiskChecker() throws Exception {
+    setup();
+    conf.setLong("hdds.datanode.periodic.disk.check.interval.minutes", -1);
+
+    StorageVolumeChecker checker =
+        new StorageVolumeChecker(conf, new FakeTimer(), "test-");
+
+    checker.start();
+
+    Field f = StorageVolumeChecker.class.getDeclaredField("periodicDiskChecker");
+    f.setAccessible(true);
+    // Expect no ScheduledFuture to be created, as a negative interval disables
+    // the periodic disk checker entirely.
+    assertThat(f.get(checker)).isNull();
+    // Verify shutdown is safe when no periodic task was scheduled.
+    checker.shutdownAndWait(0, TimeUnit.SECONDS);
+  }
+
   /**
    * A checker to wraps the result of {@link HddsVolume#check} in
    * an ImmediateFuture.