HDDS-6120. Compute block checksum using chunk checksums

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/checksum/AbstractBlockChecksumComputer.java

@@ -0,0 +1,43 @@
+/*
+ * 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.client.checksum;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Base class for computing block checksum which is a function of chunk
+ * checksums.
+ */
+public abstract class AbstractBlockChecksumComputer {
+  private ByteBuffer outByteBuffer;
+
+  /**
+   * Compute block checksum. The result can be obtained by getOutBytes().
+   * @throws IOException
+   */
+  public abstract void compute() throws IOException;
+
+  public ByteBuffer getOutByteBuffer() {
+    return outByteBuffer;
+  }
+
+  public void setOutBytes(byte[] bytes) {
+    this.outByteBuffer = ByteBuffer.wrap(bytes);
+  }
+}

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/BaseFileChecksumHelper.java → hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/checksum/BaseFileChecksumHelper.java

@@ -15,14 +15,16 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.hadoop.ozone.client;
+package org.apache.hadoop.ozone.client.checksum;
 
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.fs.FileChecksum;
 import org.apache.hadoop.fs.MD5MD5CRC32GzipFileChecksum;
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
 import org.apache.hadoop.io.DataOutputBuffer;
 import org.apache.hadoop.io.MD5Hash;
+import org.apache.hadoop.ozone.client.OzoneBucket;
+import org.apache.hadoop.ozone.client.OzoneVolume;
 import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
 import org.apache.hadoop.ozone.client.rpc.RpcClient;
 import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/checksum/ReplicatedBlockChecksumComputer.java

@@ -0,0 +1,72 @@
+/*
+ * 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.client.checksum;
+
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.io.MD5Hash;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * The implementation of AbstractBlockChecksumComputer for replicated blocks.
+ */
+public class ReplicatedBlockChecksumComputer extends
+    AbstractBlockChecksumComputer {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(ReplicatedBlockChecksumComputer.class);
+
+  private List<ContainerProtos.ChunkInfo> chunkInfoList;
+
+  public ReplicatedBlockChecksumComputer(
+      List<ContainerProtos.ChunkInfo> chunkInfoList)
+      throws IOException {
+    this.chunkInfoList = chunkInfoList;
+  }
+
+  @Override
+  public void compute() throws IOException {
+    computeMd5Crc();
+  }
+
+  // compute the block checksum, which is the md5 of chunk checksums
+  private void computeMd5Crc() throws IOException {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+    for (ContainerProtos.ChunkInfo chunkInfo : chunkInfoList) {
+      ContainerProtos.ChecksumData checksumData =
+          chunkInfo.getChecksumData();
+      List<ByteString> checksums = checksumData.getChecksumsList();
+
+      for (ByteString checksum : checksums) {
+        baos.write(checksum.toByteArray());
+      }
+    }
+
+    MD5Hash fileMD5 = MD5Hash.digest(baos.toByteArray());
+    setOutBytes(fileMD5.getDigest());
+
+    LOG.debug("number of chunks={}, md5out={}",
+        chunkInfoList.size(), fileMD5);
+  }
+}

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/ReplicatedFileChecksumHelper.java → hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/checksum/ReplicatedFileChecksumHelper.java

@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.hadoop.ozone.client;
+package org.apache.hadoop.ozone.client.checksum;
 
 import org.apache.hadoop.fs.PathIOException;
 import org.apache.hadoop.hdds.client.BlockID;
@@ -28,11 +28,14 @@
 import org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls;
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
 import org.apache.hadoop.io.MD5Hash;
