Clarify periodic disk check config behavior and add unit test

Author: Russole

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;

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

@@ -420,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/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.