Skip to content

HDDS-15857. Fix checksum calculation in KeyValueHandler for variable-sized chunks #10764

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chungen0126 wants to merge 10 commits into
base: master
Choose a base branch
from
Open

HDDS-15857. Fix checksum calculation in KeyValueHandler for variable-sized chunks #10764

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
6f964d6
fix server side bug
chungen0126 Jul 14, 2026
a65be93
HDDS-15857. Fix checksum calculation in KeyValueHandler for variable-…
chungen0126 Jul 14, 2026
3779fad
update test
chungen0126 Jul 14, 2026
8202225
Merge branch 'master' into HDDS-15857
chungen0126 Jul 14, 2026
e9b3183
throw exception if checksum data is missing
chungen0126 Jul 14, 2026
e0ae947
fix checkstyle
chungen0126 Jul 14, 2026
ec1d56b
fix findbug
chungen0126 Jul 14, 2026
86cf0f3
fix checkstyle
chungen0126 Jul 14, 2026
445579d
fix bugs
chungen0126 Jul 14, 2026
4745196
revert TestOzoneContractFSO
chungen0126 Jul 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 49 additions & 2 deletions ...-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.DatanodeID;
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.ReadBlockResponseProto;
import org.apache.hadoop.hdds.scm.OzoneClientConfig;
import org.apache.hadoop.hdds.scm.StreamingReadResponse;
Expand All @@ -55,6 +56,7 @@
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.hadoop.ozone.common.Checksum;
import org.apache.hadoop.ozone.common.ChecksumData;
import org.apache.hadoop.ozone.common.OzoneChecksumException;
import org.apache.hadoop.security.token.Token;
import org.apache.ratis.protocol.exceptions.TimeoutIOException;
import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
Expand Down Expand Up @@ -532,10 +534,14 @@ public void onNext(ContainerProtos.ContainerCommandResponseProto containerComman
ByteBuffer data = readBlock.getData().asReadOnlyByteBuffer();
if (verifyChecksum) {
ChecksumData checksumData = ChecksumData.getFromProtoBuf(readBlock.getChecksumData());
Checksum.verifyChecksum(data, checksumData, 0);
if (readBlock.hasChunkInfoList()) {
verifyChecksumForReadBlock(data, checksumData, readBlock);
} else {
throw new IOException("Checksum data is missing for block " + getBlockID());
}
Comment on lines 535 to +541
}
offerToQueue(readBlock);
} catch (Exception e) {
} catch (IOException | RuntimeException e) {
// Record the failure first: the log and observer calls below must not mask it.
setFailed(e);
final ByteString data = readBlock.getData();
Expand Down Expand Up @@ -628,5 +634,46 @@ public void setStreamingReadResponse(StreamingReadResponse streamingReadResponse
public String toString() {
return name;
}

private void verifyChecksumForReadBlock(
ByteBuffer data, ChecksumData checksumData, ReadBlockResponseProto readBlock)
throws OzoneChecksumException {
if (!checksumData.getChecksumType().equals(ChecksumType.NONE)) {
int bytesPerChecksum = checksumData.getBytesPerChecksum();
long blockOffset = readBlock.getOffset();
long readLength = data.remaining();
long currentChunkOffset = 0;
int checksumIndex = 0;
int dataOffset = 0;

for (ContainerProtos.ChunkInfo chunk : readBlock.getChunkInfoList().getChunksList()) {
long chunkStart = currentChunkOffset;
long chunkEnd = chunkStart + chunk.getLen();

Comment on lines +642 to +652
long overlapStart = Math.max(blockOffset, chunkStart);
long overlapEnd = Math.min(blockOffset + readLength, chunkEnd);

if (overlapStart < overlapEnd) {
int overlapLen = Math.toIntExact(overlapEnd - overlapStart);
ByteBuffer chunkData = data.duplicate();
chunkData.position(data.position() + dataOffset);
chunkData.limit(data.position() + dataOffset + overlapLen);

Checksum.verifyChecksum(chunkData, checksumData, checksumIndex);

dataOffset += overlapLen;

long offsetInChunk = overlapStart - chunkStart;
long endOffsetInChunk = overlapEnd - chunkStart;

int firstChecksumIndex = Math.toIntExact(offsetInChunk / bytesPerChecksum);
int lastChecksumIndex = Math.toIntExact((endOffsetInChunk - 1) / bytesPerChecksum);

checksumIndex += (lastChecksumIndex - firstChecksumIndex + 1);
}
currentChunkOffset += chunk.getLen();
}
}
}
}
}
18 changes: 16 additions & 2 deletions ...s/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,9 +46,13 @@
import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumData;
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.protocol.datanode.proto.ContainerProtos.ChunkInfoList;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadBlockResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.OzoneClientConfig;
Expand Down Expand Up @@ -773,15 +777,25 @@ private ContainerCommandResponseProto buildResponseProto(byte[] data, long offse
private ContainerCommandResponseProto buildCorruptResponseProto(byte[] data, long offset) {
return ContainerCommandResponseProto.newBuilder()
.setCmdType(Type.ReadBlock)
.setResult(ContainerProtos.Result.SUCCESS)
.setResult(Result.SUCCESS)
.setReadBlock(ReadBlockResponseProto.newBuilder()
.setOffset(offset)
.setData(ByteString.copyFrom(data))
.setChecksumData(ChecksumData.newBuilder()
.setType(ContainerProtos.ChecksumType.CRC32)
.setType(ChecksumType.CRC32)
.setBytesPerChecksum(data.length)
.addChecksums(ByteString.copyFrom(new byte[4]))
.build())
.setChunkInfoList(ChunkInfoList.newBuilder()
.addChunks(ChunkInfo.newBuilder()
.setChunkName("chunk")
.setOffset(offset)
.setChecksumData(ChecksumData.newBuilder()
.setType(ChecksumType.CRC32)
.setBytesPerChecksum(data.length)
.addChecksums(ByteString.copyFrom(new byte[4]))
.build())
.setLen(data.length).build()).build())
.build())
.build();
}
Expand Down
9 changes: 7 additions & 2 deletions ...src/main/java/org/apache/hadoop/hdds/scm/protocolPB/ContainerCommandResponseBuilders.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.BlockData;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfoList;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto.Builder;
Expand All @@ -39,6 +40,7 @@
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ListBlockResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.PutBlockResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.PutSmallFileResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadBlockResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadChunkResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadContainerResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result;
Expand Down Expand Up @@ -336,14 +338,17 @@ public static ContainerCommandResponseProto getReadChunkResponse(
}

public static ContainerCommandResponseProto getReadBlockResponse(
ContainerCommandRequestProto request, ChecksumData checksumData, ByteBuffer data, long offset) {
ContainerCommandRequestProto request, ChecksumData checksumData,
ByteBuffer data, long offset, List<ChunkInfo> chunkInfoList, boolean verifyChecksum) {

ContainerProtos.ReadBlockResponseProto response = ContainerProtos.ReadBlockResponseProto.newBuilder()
ContainerProtos.ReadBlockResponseProto response = ReadBlockResponseProto.newBuilder()
.setChecksumData(checksumData.getProtoBufMessage())
.setData(ByteString.copyFrom(data))
.setOffset(offset)
.setChunkInfoList(ChunkInfoList.newBuilder().addAllChunks(chunkInfoList).build())
.build();
Comment on lines 340 to 349


return getSuccessResponseBuilder(request)
.setReadBlock(response)
.build();
Expand Down
59 changes: 35 additions & 24 deletions ...ner-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2208,7 +2208,7 @@ private long readBlockImpl(ContainerCommandRequestProto request, RandomAccessFil
}
}
final ContainerCommandResponseProto response = getReadBlockResponse(
request, checksumData, buffer, adjustedOffset);
request, checksumData, buffer, adjustedOffset, chunkInfos, verifyChecksum);
final int dataLength = response.getReadBlock().getData().size();
LOG.debug("server onNext response {}: dataLength={}, numChecksums={}",
numResponses, dataLength, response.getReadBlock().getChecksumData().getChecksumsList().size());
Expand All @@ -2225,30 +2225,41 @@ private long readBlockImpl(ContainerCommandRequestProto request, RandomAccessFil
static List<ByteString> getChecksums(long blockOffset, int readLength, int bytesPerChunk, int bytesPerChecksum,
final List<ContainerProtos.ChunkInfo> chunks) {
assertSame(0, blockOffset % bytesPerChecksum, "blockOffset % bytesPerChecksum");
final int numChecksums = 1 + (readLength - 1) / bytesPerChecksum;
final List<ByteString> checksums = new ArrayList<>(numChecksums);
for (int i = 0; i < numChecksums; i++) {
// As the checksums are stored "chunk by chunk", we need to figure out which chunk we start reading from,
// and its offset to pull out the correct checksum bytes for each read.
final int n = i * bytesPerChecksum;
final long offset = blockOffset + n;
final int c = Math.toIntExact(offset / bytesPerChunk);
final int chunkOffset = Math.toIntExact(offset % bytesPerChunk);
final int csi = chunkOffset / bytesPerChecksum;

assertTrue(c < chunks.size(),
() -> "chunkIndex = " + c + " >= chunk.size()" + chunks.size());
final ContainerProtos.ChunkInfo chunk = chunks.get(c);
if (c < chunks.size() - 1) {
assertSame(bytesPerChunk, chunk.getLen(), "bytesPerChunk");
}
final ContainerProtos.ChecksumData checksumDataProto = chunks.get(c).getChecksumData();
assertSame(bytesPerChecksum, checksumDataProto.getBytesPerChecksum(), "bytesPerChecksum");
final List<ByteString> checksumsList = checksumDataProto.getChecksumsList();
assertTrue(csi < checksumsList.size(),
() -> "checksumIndex = " + csi + " >= checksumsList.size()" + checksumsList.size());
checksums.add(checksumsList.get(csi));
final List<ByteString> checksums = new ArrayList<>();

long currentChunkOffset = 0;
for (ContainerProtos.ChunkInfo chunk : chunks) {
long chunkStart = currentChunkOffset;
long chunkEnd = chunkStart + chunk.getLen();

long overlapStart = Math.max(blockOffset, chunkStart);
long overlapEnd = Math.min(blockOffset + readLength, chunkEnd);

Comment on lines +2230 to +2237
if (overlapStart < overlapEnd) {
long offsetInChunk = overlapStart - chunkStart;
long endOffsetInChunk = overlapEnd - chunkStart;

int firstChecksumIndex = Math.toIntExact(offsetInChunk / bytesPerChecksum);
int lastChecksumIndex = Math.toIntExact((endOffsetInChunk - 1) / bytesPerChecksum);

ContainerProtos.ChecksumData checksumDataProto = chunk.getChecksumData();
assertSame(bytesPerChecksum, checksumDataProto.getBytesPerChecksum(), "bytesPerChecksum");
List<ByteString> checksumsList = checksumDataProto.getChecksumsList();

for (int csi = firstChecksumIndex; csi <= lastChecksumIndex; csi++) {
final int finalCsi = csi;
assertTrue(finalCsi < checksumsList.size(),
() -> "checksumIndex = " + finalCsi + " >= checksumsList.size()" + checksumsList.size());
checksums.add(checksumsList.get(finalCsi));
}
}

currentChunkOffset = chunkEnd;
if (currentChunkOffset >= blockOffset + readLength) {
break;
}
}

return checksums;
}

Expand Down
103 changes: 103 additions & 0 deletions ...service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,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.stub.StreamObserver;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
Expand Down Expand Up @@ -1162,4 +1163,106 @@ boolean deleteUnreferencedFile(File file) {
return false;
}
}

@Test
public void testGetChecksumsWithVaryingChunkSizes() {
int bytesPerChecksum = 1024;
int bytesPerChunk = 16 * 1024;

ContainerProtos.ChunkInfo chunk1 = ContainerProtos.ChunkInfo.newBuilder()
.setChunkName("chunk1")
.setOffset(0)
.setLen(1024)
.setChecksumData(ContainerProtos.ChecksumData.newBuilder()
.setType(ContainerProtos.ChecksumType.CRC32)
.setBytesPerChecksum(bytesPerChecksum)
.addChecksums(org.apache.ratis.thirdparty.com.google.protobuf.ByteString.copyFromUtf8("chk1"))
.build())
.build();

ContainerProtos.ChunkInfo chunk2 = ContainerProtos.ChunkInfo.newBuilder()
.setChunkName("chunk2")
.setOffset(1024)
.setLen(10)
.setChecksumData(ContainerProtos.ChecksumData.newBuilder()
.setType(ContainerProtos.ChecksumType.CRC32)
.setBytesPerChecksum(bytesPerChecksum)
.addChecksums(org.apache.ratis.thirdparty.com.google.protobuf.ByteString.copyFromUtf8("chk2"))
.build())
.build();

ContainerProtos.ChunkInfo chunk3 = ContainerProtos.ChunkInfo.newBuilder()
.setChunkName("chunk3")
.setOffset(1034)
.setLen(2048)
.setChecksumData(ContainerProtos.ChecksumData.newBuilder()
.setType(ContainerProtos.ChecksumType.CRC32)
.setBytesPerChecksum(bytesPerChecksum)
.addChecksums(org.apache.ratis.thirdparty.com.google.protobuf.ByteString.copyFromUtf8("chk3-1"))
.addChecksums(org.apache.ratis.thirdparty.com.google.protobuf.ByteString.copyFromUtf8("chk3-2"))
.build())
.build();

List<ContainerProtos.ChunkInfo> chunks = java.util.Arrays.asList(chunk1, chunk2, chunk3);

// Read full block (1024 + 10 + 2048)
List<ByteString> checksums = KeyValueHandler.getChecksums(0, 3082, bytesPerChunk, bytesPerChecksum, chunks);
assertEquals(4, checksums.size());
assertEquals("chk1", checksums.get(0).toStringUtf8());
assertEquals("chk2", checksums.get(1).toStringUtf8());
assertEquals("chk3-1", checksums.get(2).toStringUtf8());
assertEquals("chk3-2", checksums.get(3).toStringUtf8());

// Read from offset 1024
checksums = KeyValueHandler.getChecksums(1024, 2058, bytesPerChunk, bytesPerChecksum, chunks);
assertEquals(3, checksums.size());
assertEquals("chk2", checksums.get(0).toStringUtf8());
assertEquals("chk3-1", checksums.get(1).toStringUtf8());
assertEquals("chk3-2", checksums.get(2).toStringUtf8());

// Read from offset 2048
checksums = KeyValueHandler.getChecksums(2048, 1034, bytesPerChunk, bytesPerChecksum, chunks);
assertEquals(2, checksums.size());
assertEquals("chk3-1", checksums.get(0).toStringUtf8());
assertEquals("chk3-2", checksums.get(1).toStringUtf8());
}

@Test
public void testGetChecksumsWithSmallChunks() {
int bytesPerChecksum = 1024;
int bytesPerChunk = 16 * 1024;

ContainerProtos.ChunkInfo chunk1 = ContainerProtos.ChunkInfo.newBuilder()
.setChunkName("chunk1")
.setOffset(0)
.setLen(1)
.setChecksumData(ContainerProtos.ChecksumData.newBuilder()
.setType(ContainerProtos.ChecksumType.CRC32)
.setBytesPerChecksum(bytesPerChecksum)
.addChecksums(org.apache.ratis.thirdparty.com.google.protobuf.ByteString.copyFromUtf8("chk1"))
.build())
.build();

ContainerProtos.ChunkInfo chunk2 = ContainerProtos.ChunkInfo.newBuilder()
.setChunkName("chunk2")
.setOffset(1)
.setLen(1)
.setChecksumData(ContainerProtos.ChecksumData.newBuilder()
.setType(ContainerProtos.ChecksumType.CRC32)
.setBytesPerChecksum(bytesPerChecksum)
.addChecksums(org.apache.ratis.thirdparty.com.google.protobuf.ByteString.copyFromUtf8("chk2"))
.build())
.build();

List<ContainerProtos.ChunkInfo> chunks = java.util.Arrays.asList(chunk1, chunk2);

// Read full block (1 + 1 = 2 bytes)
List<org.apache.ratis.thirdparty.com.google.protobuf.ByteString> checksums =
KeyValueHandler.getChecksums(0, 2, bytesPerChunk, bytesPerChecksum, chunks);

// According to the bug report, this should return 2 checksums.
assertEquals(2, checksums.size());
assertEquals("chk1", checksums.get(0).toStringUtf8());
assertEquals("chk2", checksums.get(1).toStringUtf8());
}
}
1 change: 1 addition & 0 deletions hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -386,6 +386,7 @@ message ReadBlockResponseProto {
required ChecksumData checksumData = 1;
required uint64 offset = 2;
required bytes data = 3;
optional ChunkInfoList chunkInfoList = 4;
}

