This is an automated email from the ASF dual-hosted git repository.
sodonnel pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 718fe1ac733 HDDS-16421. Close the stream's block file when a request
is rejected on the stream (#11250)
718fe1ac733 is described below
commit 718fe1ac733005a7b5878f88020a4926edf65262
Author: Sergey Soldatov <[email protected]>
AuthorDate: Tue Sep 22 03:39:35 2026 -0700
HDDS-16421. Close the stream's block file when a request is rejected on the
stream (#11250)
---
.../ozone/container/keyvalue/KeyValueHandler.java | 23 +--
.../container/keyvalue/TestKeyValueHandler.java | 160 ++++++++++++++-------
2 files changed, 127 insertions(+), 56 deletions(-)
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<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());
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.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
