Implement the validator

devabhishekpal · Nov 6, 2024 · 5516cce · 5516cce
1 parent e7eba94
commit 5516cce
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 14 deletions.
diff --git a/...rc/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedStripeInputStream.java b/...rc/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedStripeInputStream.java
@@ -825,9 +825,4 @@ private static SortedSet<Integer> setOfRange(
     return range(startInclusive, endExclusive)
         .boxed().collect(toCollection(TreeSet::new));
   }
-
-  public boolean validateChecksum(ByteBuffer[] buf) {
-
-  }
-
 }
diff --git a/...java/org/apache/hadoop/ozone/container/ec/reconstruction/ECReconstructionCoordinator.java b/...java/org/apache/hadoop/ozone/container/ec/reconstruction/ECReconstructionCoordinator.java
@@ -116,6 +116,7 @@ public class ECReconstructionCoordinator implements Closeable {
   private final ECReconstructionMetrics metrics;
   private final StateContext context;
   private final OzoneClientConfig ozoneClientConfig;
+  private final ECValidator ecValidator;
 
   public ECReconstructionCoordinator(
       ConfigurationSource conf, CertificateClient certificateClient,
@@ -141,6 +142,7 @@ public ECReconstructionCoordinator(
     tokenHelper = new TokenHelper(new SecurityConfig(conf), secretKeyClient);
     this.clientMetrics = ContainerClientMetrics.acquire();
     this.metrics = metrics;
+    ecValidator = new ECValidator(ozoneClientConfig);
   }
 
   public void reconstructECContainerGroup(long containerID,
@@ -330,8 +332,7 @@ public void reconstructECBlockGroup(BlockLocationInfo blockLocationInfo,
                     future = targetBlockStreams[i].write(bufs[i]);
                 checkFailures(targetBlockStreams[i], future);
               }
-              ECValidator validator = new ECValidator(ozoneClientConfig);
-              validator.validateBuffer(bufs[i], targetBlockStreams[i]);
+              ecValidator.validateBuffer(bufs[i], targetBlockStreams[i], i);
               bufs[i].clear();
             }
             length -= readLen;

diff --git a/...ervice/src/main/java/org/apache/hadoop/ozone/container/ec/reconstruction/ECValidator.java b/...ervice/src/main/java/org/apache/hadoop/ozone/container/ec/reconstruction/ECValidator.java
@@ -4,8 +4,10 @@
 import org.apache.hadoop.hdds.scm.OzoneClientConfig;
 import org.apache.hadoop.hdds.scm.storage.ECBlockOutputStream;
 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.common.OzoneChecksumException;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -16,19 +18,37 @@ public class ECValidator {
   private static final Logger LOG =
     LoggerFactory.getLogger(ECValidator.class);
   private Checksum checksum = null;
+  private final boolean isValidationEnabled;
 
   ECValidator(OzoneClientConfig config) {
     this.checksum = new Checksum(config.getChecksumType(), config.getBytesPerChecksum());
+    // We fetch the configuration value beforehand to avoid re-fetching on every validation call
+    isValidationEnabled = config.getEcReconstructionValidation();
   }
 
-  public boolean validateBuffer(ByteBuffer buf, ECBlockOutputStream ecBlockOutputStream)
+  /**
+   * Helper function to validate the checksum between recreated data and
+   * @param buf  A {@link ByteBuffer} instance that stores the chunk
+   * @param ecBlockOutputStream A {@link ECBlockOutputStream} instance that stores
+   *                            the reconstructed index ECBlockOutputStream
+   * @param idx Used to store the index at which data was recreated
+   * @throws OzoneChecksumException if the recreated checksum and the block checksum doesn't match
+   */
+  public void validateBuffer(ByteBuffer buf, ECBlockOutputStream ecBlockOutputStream, int idx)
       throws OzoneChecksumException{
-    try (ChunkBuffer chunk = ChunkBuffer.wrap(buf)) {
-      //Checksum will be stored in the 1st chunk and parity chunks
-      ContainerProtos.ChecksumData stripeChecksum = ecBlockOutputStream.getContainerBlockData()
-        .getChunks(0).getChecksumData();
-      ContainerProtos.ChecksumData chunkChecksum = checksum.computeChecksum(chunk).getProtoBufMessage();
-      LOG.info("Chunk Checksum: {}, Stripe Checksum: {}", chunkChecksum, stripeChecksum);
+    if (isValidationEnabled) {
+      try (ChunkBuffer chunk = ChunkBuffer.wrap(buf)) {
+        //Checksum will be stored in the 1st chunk and parity chunks
+        ContainerProtos.ChecksumData stripeChecksum = ecBlockOutputStream.getContainerBlockData()
+          .getChunks(0).getChecksumData();
+        ContainerProtos.ChecksumData chunkChecksum = checksum.computeChecksum(chunk).getProtoBufMessage();
+        if (stripeChecksum.getChecksums(idx) != chunkChecksum.getChecksums(0)) {
+          LOG.info("Checksum mismatched between recreated chunk and re-created chunk");
+          throw new OzoneChecksumException("Inconsistent checksum for re-created chunk and original chunk");
+        }
+      }
+    } else {
+      LOG.debug("Checksum validation was disabled, skipping check");
     }
   }
 }