HDDS-16421. Close the stream's block file when a request is rejected on the stream

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

@@ -2344,14 +2344,9 @@ private long readBlockImpl(ContainerCommandRequestProto request, RandomAccessFil
 
     final BlockData blockData = getBlockManager().getBlock(kvContainer, blockID);
     if (readBlock.getOffset() >= blockData.getSize()) {
-      // An out of range offset is a client fault, so report it on the stream instead of throwing: throwing would be
-      // turned into a CONTAINER_INTERNAL_ERROR response by readBlock, and a non-null response makes the dispatcher
-      // scan the container as if the data were corrupt.
-      streamObserver.onError(Status.OUT_OF_RANGE
-          .withDescription("Requested offset " + readBlock.getOffset() + " is beyond the end of block " + blockID
-              + " with size " + blockData.getSize())
-          .asRuntimeException());
-      return 0;
+      return rejectReadBlock(blockFile, streamObserver, Status.OUT_OF_RANGE.withDescription(
+          "Requested offset " + readBlock.getOffset() + " is beyond the end of block " + blockID + " with size "
+              + blockData.getSize()));
     }
     final List<ContainerProtos.ChunkInfo> chunkInfos = blockData.getChunks();
     final ChecksumType checksumType = chunkInfos.get(0).getChecksumData().getType();
@@ -2425,6 +2420,18 @@ private long readBlockImpl(ContainerCommandRequestProto request, RandomAccessFil
     return totalDataLength;
   }
 
+  /**
+   * Report a client fault on the stream instead of throwing, which would become a CONTAINER_INTERNAL_ERROR response
+   * and make the dispatcher scan the container. This ends the call from the datanode side, so also close the stream's
+   * block file: GrpcXceiverService closes it only when the client ends the stream or a request throws.
+   */
+  private static long rejectReadBlock(RandomAccessFileChannel blockFile,
+      StreamObserver<ContainerCommandResponseProto> streamObserver, Status status) {
+    blockFile.close();
+    streamObserver.onError(status.asRuntimeException());
+    return 0;
+  }
+
   static List<ByteString> getChecksums(long blockOffset, int readLength, int bytesPerChecksum,
       final List<ContainerProtos.ChunkInfo> chunks) {
     final int bytesPerChunk = Math.toIntExact(chunks.get(0).getLen());

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

@@ -33,6 +33,7 @@
 import static org.apache.ozone.test.MetricsAsserts.assertCounter;
 import static org.apache.ozone.test.MetricsAsserts.getMetrics;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -124,6 +125,7 @@
 import org.apache.hadoop.util.Time;
 import org.apache.ozone.test.GenericTestUtils;
 import org.apache.ozone.test.GenericTestUtils.LogCapturer;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.ratis.thirdparty.io.grpc.Status;
 import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException;
 import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
@@ -1178,12 +1180,8 @@ public void testReadBlockOutOfRangeAtBlockSize() throws Exception {
   @Test
   public void testReadBlockOutOfRangeLastByteInRange() throws Exception {
     ReadBlockResult result = readBlock(BLOCK_SIZE, BLOCK_SIZE - 1, 1024);
-    assertNull(result.getResponse(), "ReadBlock should return null on success");
-    assertThat(result.getErrors()).isEmpty();
-    assertThat(result.getDataResponses()).isNotEmpty();
-    for (ContainerCommandResponseProto response : result.getDataResponses()) {
-      assertEquals(ContainerProtos.Result.SUCCESS, response.getResult());
-    }
+    // The offset is aligned down to the checksum boundary, so the whole block is streamed.
+    assertResponses(result, 0, BLOCK_SIZE);
   }
 
   private void assertOutOfRange(ReadBlockResult result, long offset) {
@@ -1200,64 +1198,130 @@ private void assertOutOfRange(ReadBlockResult result, long offset) {
         .contains(String.valueOf(BLOCK_SIZE));
   }
 
+  /** A request rejected on the stream ends the call from the datanode side, so the block file must be closed there. */
+  @Test
+  public void testReadBlockErrorOnTheStreamClosesTheBlockFile() throws Exception {
+    try (StreamFixture fixture = new StreamFixture()) {
+      fixture.appendChunk("chunk1", 0, BLOCK_SIZE);
+      assertResponses(fixture.read(0, BLOCK_SIZE), 0, BLOCK_SIZE);
+      assertTrue(fixture.blockFile.isOpen(), "block file stays open on the stream after a successful read");
+      assertOutOfRange(fixture.read(BLOCK_SIZE, 1024), BLOCK_SIZE);
+      assertFalse(fixture.blockFile.isOpen(), "block file closed after OUT_OF_RANGE");
+    }
+  }
+
+  /** Each byte of the block is the low byte of its position. */
+  private static byte[] writtenBytes(long offset, int length) {
+    final byte[] bytes = new byte[length];
+    for (int i = 0; i < length; i++) {
+      bytes[i] = (byte) (offset + i);
+    }
+    return bytes;
+  }
+
+  /** Asserts a successful read of {@code totalLength} contiguous bytes from {@code firstOffset}. */
+  private static void assertResponses(ReadBlockResult result, long firstOffset, long totalLength) {
+    assertNull(result.getResponse());
+    assertThat(result.getErrors()).isEmpty();
+    long offset = firstOffset;
+    for (ContainerCommandResponseProto response : result.getDataResponses()) {
+      assertEquals(ContainerProtos.Result.SUCCESS, response.getResult());
+      assertEquals(offset, response.getReadBlock().getOffset());
+      final ByteString data = response.getReadBlock().getData();
+      assertArrayEquals(writtenBytes(offset, data.size()), data.toByteArray());
+      offset += data.size();
+    }
+    assertEquals(totalLength, offset - firstOffset, "total bytes delivered");
+  }
+
   /**
-   * Reads a block of {@code blockSize} bytes from a real container through {@link Handler#readBlock},
-   * collecting the streamed responses and errors.
+   * A real FILE_PER_BLOCK container with one block and one {@link RandomAccessFileChannel} standing in for the
+   * per-stream channel of GrpcXceiverService.
    */
-  private ReadBlockResult readBlock(int blockSize, long offset, long length) throws Exception {
-    Path testDir = Files.createTempDirectory("testReadBlockOutOfRange");
-    RandomAccessFileChannel blockFile = null;
-    try {
-      conf.set(OZONE_SCM_CONTAINER_LAYOUT_KEY, ContainerLayoutVersion.FILE_PER_BLOCK.name());
-      HandlerWithVolumeSet handlerWithVolume = createKeyValueHandlerWithVolumeSet(testDir);
-      KeyValueHandler kvHandler = handlerWithVolume.getHandler();
-      MutableVolumeSet volumeSet = handlerWithVolume.getVolumeSet();
-      ContainerSet containerSet = handlerWithVolume.getContainerSet();
+  private final class StreamFixture implements AutoCloseable {
+    private final Path testDir = Files.createTempDirectory("testReadBlock");
+    private final KeyValueHandler kvHandler;
+    private final KeyValueContainer container;
+    private final BlockID blockID;
+    private final BlockData blockData;
+    private final RandomAccessFileChannel blockFile = new RandomAccessFileChannel();
 
-      long containerID = ContainerTestHelper.getTestContainerID();
-      KeyValueContainerData containerData = new KeyValueContainerData(
-          containerID, ContainerLayoutVersion.FILE_PER_BLOCK,
-          (long) StorageUnit.GB.toBytes(1), UUID.randomUUID().toString(),
-          DATANODE_UUID);
-      KeyValueContainer container = new KeyValueContainer(containerData, conf);
-      container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), CLUSTER_ID);
-      containerSet.addContainer(container);
+    StreamFixture() throws Exception {
+      try {
+        conf.set(OZONE_SCM_CONTAINER_LAYOUT_KEY, ContainerLayoutVersion.FILE_PER_BLOCK.name());
+        HandlerWithVolumeSet handlerWithVolume = createKeyValueHandlerWithVolumeSet(testDir);
+        kvHandler = handlerWithVolume.getHandler();
+        MutableVolumeSet volumeSet = handlerWithVolume.getVolumeSet();
+        ContainerSet containerSet = handlerWithVolume.getContainerSet();
+
+        long containerID = ContainerTestHelper.getTestContainerID();
+        KeyValueContainerData containerData = new KeyValueContainerData(
+            containerID, ContainerLayoutVersion.FILE_PER_BLOCK,
+            (long) StorageUnit.GB.toBytes(1), UUID.randomUUID().toString(),
+            DATANODE_UUID);
+        container = new KeyValueContainer(containerData, conf);
+        container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), CLUSTER_ID);
+        containerSet.addContainer(container);
+
+        blockID = ContainerTestHelper.getTestBlockID(containerID);
+        blockData = new BlockData(blockID);
+      } catch (Exception e) {
+        close();
+        throw e;
+      }
+    }
 
-      BlockID blockID = ContainerTestHelper.getTestBlockID(containerID);
-      BlockData blockData = new BlockData(blockID);
-      ChunkInfo chunkInfo = new ChunkInfo("chunk1", 0, blockSize);
+    /** Write one more chunk and re-put the block with it. */
+    void appendChunk(String name, long offset, int length) throws Exception {
+      ChunkInfo chunkInfo = new ChunkInfo(name, offset, length);
       blockData.addChunk(chunkInfo.getProtoBufMessage());
-      kvHandler.getBlockManager().putBlock(container, blockData);
-
-      ChunkBuffer data = ChunkBuffer.wrap(ByteBuffer.allocate(blockSize));
+      ChunkBuffer data = ChunkBuffer.wrap(ByteBuffer.wrap(writtenBytes(offset, length)));
       kvHandler.getChunkManager().writeChunk(container, blockID, chunkInfo, data,
           DispatcherContext.getHandleWriteChunk());
+      kvHandler.getBlockManager().putBlock(container, blockData);
+    }
 
-      ContainerCommandRequestProto readBlockRequest =
-          ContainerCommandRequestProto.newBuilder()
-              .setCmdType(ContainerProtos.Type.ReadBlock)
-              .setContainerID(containerID)
-              .setDatanodeUuid(DATANODE_UUID)
-              .setReadBlock(ContainerProtos.ReadBlockRequestProto.newBuilder()
-                  .setBlockID(blockID.getDatanodeBlockIDProtobuf())
-                  .setOffset(offset)
-                  .setLength(length)
-                  .build())
-              .build();
+    /** Read with one checksum unit per response. */
+    ReadBlockResult read(long offset, long length) {
+      return read(offset, length, (int) BYTES_PER_CHECKSUM);
+    }
 
+    ReadBlockResult read(long offset, long length, int responseDataSize) {
+      ContainerCommandRequestProto request = ContainerCommandRequestProto.newBuilder()
+          .setCmdType(ContainerProtos.Type.ReadBlock)
+          .setContainerID(container.getContainerData().getContainerID())
+          .setDatanodeUuid(DATANODE_UUID)
+          .setReadBlock(ContainerProtos.ReadBlockRequestProto.newBuilder()
+              .setBlockID(blockID.getDatanodeBlockIDProtobuf())
+              .setOffset(offset)
+              .setLength(length)
+              .setResponseDataSize(responseDataSize)
+              .build())
+          .build();
       ReadBlockResult result = new ReadBlockResult(blockID);
-      blockFile = new RandomAccessFileChannel();
-      result.setResponse(kvHandler.readBlock(readBlockRequest, container, blockFile, result, false));
+      result.setResponse(kvHandler.readBlock(request, container, blockFile, result, false));
       return result;
-    } finally {
-      if (blockFile != null) {
-        blockFile.close();
-      }
+    }
+
+    @Override
+    public void close() throws IOException {
+      blockFile.close();
       FileUtils.deleteDirectory(testDir.toFile());
       ContainerMetrics.remove();
     }
   }
 
+  /**
+   * Reads a block of {@code blockSize} bytes from a real container through {@link Handler#readBlock},
+   * collecting the streamed responses and errors.
+   */
+  private ReadBlockResult readBlock(int blockSize, long offset, long length) throws Exception {
+    try (StreamFixture fixture = new StreamFixture()) {
+      fixture.appendChunk("chunk1", 0, blockSize);
+      return fixture.read(offset, length);
+    }
+  }
+
   /**
    * Collects everything a single readBlock call produced: the streamed data responses, the errors and the response
    * proto returned by the handler.