diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeConfiguration.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeConfiguration.java index 0287084058a..ea9eb4a1bf5 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeConfiguration.java +++ b/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; } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java index 4f6078d0bd2..bf00fb576fb 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java +++ b/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 diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java index ff1fedaa6ac..424290b92b2 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java +++ b/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}). + *

+ * 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); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestDatanodeStateMachine.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestDatanodeStateMachine.java index ef03aa148df..3db702ef3b5 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestDatanodeStateMachine.java +++ b/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