HDDS-8447. Datanodes should not process container deletes for failed volumes.

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

@@ -28,6 +28,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.hdds.annotation.InterfaceAudience;
 import org.apache.hadoop.hdds.annotation.InterfaceStability;
 
@@ -382,6 +383,11 @@ private String getIdDir() throws IOException {
     }
   }
 
+  @VisibleForTesting
+  public void setTmpDirPath(Path tmpDirPath) {
+    this.tmpDirPath = tmpDirPath;
+  }
+
   private Path createTmpPath() throws IOException {
 
     // HddsVolume root directory path

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

@@ -75,6 +75,7 @@
 import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration;
 import org.apache.hadoop.ozone.container.common.transport.server.ratis.DispatcherContext;
 import org.apache.hadoop.ozone.container.common.transport.server.ratis.DispatcherContext.WriteChunkStage;
+import org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil;
 import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 import org.apache.hadoop.ozone.container.common.volume.RoundRobinVolumeChoosingPolicy;
 import org.apache.hadoop.ozone.container.common.volume.VolumeSet;
@@ -1328,9 +1329,12 @@ private void deleteInternal(Container container, boolean force)
             .moveToTmpDeleteDirectory(keyValueContainerData, hddsVolume);
 
         if (!success) {
-          LOG.error("Failed to move container under " +
-              hddsVolume.getDeleteServiceDirPath());
-          throw new StorageContainerException("Moving container failed",
+          LOG.error("Failed to move container under " + hddsVolume
+              .getDeleteServiceDirPath());
+          String errorMsg =
+              "Failed to move container" + container.getContainerData()
+                  .getContainerID();
+          triggerVolumeScanAndThrowException(container, errorMsg,
               CONTAINER_INTERNAL_ERROR);
         }
 
@@ -1352,9 +1356,11 @@ private void deleteInternal(Container container, boolean force)
       // empty as a defensive check.
       LOG.error("Could not determine if the container {} is empty",
           container.getContainerData().getContainerID(), e);
-      throw new StorageContainerException("Could not determine if container "
-          + container.getContainerData().getContainerID() +
-          " is empty", DELETE_ON_NON_EMPTY_CONTAINER);
+      String errorMsg =
+          "Could not determine if the container " + container.getContainerData()
+              .getContainerID() + "is empty";
+      triggerVolumeScanAndThrowException(container, errorMsg,
+          DELETE_ON_NON_EMPTY_CONTAINER);
     } finally {
       container.writeUnlock();
     }
@@ -1363,4 +1369,12 @@ private void deleteInternal(Container container, boolean force)
     container.getContainerData().setState(State.DELETED);
     sendICR(container);
   }
+
+  private void triggerVolumeScanAndThrowException(Container container,
+      String msg, ContainerProtos.Result result)
+      throws StorageContainerException {
+    // Trigger a volume scan as exception occurred.
+    StorageVolumeUtil.onFailure(container.getContainerData().getVolume());
+    throw new StorageContainerException(msg, result);
+  }
 }

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

@@ -20,12 +20,15 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.List;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.conf.StorageUnit;
 import org.apache.hadoop.fs.FileUtil;
@@ -90,6 +93,7 @@ public class TestKeyValueHandler {
   private static final String DATANODE_UUID = UUID.randomUUID().toString();
 
   private static final long DUMMY_CONTAINER_ID = 9999;
+  private static final String DUMMY_PATH = "/dummy/dir/doesnt/exist";
 
   private final ContainerLayoutVersion layout;
 
@@ -351,21 +355,24 @@ public void testCloseInvalidContainer() throws IOException {
         ContainerProtos.Result.INVALID_CONTAINER_STATE, response.getResult());
   }
 
+  @SuppressFBWarnings("DMI_HARDCODED_ABSOLUTE_FILENAME")
   @Test
   public void testDeleteContainer() throws IOException {
     final String testDir = GenericTestUtils.getTempPath(
         TestKeyValueHandler.class.getSimpleName() +
             "-" + UUID.randomUUID().toString());
     try {
+      // Case 1 : Regular container delete
       final long containerID = 1L;
       final String clusterId = UUID.randomUUID().toString();
       final String datanodeId = UUID.randomUUID().toString();
       final ConfigurationSource conf = new OzoneConfiguration();
       final ContainerSet containerSet = new ContainerSet(1000);
-      final VolumeSet volumeSet = Mockito.mock(VolumeSet.class);
+      final MutableVolumeSet volumeSet = Mockito.mock(MutableVolumeSet.class);
 
       HddsVolume hddsVolume = new HddsVolume.Builder(testDir).conf(conf)
           .clusterID(clusterId).datanodeUuid(datanodeId)
+          .volumeSet(volumeSet)
           .build();
       hddsVolume.format(clusterId);
       hddsVolume.createWorkingDir(clusterId, null);
@@ -388,16 +395,7 @@ public void testDeleteContainer() throws IOException {
       kvHandler.setClusterID(clusterId);
 
       final ContainerCommandRequestProto createContainer =
-          ContainerCommandRequestProto.newBuilder()
-              .setCmdType(ContainerProtos.Type.CreateContainer)
-              .setDatanodeUuid(datanodeId)
-              .setCreateContainer(
-                  ContainerProtos.CreateContainerRequestProto.newBuilder()
-                      .setContainerType(ContainerType.KeyValueContainer)
-                      .build())
-              .setContainerID(containerID)
-              .setPipelineID(UUID.randomUUID().toString())
-              .build();
+          createContainerRequest(datanodeId, containerID);
 
       kvHandler.handleCreateContainer(createContainer, null);
       Assert.assertEquals(1, icrReceived.get());
@@ -409,8 +407,45 @@ public void testDeleteContainer() throws IOException {
 
       assertFalse(KeyValueContainerUtil.ContainerDeleteDirectory
           .getDeleteLeftovers(hddsVolume).hasNext());
+
+      // Case 2 : failed move of container dir to tmp location should trigger
+      // a volume scan
+
+      final long container2ID = 2L;
+
+      final ContainerCommandRequestProto createContainer2 =
+          createContainerRequest(datanodeId, container2ID);
+
+      kvHandler.handleCreateContainer(createContainer2, null);
+
+      Assert.assertEquals(3, icrReceived.get());
+      Assert.assertNotNull(containerSet.getContainer(container2ID));
+      Path tmpDirPath = hddsVolume.getTmpDirPath();
+      // to simulate failed move
+      hddsVolume.setTmpDirPath(Paths.get(DUMMY_PATH));
+      try {
+        kvHandler.deleteContainer(containerSet.getContainer(container2ID),
+            true);
+      } catch (StorageContainerException sce) {
+        Assert.assertTrue(
+            sce.getMessage().contains("Failed to move container"));
+      }
+      Mockito.verify(volumeSet).checkVolumeAsync(hddsVolume);
+      // cleanup
+      hddsVolume.setTmpDirPath(tmpDirPath);
     } finally {
       FileUtils.deleteDirectory(new File(testDir));
     }
   }
+
+  private static ContainerCommandRequestProto createContainerRequest(
+      String datanodeId, long containerID) {
+    return ContainerCommandRequestProto.newBuilder()
+        .setCmdType(ContainerProtos.Type.CreateContainer)
+        .setDatanodeUuid(datanodeId).setCreateContainer(
+            ContainerProtos.CreateContainerRequestProto.newBuilder()
+                .setContainerType(ContainerType.KeyValueContainer).build())
+        .setContainerID(containerID).setPipelineID(UUID.randomUUID().toString())
+        .build();
+  }
 }