HDDS-15291. DN should fail unreferenced block deletion on file errors.

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

@@ -2063,6 +2063,11 @@ public void deleteUnreferenced(Container container, long localID)
     // Since the putBlock request may fail, we don't know if the chunk exists,
     // thus we need to check it when receiving the request to delete such blocks
     String[] chunkNames = getFilesWithPrefix(prefix, chunkDir);
+    if (chunkNames == null) {
+      throw new IOException("Failed to list chunks under " + chunkDir
+          + " for unreferenced block " + localID + " in container "
+          + containerID);
+    }
     if (chunkNames.length == 0) {
       LOG.warn("Missing delete block(Container = {}, Block = {}",
           containerID, localID);
@@ -2073,12 +2078,20 @@ public void deleteUnreferenced(Container container, long localID)
       if (!file.isFile()) {
         continue;
       }
-      FileUtil.fullyDelete(file);
+      if (!deleteUnreferencedFile(file)) {
+        throw new IOException("Failed to delete unreferenced chunk/block "
+            + file + " in container " + containerID);
+      }
       LOG.info("Deleted unreferenced chunk/block {} in container {}", name,
           containerID);
     }
   }
 
+  @VisibleForTesting
+  boolean deleteUnreferencedFile(File file) {
+    return FileUtil.fullyDelete(file);
+  }
+
   @Override
   public ContainerCommandResponseProto readBlock(
       ContainerCommandRequestProto request, Container kvContainer,

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

@@ -702,6 +702,46 @@ public void testContainerChecksumInvocation(ContainerLayoutVersion layoutVersion
     Assertions.assertEquals(1, icrCount.get());
   }
 
+  @ContainerLayoutTestInfo.ContainerTest
+  public void testDeleteUnreferencedFailsWhenChunkDirCannotBeListed(
+      ContainerLayoutVersion layoutVersion) throws Exception {
+    KeyValueHandler keyValueHandler = new KeyValueHandler(conf,
+        DATANODE_UUID, newContainerSet(), mock(MutableVolumeSet.class),
+        mock(ContainerMetrics.class), c -> { },
+        new ContainerChecksumTreeManager(conf));
+    KeyValueContainer container = createContainerWithChunksPath(layoutVersion,
+        Files.createFile(tempDir.resolve("chunks-file")));
+
+    IOException exception = Assertions.assertThrows(IOException.class,
+        () -> keyValueHandler.deleteUnreferenced(container, 1L));
+
+    assertThat(exception)
+        .hasMessageContaining("Failed to list chunks under")
+        .hasMessageContaining("for unreferenced block 1")
+        .hasMessageContaining("in container " + DUMMY_CONTAINER_ID);
+  }
+
+  @ContainerLayoutTestInfo.ContainerTest
+  public void testDeleteUnreferencedFailsWhenFileDeletionFails(
+      ContainerLayoutVersion layoutVersion) throws Exception {
+    FailingUnreferencedDeleteKeyValueHandler keyValueHandler =
+        new FailingUnreferencedDeleteKeyValueHandler(conf);
+    Path chunkDir = Files.createDirectory(tempDir.resolve("chunks"));
+    Path chunkFile = Files.createFile(chunkDir.resolve(
+        getUnreferencedChunkName(layoutVersion, 1L)));
+    KeyValueContainer container =
+        createContainerWithChunksPath(layoutVersion, chunkDir);
+
+    IOException exception = Assertions.assertThrows(IOException.class,
+        () -> keyValueHandler.deleteUnreferenced(container, 1L));
+
+    assertThat(exception)
+        .hasMessageContaining("Failed to delete unreferenced chunk/block")
+        .hasMessageContaining(chunkFile.toString())
+        .hasMessageContaining("in container " + DUMMY_CONTAINER_ID);
+    assertTrue(Files.exists(chunkFile));
+  }
+
   @ContainerLayoutTestInfo.ContainerTest
   public void testUpdateContainerChecksum(ContainerLayoutVersion layoutVersion) throws Exception {
     conf = new OzoneConfiguration();
@@ -1087,4 +1127,39 @@ public void onCompleted() {
       ContainerMetrics.remove();
     }
   }
+
+  private KeyValueContainer createContainerWithChunksPath(
+      ContainerLayoutVersion layoutVersion, Path chunksPath) {
+    KeyValueContainerData data = new KeyValueContainerData(DUMMY_CONTAINER_ID,
+        layoutVersion, GB, PipelineID.randomId().toString(), DATANODE_UUID);
+    data.setChunksPath(chunksPath.toString());
+    return new KeyValueContainer(data, conf);
+  }
+
+  private static String getUnreferencedChunkName(
+      ContainerLayoutVersion layoutVersion, long localID) {
+    switch (layoutVersion) {
+    case FILE_PER_BLOCK:
+      return localID + ".block";
+    case FILE_PER_CHUNK:
+      return localID + "_chunk_0";
+    default:
+      throw new IllegalArgumentException(
+          "Unsupported container layout version " + layoutVersion);
+    }
+  }
+
+  private static final class FailingUnreferencedDeleteKeyValueHandler
+      extends KeyValueHandler {
+    private FailingUnreferencedDeleteKeyValueHandler(OzoneConfiguration conf) {
+      super(conf, DATANODE_UUID, newContainerSet(), mock(MutableVolumeSet.class),
+          mock(ContainerMetrics.class), c -> { },
+          new ContainerChecksumTreeManager(conf));
+    }
+
+    @Override
+    boolean deleteUnreferencedFile(File file) {
+      return false;
+    }
+  }
 }