diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java index 8dd68427001..bab8b03bbf8 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java +++ b/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 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 streamObserver, Status status) { + blockFile.close(); + streamObserver.onError(status.asRuntimeException()); + return 0; + } + static List getChecksums(long blockOffset, int readLength, int bytesPerChecksum, final List chunks) { final int bytesPerChunk = Math.toIntExact(chunks.get(0).getLen()); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java index b4f9895a336..4418c0a282b 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java +++ b/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.