+import org.apache.hadoop.ozone.client.OzoneBucket;
+import org.apache.hadoop.ozone.client.OzoneVolume;
 import org.apache.hadoop.ozone.client.rpc.RpcClient;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.security.token.Token;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.util.List;
 
 /**
@@ -88,9 +91,10 @@ private boolean checksumBlock(OmKeyLocationInfo keyLocationInfo)
     int bytesPerChecksum = checksumData.getBytesPerChecksum();
     setBytesPerCRC(bytesPerChecksum);
 
-    byte[] blockChecksum = getBlockChecksumFromChunkChecksums(
+    ByteBuffer blockChecksumByteBuffer = getBlockChecksumFromChunkChecksums(
         keyLocationInfo, chunkInfos);
-    String blockChecksumForDebug = populateBlockChecksumBuf(blockChecksum);
+    String blockChecksumForDebug =
+        populateBlockChecksumBuf(blockChecksumByteBuffer);
 
     LOG.debug("got reply from pipeline {} for block {}: blockChecksum={}, " +
             "blockChecksumType={}",
@@ -148,29 +152,31 @@ protected List<ContainerProtos.ChunkInfo> getChunkInfos(
   }
 
   // TODO: copy BlockChecksumHelper here
-  byte[] getBlockChecksumFromChunkChecksums(OmKeyLocationInfo keyLocationInfo,
+  ByteBuffer getBlockChecksumFromChunkChecksums(
+      OmKeyLocationInfo keyLocationInfo,
       List<ContainerProtos.ChunkInfo> chunkInfoList)
       throws IOException {
+    AbstractBlockChecksumComputer blockChecksumComputer =
+        new ReplicatedBlockChecksumComputer(chunkInfoList);
     // TODO: support composite CRC
-    final int lenOfZeroBytes = 32;
-    byte[] emptyBlockMd5 = new byte[lenOfZeroBytes];
-    MD5Hash fileMD5 = MD5Hash.digest(emptyBlockMd5);
-    return fileMD5.getDigest();
+    blockChecksumComputer.compute();
+
+    return blockChecksumComputer.getOutByteBuffer();
   }
 
   /**
-   * Parses out the raw blockChecksum bytes from {@code checksumData}
-   * according to the blockChecksumType and populates the cumulative
+   * Parses out the raw blockChecksum bytes from {@code checksumData} byte
+   * buffer according to the blockChecksumType and populates the cumulative
    * blockChecksumBuf with it.
    *
    * @return a debug-string representation of the parsed checksum if
    *     debug is enabled, otherwise null.
    */
-  String populateBlockChecksumBuf(byte[] checksumData)
+  String populateBlockChecksumBuf(ByteBuffer checksumData)
       throws IOException {
     String blockChecksumForDebug = null;
     //read md5
-    final MD5Hash md5 = new MD5Hash(checksumData);
+    final MD5Hash md5 = new MD5Hash(checksumData.array());
     md5.write(getBlockChecksumBuf());
     if (LOG.isDebugEnabled()) {
       blockChecksumForDebug = md5.toString();

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/checksum/package-info.java

@@ -0,0 +1,23 @@
+/**
+ * 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.client.checksum;
+
+/**
+ * This package contains Ozone Client classes.
+ */

hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/checksum/TestReplicatedBlockChecksumComputer.java

@@ -0,0 +1,68 @@
+/*
+ * 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.client.checksum;
+
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.io.MD5Hash;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.Assert.assertArrayEquals;
+
+/**
+ * Unit tests for ReplicatedBlockChecksumComputer class.
+ */
+public class TestReplicatedBlockChecksumComputer {
+  @Test
+  public void testComputeMd5Crc() throws IOException {
+    final int lenOfZeroBytes = 32;
+    byte[] emptyChunkChecksum = new byte[lenOfZeroBytes];
+    MD5Hash emptyBlockMD5 = MD5Hash.digest(emptyChunkChecksum);
+    byte[] emptyBlockMD5Hash = emptyBlockMD5.getDigest();
+
+    ByteString checkSum = ByteString.copyFrom(emptyChunkChecksum);
+
+    ContainerProtos.ChecksumData checksumData =
+        ContainerProtos.ChecksumData.newBuilder()
+            .addChecksums(checkSum)
+            .setBytesPerChecksum(4)
+            .setType(ContainerProtos.ChecksumType.CRC32)
+            .build();
+    ContainerProtos.ChunkInfo chunkInfo =
+        ContainerProtos.ChunkInfo.newBuilder()
+            .setChecksumData(checksumData)
+            .setChunkName("dummy_chunk")
+            .setOffset(0)
+            .setLen(lenOfZeroBytes)
+            .build();
+    List<ContainerProtos.ChunkInfo> chunkInfoList =
+        Collections.singletonList(chunkInfo);
+    AbstractBlockChecksumComputer computer =
+        new ReplicatedBlockChecksumComputer(chunkInfoList);
+
+    computer.compute();
+
+    ByteBuffer output = computer.getOutByteBuffer();
+    assertArrayEquals(emptyBlockMD5Hash, output.array());
+  }
+}