HDDS-16504. Datanode startup can hang indefinitely when Ratis group recovery stalls

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

@@ -94,6 +94,7 @@ public class DatanodeConfiguration extends ReconfigurableConfig {
   public static final String CONTAINER_CLIENT_CACHE_SIZE = "hdds.datanode.container.client.cache.size";
   public static final String CONTAINER_CLIENT_CACHE_STALE_THRESHOLD =
       "hdds.datanode.container.client.cache.stale.threshold";
+  public static final String CONTAINER_INIT_TIMEOUT_KEY = "hdds.datanode.container.init.timeout";
 
   static final boolean CHUNK_DATA_VALIDATION_CHECK_DEFAULT = false;
 
@@ -471,6 +472,22 @@ public class DatanodeConfiguration extends ReconfigurableConfig {
   )
   private Duration diskCheckTimeout = DISK_CHECK_TIMEOUT_DEFAULT;
 
+  @Config(key = "hdds.datanode.container.init.timeout",
+      defaultValue = "0s",
+      type = ConfigType.TIME,
+      tags = { DATANODE },
+      description = "Maximum time allowed for datanode container services to"
+          + " initialize at startup. Initialization builds the container set"
+          + " and starts the Ratis write channel, which recovers all Raft"
+          + " groups from disk. If it does not complete within this time,"
+          + " startup is failed so the datanode shuts down cleanly instead of"
+          + " hanging indefinitely (for example when Ratis group recovery"
+          + " stalls on a failing volume). A value of 0 (the default) disables"
+          + " the watchdog and preserves the previous behavior of waiting"
+          + " indefinitely. Unit could be defined with postfix (ns,ms,s,m,h,d)."
+  )
+  private Duration containerInitTimeout = Duration.ZERO;
+
   @Config(key = "hdds.datanode.disk.check.sliding.window.timeout",
       defaultValue = "70m",
       type = ConfigType.TIME,
@@ -1093,6 +1110,14 @@ public void setDiskCheckTimeout(Duration duration) {
     diskCheckTimeout = duration;
   }
 
+  public Duration getContainerInitTimeout() {
+    return containerInitTimeout;
+  }
+
+  public void setContainerInitTimeout(Duration duration) {
+    containerInitTimeout = duration;
+  }
+
   public void setDiskCheckEnabled(boolean diskCheckEnabled) {
     isDiskCheckEnabled = diskCheckEnabled;
   }

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

@@ -400,6 +400,18 @@ private void startStateMachineThread() throws IOException {
     }
   }
 
+  /**
+   * Terminates the datanode when startup cannot make progress and the state
+   * machine loop cannot drive the shutdown itself (for example when container
+   * initialization has stalled on a thread that ignores interruption). Invoked
+   * from the {@link org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer}
+   * startup watchdog. Enters JVM shutdown directly rather than the synchronous
+   * stop service, whose stop() could itself block on the same failing disk.
+   */
+  public void triggerFatalShutdown(String reason) {
+    ExitUtils.terminate(1, reason, LOG);
+  }
+
   public void handleFatalVolumeFailures() {
     LOG.error("DatanodeStateMachine Shutdown due to too many bad volumes, "
         + "check " + DatanodeConfiguration.FAILED_DATA_VOLUMES_TOLERATED_KEY

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java

@@ -50,6 +50,7 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
@@ -84,6 +85,7 @@
 import org.apache.hadoop.ozone.container.common.interfaces.VolumeChoosingPolicy;
 import org.apache.hadoop.ozone.container.common.report.IncrementalReportSender;
 import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration;
+import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine;
 import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
 import org.apache.hadoop.ozone.container.common.transport.server.XceiverServerDomainSocket;
 import org.apache.hadoop.ozone.container.common.transport.server.XceiverServerGrpc;
@@ -561,7 +563,7 @@ public void start(String clusterId) throws IOException {
 
       initializingStatus = InitializingStatus.INITIALIZING;
       try {
-        initializeContainerServices(clusterId);
+        initializeContainerServicesWithTimeout(clusterId);
         initializingStatus = InitializingStatus.INITIALIZED;
       } catch (IOException | RuntimeException | Error ex) {
         // Partially started services cannot safely be initialized again.
@@ -572,6 +574,73 @@ public void start(String clusterId) throws IOException {
     }
   }
 
+  /**
+   * Runs {@link #initializeContainerServices(String)} on the calling thread,
+   * optionally guarded by a startup watchdog
+   * ({@code hdds.datanode.container.init.timeout}).
+   * <p>
+   * When the timeout is a positive duration, a watchdog is scheduled before
+   * initialization begins. If initialization has not finished by then - for
+   * example because Ratis group recovery inside {@code writeChannel.start()}
+   * has stalled on a failing volume - the watchdog terminates the datanode via
+   * {@link DatanodeStateMachine#triggerFatalShutdown(String)} (which enters JVM
+   * shutdown) instead of letting startup hang indefinitely. Because
+   * initialization runs on a single thread, no separate worker can keep starting
+   * services after the timeout. A non-positive timeout disables the watchdog,
+   * preserving the previous behavior.
+   */
+  private void initializeContainerServicesWithTimeout(String clusterId) throws IOException {
+    Duration initTimeout =
+        config.getObject(DatanodeConfiguration.class).getContainerInitTimeout();
+    if (initTimeout == null || initTimeout.isZero() || initTimeout.isNegative()) {
+      initializeContainerServices(clusterId);
+      return;
+    }
+
+    // Guards against the watchdog firing just as initialization finishes.
+    AtomicBoolean initInProgress = new AtomicBoolean(true);
+    ScheduledExecutorService watchdog = Executors.newSingleThreadScheduledExecutor(
+        new ThreadFactoryBuilder().setDaemon(true)
+            .setNameFormat("OzoneContainerInitWatchdog").build());
+    watchdog.schedule(() -> {
+      if (initInProgress.compareAndSet(true, false)) {
+        terminateOnInitializationTimeout(initTimeout);
+      }
+    }, initTimeout.toMillis(), TimeUnit.MILLISECONDS);
+    try {
+      initializeContainerServices(clusterId);
+    } finally {
+      initInProgress.set(false);
+      watchdog.shutdownNow();
+    }
+  }
+
+  /**
+   * Terminates the datanode when container initialization has stalled past the
+   * watchdog timeout. A hung step (for example a Ratis {@code join()} or a
+   * blocking disk read) cannot be interrupted reliably, so the only safe
+   * recovery is to fail the process and let it be restarted; the JVM exit also
+   * stops the stalled initialization thread.
+   */
+  private void terminateOnInitializationTimeout(Duration initTimeout) {
+    String message = "OzoneContainer initialization did not complete within " + initTimeout
+        + ". Terminating datanode; a stalled Ratis group recovery or volume I/O is the likely cause.";
+    LOG.error(message);
+    // Enter JVM shutdown directly rather than calling the datanode stop service.
+    // Its synchronous stop() can itself block on the same failing disk (for
+    // example a container scanner's Thread.join()), so it could hang before the
+    // process exits. triggerFatalShutdown() calls ExitUtils.terminate, and
+    // System.exit runs the datanode cleanup registered with ShutdownHookManager,
+    // which bounds each hook with a timeout.
+    StateContext current = context;
+    DatanodeStateMachine dsm = current == null ? null : current.getParent();
+    if (dsm != null) {
+      dsm.triggerFatalShutdown(message);
+    } else {
+      LOG.error("No datanode state machine available; cannot automatically terminate the stalled datanode.");
+    }
+  }
+
   private void initializeContainerServices(String clusterId) throws IOException {
     DatanodeLayoutStorage layoutStorage
         = new DatanodeLayoutStorage(config);

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

@@ -58,6 +58,7 @@
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
 import org.apache.hadoop.ozone.container.common.interfaces.VolumeChoosingPolicy;
+import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration;
 import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine;
 import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine.DatanodeStates;
 import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine;
@@ -73,6 +74,7 @@
 import org.apache.hadoop.util.concurrent.HadoopExecutors;
 import org.apache.ozone.test.GenericTestUtils;
 import org.apache.ozone.test.GenericTestUtils.LogCapturer;
+import org.apache.ratis.util.ExitUtils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -240,6 +242,50 @@ void testDelayedRatisStartupFailureStopsDatanode() throws Exception {
     }
   }
 
+  @Test
+  @Timeout(60)
+  void testStalledInitializationTimeoutTerminatesDatanode() throws Exception {
+    ExitUtils.disableSystemExit();
+    ExitUtils.clear();
+    conf.setFromObject(conf.getObject(ReplicationConfig.class).setPort(0));
+    // Enable the startup watchdog with a short timeout.
+    conf.setTimeDuration(DatanodeConfiguration.CONTAINER_INIT_TIMEOUT_KEY, 2, TimeUnit.SECONDS);
+    DatanodeDetails datanodeDetails = getNewDatanodeDetails();
+    ContainerTestUtils.initializeDatanodeLayout(conf, datanodeDetails);
+    CountDownLatch initializing = new CountDownLatch(1);
+    CountDownLatch release = new CountDownLatch(1);
+    DatanodeStateMachine stateMachine = new DatanodeStateMachine(null, datanodeDetails, conf, null, null,
+        mock(HddsDatanodeStopService.class), new ReconfigurationHandler("DN", conf, op -> { }));
+    try {
+      OzoneContainer container = stateMachine.getContainer();
+      XceiverServerSpi writeChannel = spy(container.getWriteChannel());
+      Field writeChannelField = OzoneContainer.class.getDeclaredField("writeChannel");
+      writeChannelField.setAccessible(true);
+      writeChannelField.set(container, writeChannel);
+      // Stall Ratis startup: never returns and never throws, simulating a hang
+      // (for example a group recovery blocked on a failing volume).
+      doAnswer(invocation -> {
+        initializing.countDown();
+        assertThat(release.await(30, TimeUnit.SECONDS)).isTrue();
+        return null;
+      }).when(writeChannel).start();
+
+      stateMachine.startDaemon();
+      assertThat(initializing.await(10, TimeUnit.SECONDS)).isTrue();
+
+      // The watchdog must enter JVM shutdown even though writeChannel.start()
+      // never returns and never throws. Going straight to ExitUtils (System.exit)
+      // avoids the synchronous stop() path, which could itself block on the disk.
+      GenericTestUtils.waitFor(ExitUtils::isTerminated, 100, 20000);
+      assertThat(ExitUtils.getFirstExitException().getStatus()).isEqualTo(1);
+      assertThat(ExitUtils.getFirstExitException().getMessage()).contains("did not complete within");
+    } finally {
+      release.countDown();
+      stateMachine.stopDaemon();
+      ExitUtils.clear();
+    }
+  }
+
   /**
    * This test explores the state machine by invoking each call in sequence just
    * like as if the state machine would call it. Because this is a test we are