This is an automated email from the ASF dual-hosted git repository.
szetszwo 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 d3843b157b1 HDDS-15857. Fix checksum calculation in KeyValueHandler
for variable-sized chunks (#10764)
d3843b157b1 is described below
commit d3843b157b15094c10154f429ce33f260b23209d
Author: Chung En Lee <[email protected]>
AuthorDate: Tue Sep 15 13:05:26 2026 +0800
HDDS-15857. Fix checksum calculation in KeyValueHandler for variable-sized
chunks (#10764)
---
.../org/apache/hadoop/ozone/common/Checksum.java | 56 ++++++
.../container/common/impl/HddsDispatcher.java | 2 +-
.../ozone/container/common/interfaces/Handler.java | 2 +-
.../ozone/container/keyvalue/KeyValueHandler.java | 67 ++++---
.../container/keyvalue/ReadBlockComputation.java | 157 ++++++++++++++++
.../container/keyvalue/TestKeyValueHandler.java | 199 ++++++++++++++++++++-
6 files changed, 450 insertions(+), 33 deletions(-)
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/Checksum.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/Checksum.java
index aefe4399a46..128666784eb 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/Checksum.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/Checksum.java
@@ -17,6 +17,8 @@
package org.apache.hadoop.ozone.common;
+import static org.apache.ratis.util.Preconditions.assertSame;
+
import com.google.common.annotations.VisibleForTesting;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
@@ -27,6 +29,7 @@
import java.util.function.Supplier;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType;
+import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
import org.apache.hadoop.hdds.utils.db.IntegerCodec;
import org.apache.hadoop.ozone.common.utils.BufferUtils;
import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
@@ -435,6 +438,59 @@ public static void verifyChecksum(List<ByteBuffer>
bufferList, int startIndex, C
checksumData.verifyChecksumDataMatches(startIndex, computed);
}
+ public static void validateChecksums(ByteBuffer data, long blockOffset, int
startIndex,
+ final List<ChunkInfo> chunks) throws OzoneChecksumException {
+
+ long firstChunkOffset = blockOffset - chunks.get(startIndex).getOffset();
+ int bytesPerChecksum =
chunks.get(startIndex).getChecksumData().getBytesPerChecksum();
+ long readLength = data.remaining();
+ int dataOffset = data.position();
+
+ if (readLength <= 0) {
+ return;
+ }
+
+ assertSame(0, firstChunkOffset % bytesPerChecksum, "blockOffset %
bytesPerChecksum");
+ ContainerProtos.ChunkInfo firstChunk = chunks.get(startIndex);
+
+ // verify first chunk, only the first chunk is not start at zero.
+ int firstChunkIndex = (int) (firstChunkOffset / bytesPerChecksum);
+ int dataLimit = (int) Math.min(firstChunk.getLen() - firstChunkOffset,
readLength);
+ Checksum.verifySingleChunk(data, dataOffset, dataLimit, firstChunk,
firstChunkIndex);
+ dataOffset += dataLimit;
+ readLength -= dataLimit;
+ startIndex++;
+
+ while (readLength > 0) {
+ ContainerProtos.ChunkInfo chunkInfo = chunks.get(startIndex);
+ dataLimit = (int) Math.min(chunkInfo.getLen(), readLength);
+
+ Checksum.verifySingleChunk(data, dataOffset, dataLimit, chunkInfo, 0);
+
+ dataOffset += dataLimit;
+ readLength -= dataLimit;
+ startIndex++;
+ }
+
+ }
+
+ private static void verifySingleChunk(ByteBuffer data, int dataOffset, int
dataLimit,
+ ContainerProtos.ChunkInfo chunkInfo, int checksumIndex) throws
OzoneChecksumException {
+
+ ContainerProtos.ChecksumData protoChecksum = chunkInfo.getChecksumData();
+ final ByteBuffer buffer = data.duplicate();
+ buffer.position(dataOffset);
+ buffer.limit(dataOffset + dataLimit);
+
+ final ChecksumData checksum = new ChecksumData(
+ protoChecksum.getType(),
+ protoChecksum.getBytesPerChecksum(),
+ protoChecksum.getChecksumsList()
+ );
+
+ Checksum.verifyChecksum(buffer, checksum, checksumIndex);
+ }
+
/**
* Returns a ChecksumData with type NONE for testing.
*/
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java
index 1564e3ddbf5..34298c7263f 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java
@@ -877,7 +877,7 @@ public void streamDataReadOnly(ContainerCommandRequestProto
msg,
ContainerProtos.Result.CONTAINER_INTERNAL_ERROR);
}
perf.appendPreOpLatencyMs(Time.monotonicNow() - startTime);
- responseProto = handler.readBlock(msg, container, blockFile,
streamObserver);
+ responseProto = handler.readBlock(msg, container, blockFile,
streamObserver, false);
long oPLatencyMS = Time.monotonicNow() - startTime;
metrics.incContainerOpsLatencies(cmdType, oPLatencyMS);
if (responseProto == null) {
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Handler.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Handler.java
index d6828423a36..53dd03627e1 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Handler.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Handler.java
@@ -302,5 +302,5 @@ public abstract void copyContainer(
public abstract ContainerCommandResponseProto readBlock(
ContainerCommandRequestProto msg, Container container,
RandomAccessFileChannel blockFile,
- StreamObserver<ContainerCommandResponseProto> streamObserver);
+ StreamObserver<ContainerCommandResponseProto> streamObserver, boolean
testVariableChunks);
}
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 d872d7a91e2..8dd68427001 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
@@ -57,7 +57,6 @@
import static
org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.putBlockResponseSuccess;
import static
org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.unsupportedRequest;
import static
org.apache.hadoop.hdds.scm.utils.ClientCommandsUtils.getReadChunkVersion;
-import static org.apache.hadoop.hdds.utils.IOUtils.roundUp;
import static org.apache.hadoop.ozone.OzoneConsts.INCREMENTAL_CHUNK_LIST;
import static
org.apache.hadoop.ozone.container.checksum.DNContainerOperationClient.createSingleNodePipeline;
import static
org.apache.hadoop.ozone.container.common.impl.ContainerLayoutVersion.DEFAULT_LAYOUT;
@@ -2285,7 +2284,7 @@ boolean deleteUnreferencedFile(File file) {
public ContainerCommandResponseProto readBlock(
ContainerCommandRequestProto request, Container kvContainer,
RandomAccessFileChannel blockFile,
- StreamObserver<ContainerCommandResponseProto> streamObserver) {
+ StreamObserver<ContainerCommandResponseProto> streamObserver, boolean
testVariableChunks) {
if (kvContainer.getContainerData().getLayoutVersion() != FILE_PER_BLOCK) {
return ContainerUtils.logAndReturnError(LOG,
@@ -2301,7 +2300,7 @@ public ContainerCommandResponseProto readBlock(
}
try {
final long startTime = Time.monotonicNow();
- final long bytesRead = readBlockImpl(request, blockFile, kvContainer,
streamObserver, false);
+ final long bytesRead = readBlockImpl(request, blockFile, kvContainer,
streamObserver, testVariableChunks);
KeyValueContainerData containerData = (KeyValueContainerData) kvContainer
.getContainerData();
HddsVolume volume = containerData.getVolume();
@@ -2325,7 +2324,7 @@ public ContainerCommandResponseProto readBlock(
}
private long readBlockImpl(ContainerCommandRequestProto request,
RandomAccessFileChannel blockFile,
- Container kvContainer, StreamObserver<ContainerCommandResponseProto>
streamObserver, boolean verifyChecksum)
+ Container kvContainer, StreamObserver<ContainerCommandResponseProto>
streamObserver, boolean testVariableChunks)
throws IOException {
final ReadBlockRequestProto readBlock = request.getReadBlock();
int responseDataSize = readBlock.getResponseDataSize();
@@ -2355,32 +2354,36 @@ private long readBlockImpl(ContainerCommandRequestProto
request, RandomAccessFil
return 0;
}
final List<ContainerProtos.ChunkInfo> chunkInfos = blockData.getChunks();
- final int bytesPerChunk = Math.toIntExact(chunkInfos.get(0).getLen());
final ChecksumType checksumType =
chunkInfos.get(0).getChecksumData().getType();
- ChecksumData checksumData = null;
int bytesPerChecksum = STREAMING_BYTES_PER_CHUNK;
- if (checksumType == ContainerProtos.ChecksumType.NONE) {
- checksumData = new ChecksumData(checksumType, 0);
- } else {
+ if (checksumType != ContainerProtos.ChecksumType.NONE) {
bytesPerChecksum =
chunkInfos.get(0).getChecksumData().getBytesPerChecksum();
}
- // We have to align the read to checksum boundaries, so whatever offset is
requested, we have to move back to the
- // previous checksum boundary.
- // eg if bytesPerChecksum is 512, and the requested offset is 600, we have
to move back to 512.
- // If the checksum type is NONE, we don't have to do this, but using no
checksums should be rare in practice and
- // it simplifies the code to always do this.
- final long offsetAlignment = readBlock.getOffset() % bytesPerChecksum;
- long adjustedOffset = readBlock.getOffset() - offsetAlignment;
+ // TODO: Support client-side flag to toggle checksum verification.
+ // If checksum is disabled, chunk offset adjustment can be skipped.
+ int chunkIndex = ReadBlockComputation.searchChunk(readBlock.getOffset(),
chunkInfos);
+ ReadBlockComputation readBlockComputation =
+ new ReadBlockComputation(responseDataSize, bytesPerChecksum,
chunkInfos, chunkIndex);
+ long adjustedOffset =
readBlockComputation.computeAdjustedOffset(readBlock.getOffset());
+
+ long adjustLength = readBlockComputation.computeAdjustedLength(
+ readBlock.getOffset(), readBlock.getLength(), adjustedOffset);
+
+ ChecksumData checksumData = new ChecksumData(checksumType,
bytesPerChecksum);
final ByteBuffer buffer = ByteBuffer.allocate(responseDataSize);
blockFile.position(adjustedOffset);
long totalDataLength = 0;
int numResponses = 0;
- final long rounded = roundUp(readBlock.getLength() + offsetAlignment,
bytesPerChecksum);
- final long requiredLength = Math.min(rounded, blockData.getSize() -
adjustedOffset);
+ Preconditions.checkState(adjustLength <= blockData.getSize() -
adjustedOffset);
LOG.debug("adjustedOffset {}, requiredLength {}, blockSize {}",
- adjustedOffset, requiredLength, blockData.getSize());
- for (boolean shouldRead = true; totalDataLength < requiredLength &&
shouldRead;) {
+ adjustedOffset, adjustLength, blockData.getSize());
+ for (boolean shouldRead = true; totalDataLength < adjustLength &&
shouldRead;) {
+
+ int bufferLimit =
readBlockComputation.computeBufferLimit(adjustedOffset, adjustLength -
totalDataLength);
+
+ buffer.limit(bufferLimit);
+
shouldRead = blockFile.read(buffer);
buffer.flip();
final int readLength = buffer.remaining();
@@ -2391,14 +2394,20 @@ private long readBlockImpl(ContainerCommandRequestProto
request, RandomAccessFil
assertTrue(readLength > 0, () -> "readLength = " + readLength + " <= 0");
if (checksumType != ContainerProtos.ChecksumType.NONE) {
- final List<ByteString> checksums = getChecksums(adjustedOffset,
readLength,
- bytesPerChunk, bytesPerChecksum, chunkInfos);
- LOG.debug("Read {} at adjustedOffset {}, readLength {}, bytesPerChunk
{}, bytesPerChecksum {}",
- readBlock, adjustedOffset, readLength, bytesPerChunk,
bytesPerChecksum);
- checksumData = new ChecksumData(checksumType, bytesPerChecksum,
checksums);
- if (verifyChecksum) {
- Checksum.verifyChecksum(buffer.duplicate(), checksumData, 0);
+
+ if (!testVariableChunks) {
+ // This is the old approach.
+ // It has some bugs, but we are keeping it for now due to backward
compatibility.
+ checksumData = new ChecksumData(checksumType, bytesPerChecksum,
getChecksums(adjustedOffset, readLength,
+ bytesPerChecksum, chunkInfos));
+ }
+
+
+ if (validateChunkChecksumData) {
+ Checksum.validateChecksums(buffer.duplicate(), adjustedOffset,
chunkIndex, chunkInfos);
}
+ LOG.debug("Read {} at adjustedOffset {}, readLength {},
bytesPerChecksum {}",
+ readBlock, adjustedOffset, readLength, bytesPerChecksum);
}
final ContainerCommandResponseProto response = getReadBlockResponse(
request, checksumData, buffer, adjustedOffset);
@@ -2411,12 +2420,14 @@ private long readBlockImpl(ContainerCommandRequestProto
request, RandomAccessFil
adjustedOffset += readLength;
totalDataLength += dataLength;
numResponses++;
+ chunkIndex = readBlockComputation.findChunk(adjustedOffset);
}
return totalDataLength;
}
- static List<ByteString> getChecksums(long blockOffset, int readLength, int
bytesPerChunk, int bytesPerChecksum,
+ static List<ByteString> getChecksums(long blockOffset, int readLength, int
bytesPerChecksum,
final List<ContainerProtos.ChunkInfo> chunks) {
+ final int bytesPerChunk = Math.toIntExact(chunks.get(0).getLen());
assertSame(0, blockOffset % bytesPerChecksum, "blockOffset %
bytesPerChecksum");
final int numChecksums = 1 + (readLength - 1) / bytesPerChecksum;
final List<ByteString> checksums = new ArrayList<>(numChecksums);
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/ReadBlockComputation.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/ReadBlockComputation.java
new file mode 100644
index 00000000000..b08e9062ac0
--- /dev/null
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/ReadBlockComputation.java
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.container.keyvalue;
+
+import java.util.List;
+import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
+import org.apache.ratis.util.Preconditions;
+
+/**
+ * Utility class to compute checksum-aligned offsets, lengths, and buffer
+ * limits for streaming readBlock operations.
+ */
+class ReadBlockComputation {
+
+ private final int responseDataSize;
+ private final int bitMask;
+ private final int bytesPerChecksum;
+ private final List<ChunkInfo> chunks;
+ private int chunkIndex;
+ private long lastPosition;
+
+ ReadBlockComputation(int responseDataSize, int bytesPerChecksum,
+ List<ChunkInfo> chunks, int firstChunkIndex) {
+ this.responseDataSize = responseDataSize;
+
+ Preconditions.assertSame(1, Long.bitCount(bytesPerChecksum), "bitCount");
// check power of 2.
+ // Suppose bytesPerChecksum = 00001000 (a power of 2), then bitMask =
11111000.
+ // The following two computations are the same:
+ // We will use (n & bitMask) to compute ((n / bytesPerChecksum) *
bytesPerChecksum).
+ this.bitMask = -bytesPerChecksum;
+ this.bytesPerChecksum = bytesPerChecksum;
+
+ this.chunks = chunks;
+ this.chunkIndex = firstChunkIndex;
+ this.lastPosition = 0;
+ }
+
+ /**
+ * Binary-search for the index of the chunk whose start offset is the
+ * largest value ≤ {@code targetOffset}.
+ */
+ static int searchChunk(long targetOffset, List<ChunkInfo> chunkInfoList) {
+ int low = 0;
+ int high = chunkInfoList.size() - 1;
+ while (low <= high) {
+ int mid = (low + high) >>> 1;
+ long midVal = chunkInfoList.get(mid).getOffset();
+ if (midVal <= targetOffset) {
+ low = mid + 1;
+ } else {
+ high = mid - 1;
+ }
+ }
+ return high;
+ }
+
+ /**
+ * Return the read offset aligned back to the previous checksum boundary
+ * within the chunk that contains {@code blockOffset}.
+ */
+ long computeAdjustedOffset(long blockOffset) {
+ long offsetAlignment = (blockOffset - chunks.get(chunkIndex).getOffset())
% bytesPerChecksum;
+ return blockOffset - offsetAlignment;
+ }
+
+ /**
+ * Return the length of data that must be read so that the range
+ * {@code [blockOffset, blockOffset + blockLength)} is fully covered
+ * with checksum-aligned boundaries.
+ */
+ long computeAdjustedLength(long blockOffset, long blockLength, long
adjustedOffset) {
+ // We use an inclusive blockEnd to straightforwardly identify which chunk
and checksum
+ // cover the final byte.
+ // For example, if bytesPerChecksum = 16 and we read up to byte 36
(exclusive readEnd = 36),
+ // it correctly locates the chunk. However, calculating the checksum index
from an
+ // exclusive boundary requires extra +1/-1 edge-case handling,
specifically when the
+ // end aligns exactly on a checksum boundary.
+ // Using the inclusive index makes ((blockEnd - chunkOffset) /
bytesPerChecksum) straightforward.
+ long blockEnd = blockOffset + blockLength - 1;
+ ChunkInfo lastChunk = chunks.get(searchChunk(blockEnd, chunks));
+ long chunkOffset = lastChunk.getOffset();
+
+ long roundDown = (blockEnd - chunkOffset) & bitMask;
+ long roundUp = roundDown + bytesPerChecksum;
+ long chunkLength = Math.min(roundUp, lastChunk.getLen());
+ return chunkOffset + chunkLength - adjustedOffset;
+ }
+
+ /**
+ * Compute the buffer limit for the next read iteration, aligned to
+ * checksum boundaries so that partial checksums are not sent across
+ * response messages.
+ */
+ public int computeBufferLimit(long offset, long remainingLength) {
+ if (responseDataSize >= remainingLength) {
+ return Math.toIntExact(remainingLength);
+ }
+
+ /*
+ * Use an exclusive boundary here. If we used an inclusive boundary
+ * that perfectly aligned with the start of the next chunk, it would
+ * incorrectly truncate the buffer limit.
+ *
+ * Example Scenario:
+ * - bytesPerChecksum = 16 (bitMask = -16)
+ * - Chunk 1: offset = 0, length = 20
+ * - Chunk 2: offset = 20, length = 20
+ * - Current offset = 0, responseDataSize = 20
+ *
+ * If using an INCLUSIVE boundary (end = 20 - 1 = 19):
+ * - endChunk = findChunk(19) -> Chunk 1 (offset = 0)
+ * - lengthExcludingEndChunk = 0 - 0 = 0
+ * - Result = ((20 - 0) & -16) + 0 = 16 (Incorrectly truncates 4 bytes)
+ *
+ * If using an EXCLUSIVE boundary (end = 20):
+ * - endChunk = findChunk(20) -> Chunk 2 (offset = 20)
+ * - lengthExcludingEndChunk = 20 - 0 = 20
+ * - Result = ((20 - 20) & -16) + 20 = 20 (Correctly reads the entire
Chunk 1)
+ */
+ ChunkInfo endChunk = chunks.get(findChunk(offset + responseDataSize));
+ final int lengthExcludingEndChunk = Math.toIntExact(endChunk.getOffset() -
offset);
+ // bytesPerChecksum must be a power of 2.
+ return ((responseDataSize - lengthExcludingEndChunk) & bitMask) +
lengthExcludingEndChunk;
+ }
+
+ /**
+ * @param position must be increasing in subsequent calls to this method.
+ * @return the chunk containing the given position.
+ */
+ int findChunk(long position) {
+ Preconditions.assertTrue(position >= lastPosition);
+ lastPosition = position;
+
+ for (; chunkIndex < chunks.size(); chunkIndex++) {
+ final ChunkInfo chunk = chunks.get(chunkIndex);
+ if (position >= chunk.getOffset() && position < chunk.getOffset() +
chunk.getLen()) {
+ return chunkIndex;
+ }
+ }
+ return chunkIndex;
+ }
+}
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 4cc1bae8073..b4f9895a336 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
@@ -70,6 +70,7 @@
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.StorageUnit;
@@ -90,6 +91,8 @@
import org.apache.hadoop.hdds.security.token.TokenVerifier;
import org.apache.hadoop.hdds.utils.io.RandomAccessFileChannel;
import org.apache.hadoop.metrics2.MetricsRecordBuilder;
+import org.apache.hadoop.ozone.common.Checksum;
+import org.apache.hadoop.ozone.common.ChecksumData;
import org.apache.hadoop.ozone.common.ChunkBuffer;
import org.apache.hadoop.ozone.container.ContainerTestHelper;
import org.apache.hadoop.ozone.container.checksum.ContainerChecksumTreeManager;
@@ -1120,7 +1123,7 @@ public void onCompleted() {
blockFile = new RandomAccessFileChannel();
ContainerCommandResponseProto response = kvHandler.readBlock(
- readBlockRequest, container, blockFile, streamObserver);
+ readBlockRequest, container, blockFile, streamObserver, false);
assertNull(response, "ReadBlock should return null on success");
assertTrue(responseCount.get() > 0, "Should receive at least one
response");
@@ -1198,7 +1201,7 @@ private void assertOutOfRange(ReadBlockResult result,
long offset) {
}
/**
- * Reads a block of {@code blockSize} bytes from a real container through
{@link KeyValueHandler#readBlock},
+ * 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 {
@@ -1244,7 +1247,7 @@ private ReadBlockResult readBlock(int blockSize, long
offset, long length) throw
ReadBlockResult result = new ReadBlockResult(blockID);
blockFile = new RandomAccessFileChannel();
- result.setResponse(kvHandler.readBlock(readBlockRequest, container,
blockFile, result));
+ result.setResponse(kvHandler.readBlock(readBlockRequest, container,
blockFile, result, false));
return result;
} finally {
if (blockFile != null) {
@@ -1338,4 +1341,194 @@ boolean deleteUnreferencedFile(File file) {
return false;
}
}
+
+ @Test
+ public void testReadBlockWithSmallChunks() throws Exception {
+ int[] chunkLens = {1, 1, 1, 1, 1};
+ Path testDir = Files.createTempDirectory("testReadBlock");
+ try {
+ HandlerWithVolumeSet handlerWithVolume =
createKeyValueHandlerWithVolumeSet(testDir);
+ 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);
+ BlockID blockID =
ContainerTestHelper.getTestBlockID(container.getContainerData().getContainerID());
+ int totalLen = 0;
+ for (int len : chunkLens) {
+ totalLen += len;
+ }
+
+ byte[] rawData = new byte[totalLen];
+ ThreadLocalRandom.current().nextBytes(rawData);
+ writeBlock(handlerWithVolume, container, blockID, chunkLens, rawData);
+ readBlockAndVerify(handlerWithVolume, container, blockID, rawData, 2, 2,
2, 2);
+ } finally {
+ FileUtils.deleteDirectory(testDir.toFile());
+ ContainerMetrics.remove();
+ }
+ }
+
+ @Test
+ public void testReadBlockWithChunksNotMultipleOfBytesPerChecksum() throws
Exception {
+ int[] chunkLens = {1024, 20, 4096};
+ Path testDir = Files.createTempDirectory("testReadBlock");
+ try {
+ HandlerWithVolumeSet handlerWithVolume =
createKeyValueHandlerWithVolumeSet(testDir);
+ 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);
+ BlockID blockID =
ContainerTestHelper.getTestBlockID(container.getContainerData().getContainerID());
+ int totalLen = 0;
+ for (int len : chunkLens) {
+ totalLen += len;
+ }
+
+ byte[] rawData = new byte[totalLen];
+ ThreadLocalRandom.current().nextBytes(rawData);
+ writeBlock(handlerWithVolume, container, blockID, chunkLens, rawData);
+ readBlockAndVerify(handlerWithVolume, container, blockID, rawData, 2048,
2048, 1044, 3072);
+ } finally {
+ FileUtils.deleteDirectory(testDir.toFile());
+ ContainerMetrics.remove();
+ }
+ }
+
+ @Test
+ public void testReadBlockMultipleResponse() throws Exception {
+ int[] chunkLens = {(1 << 20) + 10, 20, 4096};
+ Path testDir = Files.createTempDirectory("testReadBlock");
+ try {
+ HandlerWithVolumeSet handlerWithVolume =
createKeyValueHandlerWithVolumeSet(testDir);
+ 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);
+ BlockID blockID =
ContainerTestHelper.getTestBlockID(container.getContainerData().getContainerID());
+ int totalLen = 0;
+ for (int len : chunkLens) {
+ totalLen += len;
+ }
+
+ byte[] rawData = new byte[totalLen];
+ ThreadLocalRandom.current().nextBytes(rawData);
+ writeBlock(handlerWithVolume, container, blockID, chunkLens, rawData);
+ readBlockAndVerify(handlerWithVolume, container, blockID, rawData, 20,
(1 << 20) + 20, 0, (1 << 20) + 30 + 1024);
+ } finally {
+ FileUtils.deleteDirectory(testDir.toFile());
+ ContainerMetrics.remove();
+ }
+
+ }
+
+ /**
+ * Creates a container with the given chunk layout, generates random data,
+ * computes real CRC32 checksums, writes each chunk to disk, and persists
+ * the block metadata.
+ */
+ private void writeBlock(HandlerWithVolumeSet handlerWithVolume,
KeyValueContainer container,
+ BlockID blockID, int[] chunkLens, byte[] rawData) throws Exception {
+ KeyValueHandler kvHandler = handlerWithVolume.getHandler();
+ MutableVolumeSet volumeSet = handlerWithVolume.getVolumeSet();
+ ContainerSet containerSet = handlerWithVolume.getContainerSet();
+
+ container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
CLUSTER_ID);
+ containerSet.addContainer(container);
+
+ int bytesPerChecksum = 1024;
+
+ BlockData blockData = new BlockData(blockID);
+
+ long offset = 0;
+ Checksum checksum = new Checksum(ContainerProtos.ChecksumType.CRC32,
bytesPerChecksum);
+ for (int c = 0; c < chunkLens.length; c++) {
+ int len = chunkLens[c];
+ byte[] chunkBytes = new byte[len];
+ System.arraycopy(rawData, (int) offset, chunkBytes, 0, len);
+
+ ChunkBuffer chunkData = ChunkBuffer.wrap(ByteBuffer.wrap(chunkBytes));
+ ChecksumData checksumData = checksum.computeChecksum(chunkData);
+ chunkData.rewind();
+
+ ChunkInfo chunkInfo = new ChunkInfo("chunk" + (c + 1), offset, len);
+ chunkInfo.setChecksumData(checksumData);
+ blockData.addChunk(chunkInfo.getProtoBufMessage());
+
+ kvHandler.getChunkManager().writeChunk(container, blockID, chunkInfo,
chunkData,
+ DispatcherContext.getHandleWriteChunk());
+ offset += len;
+ }
+ kvHandler.getBlockManager().putBlock(container, blockData);
+ }
+
+ /**
+ * Reads the block at the given offset/length via
+ * {@link Handler#readBlock} and verifies that the response data
+ * matches the original bytes written by {@link #writeBlock}.
+ */
+ @SuppressWarnings("checkstyle:ParameterNumber")
+ private void readBlockAndVerify(HandlerWithVolumeSet handlerWithVolume,
KeyValueContainer container,
+ BlockID blockID, byte[] rawData, long readOffset, long length, long
readBackOffset, long readBackLength) {
+ ContainerCommandRequestProto readBlockRequest =
+ ContainerCommandRequestProto.newBuilder()
+ .setCmdType(ContainerProtos.Type.ReadBlock)
+ .setContainerID(container.getContainerData().getContainerID())
+ .setDatanodeUuid(DATANODE_UUID)
+ .setReadBlock(ContainerProtos.ReadBlockRequestProto.newBuilder()
+ .setBlockID(blockID.getDatanodeBlockIDProtobuf())
+ .setOffset(readOffset)
+ .setLength(length)
+ .build())
+ .build();
+
+ @SuppressWarnings("unchecked")
+ StreamObserver<ContainerCommandResponseProto> streamObserver =
mock(StreamObserver.class);
+ List<ContainerCommandResponseProto> capturedResponses = new
java.util.ArrayList<>();
+ doAnswer(invocation -> {
+ capturedResponses.add(invocation.getArgument(0));
+ return null;
+ }).when(streamObserver).onNext(any(ContainerCommandResponseProto.class));
+
+ try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) {
+ ContainerCommandResponseProto response =
handlerWithVolume.getHandler().readBlock(
+ readBlockRequest, container, blockFile, streamObserver, true);
+
+ assertNull(response, "ReadBlock should return null on success");
+ }
+
+ assertFalse(capturedResponses.isEmpty(), "Should receive at least one
response");
+ verify(streamObserver, atLeastOnce()).onNext(any());
+
+ int returnedDataLen = 0;
+ for (ContainerCommandResponseProto resp : capturedResponses) {
+ returnedDataLen += resp.getReadBlock().getData().size();
+ }
+ ByteBuffer allData = ByteBuffer.allocate(returnedDataLen);
+ long firstResponseOffset =
capturedResponses.get(0).getReadBlock().getOffset();
+
+ for (ContainerCommandResponseProto resp : capturedResponses) {
+ assertEquals(ContainerProtos.Result.SUCCESS, resp.getResult());
+ assertTrue(resp.hasReadBlock());
+ allData.put(resp.getReadBlock().getData().asReadOnlyByteBuffer());
+ }
+
+ assertEquals(readBackOffset, firstResponseOffset);
+ assertEquals(readBackLength, returnedDataLen);
+
+ // Verify the returned data matches the expected slice of the original data
+ allData.flip();
+ byte[] readBack = new byte[allData.remaining()];
+ allData.get(readBack);
+ for (int i = 0; i < readBack.length; i++) {
+ assertEquals(rawData[(int) firstResponseOffset + i], readBack[i],
+ "Data mismatch at returned byte index " + i);
+ }
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
