HDDS-16425. Fail datanode startup cleanly on initialization errors

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

@@ -49,7 +49,9 @@ public class EndpointStateMachine
   private final HostAndPort hostAndPort;
   private final Lock lock;
   private final ConfigurationSource conf;
-  private EndPointStates state = EndPointStates.FIRST;
+  // RunningDatanodeState reads this without the endpoint lock and must see late SHUTDOWN transitions,
+  // even after its wait for the endpoint task has timed out.
+  private volatile EndPointStates state = EndPointStates.FIRST;
   private VersionResponse version;
   private ZonedDateTime lastSuccessfulHeartbeat;
   private boolean isPassive;

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/RunningDatanodeState.java

@@ -52,9 +52,7 @@ public class RunningDatanodeState implements DatanodeState {
   private final ConfigurationSource conf;
   private final StateContext context;
   private CompletionService<EndPointStates> ecs;
-  // Since we connectionManager endpoints can be changed by reconfiguration
-  // we should not rely on ConnectionManager#getValues being unchanged between
-  // execute and await
+  // Include tasks from earlier heartbeats whose completions have not been collected yet.
   private int executingEndpointCount = 0;
 
   public RunningDatanodeState(ConfigurationSource conf,
@@ -88,8 +86,10 @@ public void onExit() {
    */
   @Override
   public void execute(ExecutorService executor) {
-    ecs = new ExecutorCompletionService<>(executor);
-    executingEndpointCount = 0;
+    // Reuse the completion queue across heartbeats so results arriving after await() times out are not lost.
+    if (ecs == null) {
+      ecs = new ExecutorCompletionService<>(executor);
+    }
     for (EndpointStateMachine endpoint : connectionManager.getValues()) {
       Callable<EndPointStates> endpointTask = buildEndPointTask(endpoint);
       if (endpointTask != null) {
@@ -210,26 +210,21 @@ private Callable<EndPointStates> buildEndPointTask(
   public DatanodeStateMachine.DatanodeStates
       await(long duration, TimeUnit timeUnit)
       throws InterruptedException {
-    int returned = 0;
     long durationMS = timeUnit.toMillis(duration);
     long timeLeft = durationMS;
     long startTime = Time.monotonicNow();
     List<Future<EndPointStates>> results = new LinkedList<>();
 
-    while (returned < executingEndpointCount && timeLeft > 0) {
+    while (executingEndpointCount > 0 && timeLeft > 0) {
       Future<EndPointStates> result =
           ecs.poll(timeLeft, TimeUnit.MILLISECONDS);
       if (result != null) {
         results.add(result);
-        returned++;
+        executingEndpointCount--;
       }
       timeLeft = durationMS - (Time.monotonicNow() - startTime);
     }
     return computeNextContainerState(results);
   }
 
-  @Override
-  public void clear() {
-    ecs = null;
-  }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/VersionEndpointTask.java

@@ -18,7 +18,6 @@
 package org.apache.hadoop.ozone.container.common.states.endpoint;
 
 import java.io.IOException;
-import java.net.BindException;
 import java.util.Objects;
 import java.util.concurrent.Callable;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
@@ -72,20 +71,27 @@ public EndpointStateMachine.EndPointStates call() throws Exception {
 
         if (!rpcEndPoint.isPassive()) {
           // If end point is passive, datanode does not need to check volumes.
-          String scmId = response.getValue(OzoneConsts.SCM_ID);
-          String clusterId = response.getValue(OzoneConsts.CLUSTER_ID);
+          try {
+            String scmId = response.getValue(OzoneConsts.SCM_ID);
+            String clusterId = response.getValue(OzoneConsts.CLUSTER_ID);
 
-          Objects.requireNonNull(scmId, "scmId == null");
-          Objects.requireNonNull(clusterId, "clusterId == null");
+            Objects.requireNonNull(scmId, "scmId == null");
+            Objects.requireNonNull(clusterId, "clusterId == null");
 
-          // Check DbVolumes, format DbVolume at first register time.
-          checkVolumeSet(ozoneContainer.getDbVolumeSet(), scmId, clusterId);
+            // Check DbVolumes, format DbVolume at first register time.
+            checkVolumeSet(ozoneContainer.getDbVolumeSet(), scmId, clusterId);
 
-          // Check HddsVolumes
-          checkVolumeSet(ozoneContainer.getVolumeSet(), scmId, clusterId);
+            // Check HddsVolumes
+            checkVolumeSet(ozoneContainer.getVolumeSet(), scmId, clusterId);
 
-          // Start the container services after getting the version information
-          ozoneContainer.start(clusterId);
+            // Start the container services after getting the version information
+            ozoneContainer.start(clusterId);
+          } catch (Exception | Error ex) {
+            // Handle this in the task: its caller may already have timed out waiting for startup.
+            LOG.error("Failed to start required container services for SCM {}. Shutting down datanode.",
+                rpcEndPoint.getAddress(), ex);
+            return rpcEndPoint.setState(EndpointStateMachine.EndPointStates.SHUTDOWN);
+          }
         }
         EndpointStateMachine.EndPointStates nextState =
             rpcEndPoint.getState().getNextState();
@@ -95,9 +101,8 @@ public EndpointStateMachine.EndPointStates call() throws Exception {
         LOG.debug("Cannot execute GetVersion task as endpoint state machine " +
             "is in {} state", rpcEndPoint.getState());
       }
-    } catch (DiskOutOfSpaceException | BindException ex) {
-      rpcEndPoint.setState(EndpointStateMachine.EndPointStates.SHUTDOWN);
     } catch (IOException ex) {
+      // Communication failures are retryable; local initialization failures are handled above.
       rpcEndPoint.logIfNeeded(ex);
     } finally {
       rpcEndPoint.unlock();

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

@@ -51,7 +51,6 @@
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
 import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -147,7 +146,10 @@ public class OzoneContainer {
       recoveringContainerScrubbingService;
   private final GrpcTlsConfig tlsClientConfig;
   private DiskBalancerService diskBalancerService;
-  private final AtomicReference<InitializingStatus> initializingStatus;
+  private final Object initializationLock = new Object();
+  // Guarded by initializationLock.
+  private InitializingStatus initializingStatus;
+  private Throwable initializationFailure;
   private final ReplicationServer replicationServer;
   private DatanodeDetails datanodeDetails;
   private StateContext context;
@@ -160,7 +162,7 @@ public class OzoneContainer {
   private final DatanodeStorageMetrics datanodeStorageMetrics;
 
   enum InitializingStatus {
-    UNINITIALIZED, INITIALIZING, INITIALIZED
+    UNINITIALIZED, INITIALIZING, INITIALIZED, FAILED
   }
 
   /**
@@ -334,7 +336,7 @@ public OzoneContainer(HddsDatanodeService hddsDatanodeService,
 
     datanodeStorageMetrics = DatanodeStorageMetrics.create(volumeSet);
 
-    initializingStatus = new AtomicReference<>(InitializingStatus.UNINITIALIZED);
+    initializingStatus = InitializingStatus.UNINITIALIZED;
   }
 
   /**
@@ -547,24 +549,30 @@ public OnDemandContainerScanner getOnDemandScanner() {
    * @throws IOException
    */
   public void start(String clusterId) throws IOException {
-    // If SCM HA is enabled, OzoneContainer#start() will be called multi-times
-    // from VersionEndpointTask. The first call should do the initializing job,
-    // the successive calls should wait until OzoneContainer is initialized.
-    if (!initializingStatus.compareAndSet(
-        InitializingStatus.UNINITIALIZED, InitializingStatus.INITIALIZING)) {
-
-      // wait OzoneContainer to finish its initializing.
-      while (initializingStatus.get() != InitializingStatus.INITIALIZED) {
-        try {
-          Thread.sleep(1);
-        } catch (InterruptedException e) {
-          Thread.currentThread().interrupt();
-        }
+    synchronized (initializationLock) {
+      // SCM endpoints share one initialization attempt, including its failure.
+      if (initializingStatus == InitializingStatus.INITIALIZED) {
+        LOG.info("Ignore. OzoneContainer already started.");
+        return;
+      }
+      if (initializingStatus == InitializingStatus.FAILED) {
+        throw new IOException("OzoneContainer initialization previously failed", initializationFailure);
+      }
+
+      initializingStatus = InitializingStatus.INITIALIZING;
+      try {
+        initializeContainerServices(clusterId);
+        initializingStatus = InitializingStatus.INITIALIZED;
+      } catch (IOException | RuntimeException | Error ex) {
+        // Partially started services cannot safely be initialized again.
+        initializationFailure = ex;
+        initializingStatus = InitializingStatus.FAILED;
+        throw ex;
       }
-      LOG.info("Ignore. OzoneContainer already started.");
-      return;
     }
+  }
 
+  private void initializeContainerServices(String clusterId) throws IOException {
     DatanodeLayoutStorage layoutStorage
         = new DatanodeLayoutStorage(config);
     layoutStorage.setClusterId(clusterId);
@@ -607,9 +615,6 @@ public void start(String clusterId) throws IOException {
     recoveringContainerScrubbingService.start();
 
     initHddsVolumeContainer();
-
-    // mark OzoneContainer as INITIALIZED.
-    initializingStatus.set(InitializingStatus.INITIALIZED);
   }
 
   /**

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

@@ -18,44 +18,65 @@
 package org.apache.hadoop.ozone.container.common;
 
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_HEARTBEAT_RPC_TIMEOUT;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 
 import com.google.common.collect.Maps;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.io.File;
 import java.io.IOException;
+import java.lang.reflect.Field;
+import java.net.Socket;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.conf.ReconfigurationHandler;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.upgrade.HDDSLayoutFeature;
 import org.apache.hadoop.ipc_.RPC;
+import org.apache.hadoop.ozone.HddsDatanodeStopService;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 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.DatanodeStateMachine;
+import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine.DatanodeStates;
 import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine;
 import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager;
 import org.apache.hadoop.ozone.container.common.states.DatanodeState;
 import org.apache.hadoop.ozone.container.common.states.datanode.InitDatanodeState;
 import org.apache.hadoop.ozone.container.common.states.datanode.RunningDatanodeState;
+import org.apache.hadoop.ozone.container.common.states.endpoint.VersionEndpointTask;
+import org.apache.hadoop.ozone.container.common.transport.server.XceiverServerSpi;
 import org.apache.hadoop.ozone.container.common.volume.CapacityVolumeChoosingPolicy;
+import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer;
+import org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig;
 import org.apache.hadoop.util.concurrent.HadoopExecutors;
 import org.apache.ozone.test.GenericTestUtils;
+import org.apache.ozone.test.GenericTestUtils.LogCapturer;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.api.io.TempDir;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -149,6 +170,76 @@ public void testStartStopDatanodeStateMachine() throws IOException,
     }
   }
 
+  @Test
+  @Timeout(60)
+  void testDelayedRatisStartupFailureStopsDatanode() throws Exception {
+    conf.setFromObject(conf.getObject(ReplicationConfig.class).setPort(0));
+    DatanodeDetails datanodeDetails = getNewDatanodeDetails();
+    ContainerTestUtils.initializeDatanodeLayout(conf, datanodeDetails);
+    CountDownLatch initializing = new CountDownLatch(1);
+    CountDownLatch release = new CountDownLatch(1);
+    CountDownLatch shutdown = new CountDownLatch(1);
+    HddsDatanodeStopService stopService = mock(HddsDatanodeStopService.class);
+    doAnswer(invocation -> {
+      shutdown.countDown();
+      return null;
+    }).when(stopService).stopService();
+    DatanodeStateMachine stateMachine = new DatanodeStateMachine(null, datanodeDetails, conf, null, null,
+        stopService, new ReconfigurationHandler("DN", conf, op -> { }));
+    try (LogCapturer startupLogs = LogCapturer.captureLogs(VersionEndpointTask.class);
+         LogCapturer stateLogs = LogCapturer.captureLogs(RunningDatanodeState.class)) {
+      OzoneContainer container = stateMachine.getContainer();
+      XceiverServerSpi writeChannel = spy(container.getWriteChannel());
+      Field writeChannelField = OzoneContainer.class.getDeclaredField("writeChannel");
+      writeChannelField.setAccessible(true);
+      writeChannelField.set(container, writeChannel);
+      IllegalStateException failure = new IllegalStateException("Failed to initRaftLog",
+          new IOException("Corrupt Raft log"));
+      doAnswer(invocation -> {
+        initializing.countDown();
+        assertThat(release.await(30, TimeUnit.SECONDS)).isTrue();
+        throw failure;
+      }).when(writeChannel).start();
+
+      stateMachine.startDaemon();
+      assertThat(initializing.await(10, TimeUnit.SECONDS)).isTrue();
+      // Replication is already serving when Ratis startup fails.
+      try (Socket socket = new Socket("127.0.0.1", container.getReplicationServer().getPort())) {
+        assertThat(socket.isConnected()).isTrue();
+      }
+      String clusterId = mockServers.get(0).getClusterId();
+      CompletableFuture<Thread> waitingThread = new CompletableFuture<>();
+      Future<?> waitingCaller = executorService.submit(() -> {
+        waitingThread.complete(Thread.currentThread());
+        container.start(clusterId);
+        return null;
+      });
+      Thread caller = waitingThread.get(10, TimeUnit.SECONDS);
+      GenericTestUtils.waitFor(() -> caller.getState() == Thread.State.BLOCKED, 10, 10000);
+      // Let the real heartbeat wait expire before completing the initialization attempt.
+      GenericTestUtils.waitFor(() -> stateLogs.getOutput().contains("Detected timeout"), 10, 10000);
+      assertThat(stateMachine.getContext().getState()).isEqualTo(DatanodeStates.RUNNING);
+      assertThat(shutdown.getCount()).isEqualTo(1);
+      release.countDown();
+
+      assertThat(assertThrows(ExecutionException.class, () -> waitingCaller.get(10, TimeUnit.SECONDS)).getCause())
+          .isInstanceOf(IOException.class).hasCause(failure);
+      assertThat(shutdown.await(10, TimeUnit.SECONDS)).isTrue();
+      stateMachine.getStateMachineThread().join(10000);
+      assertThat(stateMachine.getStateMachineThread().isAlive()).isFalse();
+      assertThat(stateMachine.getContext().getState()).isEqualTo(DatanodeStates.SHUTDOWN);
+      assertThat(stateMachine.getContext().getShutdownOnError()).isTrue();
+      assertThat(startupLogs.getOutput()).contains("Failed to start required container services",
+          "java.lang.IllegalStateException: Failed to initRaftLog", "Caused by: java.io.IOException: Corrupt Raft log");
+      assertThat(assertThrows(IOException.class, () -> container.start(clusterId))).hasCause(failure);
+      verify(writeChannel, times(1)).start();
+      verify(stopService, times(1)).stopService();
+    } finally {
+      release.countDown();
+      stateMachine.stopDaemon();
+    }
+  }
+
   /**
    * 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