HDDS-14577. Handle missing metadata dir when updating container state

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java

@@ -391,7 +391,18 @@ public void markContainerUnhealthy() throws StorageContainerException {
     writeLock();
     final State prevState = containerData.getState();
     try {
-      updateContainerState(UNHEALTHY);
+      if (!getContainerFile().getParentFile().exists()) {
+        // Metadata directory is absent (e.g. MISSING_METADATA_DIR detected by scanner).
+        // Attempting to write the .container file would fail, and writing a partial file
+        // with only the state field is more harmful than not writing one at all.
+        // The in-memory UNHEALTHY state is sufficient: SCM will receive it via ICR
+        // and schedule deletion without requiring a persisted .container file.
+        containerData.setState(UNHEALTHY);
+        LOG.debug("Skipping .container file update for container {} with missing metadata directory",
+            containerData.getContainerID());
+      } else {
+        updateContainerState(UNHEALTHY);
+      }
       clearPendingPutBlockCache();
     } finally {
       writeUnlock();

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java

@@ -23,6 +23,7 @@
 import static org.apache.hadoop.ozone.container.checksum.ContainerMerkleTreeTestUtils.buildTestTree;
 import static org.apache.hadoop.ozone.container.checksum.ContainerMerkleTreeTestUtils.verifyAllDataChecksumsMatch;
 import static org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration.CONTAINER_SCHEMA_V3_ENABLED;
+import static org.apache.hadoop.ozone.container.keyvalue.TestContainerCorruptions.MISSING_METADATA_DIR;
 import static org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerUtil.isSameSchemaVersion;
 import static org.apache.hadoop.ozone.container.replication.CopyContainerCompression.NO_COMPRESSION;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -759,6 +760,35 @@ public void testReportOfUnhealthyContainer(
     assertNotNull(keyValueContainer.getContainerReport());
   }
 
+  /**
+   * When a container's metadata directory is missing (MISSING_METADATA_DIR detected by the scanner),
+   * markContainerUnhealthy must succeed without throwing. Writing a partial .container file with only
+   * the state field would lose other metadata and is more harmful than writing nothing. The in-memory
+   * UNHEALTHY state is sufficient for SCM to receive it via ICR and schedule deletion.
+   */
+  @ContainerTestVersionInfo.ContainerTest
+  public void testMarkUnhealthyWithMissingMetadataDir(ContainerTestVersionInfo versionInfo) throws Exception {
+    init(versionInfo);
+    keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+
+    // Simulate MISSING_METADATA_DIR using the same corruption helper used in scanner tests.
+    File metadataDir = new File(keyValueContainerData.getMetadataPath());
+    assertTrue(metadataDir.exists(), "Metadata dir should exist before corruption");
+    MISSING_METADATA_DIR.applyTo(keyValueContainer);
+
+    // markContainerUnhealthy must not throw even though the metadata dir is absent.
+    keyValueContainer.markContainerUnhealthy();
+
+    // In-memory state must be UNHEALTHY.
+    assertEquals(ContainerProtos.ContainerDataProto.State.UNHEALTHY,
+        keyValueContainer.getContainerState());
+
+    // The metadata directory must NOT be recreated; no partial .container file should be written.
+    assertFalse(metadataDir.exists(), "Metadata dir should not be recreated by markContainerUnhealthy");
+    assertFalse(keyValueContainer.getContainerFile().exists(),
+        "Container file should not be written when metadata dir is missing");
+  }
+
   @ContainerTestVersionInfo.ContainerTest
   public void testUpdateContainer(ContainerTestVersionInfo versionInfo)
       throws Exception {