message EchoRequestProto {
Expand Down
40 changes: 40 additions & 0 deletions ...ntegration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -301,4 +301,44 @@ static String runTestReadKey(String name, SizeInBytes keySize, SizeInBytes buffe
print(name, keySizeByte, elapsedNanos, bufferSize, computedMD5);
return computedMD5;
}

@Test
void testSmallChunksWithLargeChecksum() throws Exception {
final SizeInBytes bytesPerChecksum = SizeInBytes.valueOf(512);
System.out.println("cluster starting ...");
try (MiniOzoneCluster cluster = newCluster(bytesPerChecksum.getSizeInt())) {
cluster.waitForClusterToBeReady();
System.out.println("cluster ready");

OzoneConfiguration conf = cluster.getConf();
OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class);
clientConfig.setStreamReadBlock(true);
clientConfig.setStreamBufferFlushDelay(false);
final OzoneConfiguration steamReadConf = new OzoneConfiguration(conf);
steamReadConf.setFromObject(clientConfig);

try (OzoneClient streamReadClient = OzoneClientFactory.getRpcClient(steamReadConf)) {
Comment on lines +313 to +320
final TestBucket testBucket = TestBucket.newBuilder(streamReadClient).build();
final String keyName = "keySmallChunks";

// Write a key with multiple small chunks
byte[] data = new byte[]{1, 2, 3, 4, 5};
try (OutputStream out = testBucket.delegate()
.createKey(keyName, data.length, RatisReplicationConfig.getInstance(ONE), Collections.emptyMap())) {
for (byte b : data) {
out.write(b);
out.flush(); // Forces a chunk to be created
}
}

// Read it back using stream read
try (InputStream in = testBucket.delegate().readKey(keyName)) {
byte[] readData = new byte[data.length];
int bytesRead = in.read(readData);
assertEquals(data.length, bytesRead);
org.junit.jupiter.api.Assertions.assertArrayEquals(data, readData);
}
}
}
}
}