Fix handling of fully truncated block of 0 size

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/checksum/ContainerMerkleTree.java

@@ -64,6 +64,16 @@ public void addChunks(long blockID, boolean healthy, ContainerProtos.ChunkInfo..
     id2Block.computeIfAbsent(blockID, BlockMerkleTree::new).addChunks(healthy, chunks);
   }
 
+  /**
+   * Adds an empty block to the tree. This method is not a pre-requisite to {@code addChunks}.
+   * If the block entry already exists, it will not be modified.
+   *
+   * @param blockID The ID of the empty block to add to the tree
+   */
+  public void addBlock(long blockID) {
+    addChunks(blockID, true);
+  }
+
   /**
    * Uses chunk hashes to compute all remaining hashes in the tree, and returns it as a protobuf object. No checksum
    * computation for the tree happens outside of this method.

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

@@ -378,12 +378,13 @@ private List<ContainerScanError> scanBlock(DBHandle db, File dbFile, BlockData b
           // So, we need to make sure, chunk length > 0, before declaring
           // the missing chunk file.
           if (!block.getChunks().isEmpty() && block.getChunks().get(0).getLen() > 0) {
-            ContainerScanError error = new ContainerScanError(FailureType.MISSING_CHUNK_FILE,
+            ContainerScanError error = new ContainerScanError(FailureType.MISSING_DATA_FILE,
                 new File(containerDataFromDisk.getChunksPath()), new IOException("Missing chunk file " +
                 chunkFile.getAbsolutePath()));
             blockErrors.add(error);
           }
         } else if (chunk.getChecksumData().getType() != ContainerProtos.ChecksumType.NONE) {
+          currentTree.addBlock(block.getBlockID().getLocalID());
           int bytesPerChecksum = chunk.getChecksumData().getBytesPerChecksum();
           ByteBuffer buffer = BUFFER_POOL.getBuffer(bytesPerChecksum);
           // Keep scanning the block even if there are errors with individual chunks.
@@ -427,6 +428,7 @@ private static List<ContainerScanError> verifyChecksum(BlockData block,
     ContainerProtos.ChecksumData.Builder observedChecksumData = chunk.getChecksumData().toBuilder();
     observedChecksumData.clearChecksums();
     boolean chunkHealthy = true;
+    boolean chunkMissing = false;
 
     ChecksumData checksumData =
         ChecksumData.getFromProtoBuf(chunk.getChecksumData());
@@ -481,26 +483,41 @@ private static List<ContainerScanError> verifyChecksum(BlockData block,
               new OzoneChecksumException(message)));
         }
       }
-      // If all the checksums match, also check that the length stored in the metadata matches the number of bytes
-      // seen on the disk.
+
       observedChunkBuilder.setLen(bytesRead);
+      // If we haven't seen any errors after scanning the whole chunk, verify that the length stored in the metadata
+      // matches the number of bytes seen on the disk.
       if (chunkHealthy && bytesRead != chunk.getLen()) {
-        String message = String
-            .format("Inconsistent read for chunk=%s expected length=%d"
-                    + " actual length=%d for block %s",
-                chunk.getChunkName(),
-                chunk.getLen(), bytesRead, block.getBlockID());
-        chunkHealthy = false;
-        scanErrors.add(new ContainerScanError(FailureType.INCONSISTENT_CHUNK_LENGTH, chunkFile,
-            new IOException(message)));
+        if (bytesRead == 0) {
+          // If we could not find any data for the chunk, report it as missing.
+          chunkMissing = true;
+          chunkHealthy = false;
+          String message = String.format("Missing chunk=%s with expected length=%d for block %s",
+                  chunk.getChunkName(), chunk.getLen(), block.getBlockID());
+          scanErrors.add(new ContainerScanError(FailureType.MISSING_CHUNK, chunkFile, new IOException(message)));
+        } else {
+          // We found data for the chunk, but it was shorter than expected.
+          String message = String
+              .format("Inconsistent read for chunk=%s expected length=%d"
+                      + " actual length=%d for block %s",
+                  chunk.getChunkName(),
+                  chunk.getLen(), bytesRead, block.getBlockID());
+          chunkHealthy = false;
+          scanErrors.add(new ContainerScanError(FailureType.INCONSISTENT_CHUNK_LENGTH, chunkFile,
+              new IOException(message)));
+        }
       }
     } catch (IOException ex) {
+      // An unknown error occurred trying to access the chunk. Report it as corrupted.
       chunkHealthy = false;
-      scanErrors.add(new ContainerScanError(FailureType.MISSING_CHUNK_FILE, chunkFile, ex));
+      scanErrors.add(new ContainerScanError(FailureType.CORRUPT_CHUNK, chunkFile, ex));
     }
 
-    observedChunkBuilder.setChecksumData(observedChecksumData);
-    currentTree.addChunks(block.getBlockID().getLocalID(), chunkHealthy, observedChunkBuilder.build());
+    // Missing chunks should not be added to the merkle tree.
+    if (!chunkMissing) {
+      observedChunkBuilder.setChecksumData(observedChecksumData);
+      currentTree.addChunks(block.getBlockID().getLocalID(), chunkHealthy, observedChunkBuilder.build());
+    }
     return scanErrors;
   }
 

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

@@ -32,9 +32,10 @@ public enum FailureType {
     MISSING_METADATA_DIR,
     MISSING_CONTAINER_FILE,
     MISSING_CHUNKS_DIR,
-    MISSING_CHUNK_FILE,
+    MISSING_DATA_FILE,
     CORRUPT_CONTAINER_FILE,
     CORRUPT_CHUNK,
+    MISSING_CHUNK,
     INCONSISTENT_CHUNK_LENGTH,
     INACCESSIBLE_DB,
     WRITE_FAILURE,

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

@@ -93,7 +93,7 @@ public enum TestContainerCorruptions {
   MISSING_BLOCK((container, blockID) -> {
     File blockFile = getBlock(container, blockID);
     assertTrue(blockFile.delete());
-  }, ContainerScanError.FailureType.MISSING_CHUNK_FILE),
+  }, ContainerScanError.FailureType.MISSING_DATA_FILE),
 
   CORRUPT_CONTAINER_FILE((container, blockID) -> {
     File containerFile = container.getContainerFile();
@@ -113,7 +113,7 @@ public enum TestContainerCorruptions {
   TRUNCATED_BLOCK((container, blockID) -> {
     File blockFile = getBlock(container, blockID);
     truncateFile(blockFile);
-  }, ContainerScanError.FailureType.INCONSISTENT_CHUNK_LENGTH);
+  }, ContainerScanError.FailureType.MISSING_CHUNK);
 
   private final BiConsumer<Container<?>, Long> corruption;
   private final ContainerScanError.FailureType expectedResult;

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

@@ -160,6 +160,8 @@ public void testAllDataErrorsCollected(ContainerTestVersionInfo versionInfo) thr
 
     DataScanResult result = kvCheck.fullCheck(throttler, null);
     assertTrue(result.isHealthy());
+    // The scanner would write the checksum file to disk. `KeyValueContainerCheck` does not, so we will create the
+    // result here.
     ContainerProtos.ContainerChecksumInfo healthyChecksumInfo = ContainerProtos.ContainerChecksumInfo.newBuilder()
         .setContainerID(containerID)
         .setContainerMerkleTree(result.getDataTree().toProto())
@@ -201,23 +203,26 @@ public void testAllDataErrorsCollected(ContainerTestVersionInfo versionInfo) thr
     // compare it against the original healthy tree. The diff we get back should match the failures we injected.
     ContainerDiffReport diffReport = checksumManager.diff(container.getContainerData(), healthyChecksumInfo);
 
+    LOG.info("Diff of healthy container with actual container {}", diffReport);
+
     // Check that the new tree identified all the expected errors by checking the diff.
     Map<Long, List<ContainerProtos.ChunkMerkleTree>> corruptChunks = diffReport.getCorruptChunks();
     // One block had corrupted chunks.
     assertEquals(1, corruptChunks.size());
     List<ContainerProtos.ChunkMerkleTree> corruptChunksInBlock = corruptChunks.get(corruptBlockID);
     assertEquals(2, corruptChunksInBlock.size());
 
-    // Check missing block was correctly identified in the tree diff.
-    List<ContainerProtos.BlockMerkleTree> missingBlocks = diffReport.getMissingBlocks();
-    assertEquals(1, missingBlocks.size());
-    assertEquals(missingBlockID, missingBlocks.get(0).getBlockID());
-
     // One block was truncated which resulted in all of its chunks being reported as missing.
     Map<Long, List<ContainerProtos.ChunkMerkleTree>> missingChunks = diffReport.getMissingChunks();
     assertEquals(1, missingChunks.size());
     List<ContainerProtos.ChunkMerkleTree> missingChunksInBlock = missingChunks.get(truncatedBlockID);
     assertEquals(CHUNKS_PER_BLOCK, missingChunksInBlock.size());
+
+    // Check missing block was correctly identified in the tree diff.
+    List<ContainerProtos.BlockMerkleTree> missingBlocks = diffReport.getMissingBlocks();
+    assertEquals(1, missingBlocks.size());
+    assertEquals(missingBlockID, missingBlocks.get(0).getBlockID());
+
   }
 
   /**