HDDS-11763. Implement container repair logic within datanodes.

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java

@@ -112,6 +112,8 @@ public class BlockInputStream extends BlockExtendedInputStream {
 
   private final Function<BlockID, BlockLocationInfo> refreshFunction;
 
+  private BlockData blockData;
+
   public BlockInputStream(
       BlockLocationInfo blockInfo,
       Pipeline pipeline,
@@ -153,7 +155,6 @@ public synchronized void initialize() throws IOException {
       return;
     }
 
-    BlockData blockData = null;
     List<ChunkInfo> chunks = null;
     IOException catchEx = null;
     do {
@@ -554,8 +555,7 @@ public long getLength() {
     return length;
   }
 
-  @VisibleForTesting
-  synchronized int getChunkIndex() {
+  public synchronized int getChunkIndex() {
     return chunkIndex;
   }
 
@@ -618,9 +618,12 @@ private void handleReadError(IOException cause) throws IOException {
     refreshBlockInfo(cause);
   }
 
-  @VisibleForTesting
   public synchronized List<ChunkInputStream> getChunkStreams() {
     return chunkStreams;
   }
 
+  public BlockData getStreamBlockData() {
+    return blockData;
+  }
+
 }

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java

@@ -747,4 +747,8 @@ public synchronized void unbuffer() {
   public ByteBuffer[] getCachedBuffers() {
     return BufferUtils.getReadOnlyByteBuffers(buffers);
   }
+
+  public ChunkInfo getChunkInfo() {
+    return chunkInfo;
+  }
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsUtils.java

@@ -882,6 +882,11 @@ public static HddsProtos.UUID toProtobuf(UUID uuid) {
         : null;
   }
 
+  /** @return Hex string representation of {@code value} */
+  public static String checksumToString(long value) {
+    return Long.toHexString(value);
+  }
+
   /**
    * Logs a warning to report that the class is not closed properly.
    */

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/ReplicationConfig.java

@@ -112,6 +112,7 @@ static HddsProtos.ReplicationFactor getLegacyFactor(
       return ((ReplicatedReplicationConfig) replicationConfig)
           .getReplicationFactor();
     }
+
     throw new UnsupportedOperationException(
         "Replication configuration of type "
             + replicationConfig.getReplicationType()

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReplicaInfo.java

@@ -17,6 +17,8 @@
 
 package org.apache.hadoop.hdds.scm.container;
 
+import static org.apache.hadoop.hdds.HddsUtils.checksumToString;
+
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.databind.JsonSerializer;
 import com.fasterxml.jackson.databind.SerializerProvider;
@@ -26,7 +28,6 @@
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 
-
 /**
  * Class which stores ContainerReplica details on the client.
  */
@@ -102,7 +103,7 @@ public long getDataChecksum() {
   private static class LongToHexJsonSerializer extends JsonSerializer<Long> {
     @Override
     public void serialize(Long value, JsonGenerator gen, SerializerProvider provider) throws IOException {
-      gen.writeString(Long.toHexString(value));
+      gen.writeString(checksumToString(value));
     }
   }
 

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/checksum/ContainerChecksumTreeManager.java

@@ -82,7 +82,9 @@ public void stop() {
    * file remains unchanged.
    * Concurrent writes to the same file are coordinated internally.
    */
-  public void writeContainerDataTree(ContainerData data, ContainerMerkleTreeWriter tree) throws IOException {
+  public ContainerProtos.ContainerChecksumInfo writeContainerDataTree(ContainerData data,
+                                                                      ContainerMerkleTreeWriter tree)
+      throws IOException {
     long containerID = data.getContainerID();
     Lock writeLock = getLock(containerID);
     writeLock.lock();
@@ -98,11 +100,13 @@ public void writeContainerDataTree(ContainerData data, ContainerMerkleTreeWriter
         checksumInfoBuilder = ContainerProtos.ContainerChecksumInfo.newBuilder();
       }
 
-      checksumInfoBuilder
+      ContainerProtos.ContainerChecksumInfo checksumInfo = checksumInfoBuilder
           .setContainerID(containerID)
-          .setContainerMerkleTree(captureLatencyNs(metrics.getCreateMerkleTreeLatencyNS(), tree::toProto));
-      write(data, checksumInfoBuilder.build());
+          .setContainerMerkleTree(captureLatencyNs(metrics.getCreateMerkleTreeLatencyNS(), tree::toProto))
+          .build();
+      write(data, checksumInfo);
       LOG.debug("Data merkle tree for container {} updated", containerID);
+      return checksumInfo;
     } finally {
       writeLock.unlock();
     }
@@ -146,33 +150,32 @@ public void markBlocksAsDeleted(KeyValueContainerData data, Collection<Long> del
     }
   }
 
-  public ContainerDiffReport diff(KeyValueContainerData thisContainer,
+  /**
+   * Compares the checksum info of the container with the peer's checksum info and returns a report of the differences.
+   * @param thisChecksumInfo The checksum info of the container on this datanode.
+   * @param peerChecksumInfo The checksum info of the container on the peer datanode.
+   */
+  public ContainerDiffReport diff(ContainerProtos.ContainerChecksumInfo thisChecksumInfo,
                                   ContainerProtos.ContainerChecksumInfo peerChecksumInfo) throws
       StorageContainerException {
 
     ContainerDiffReport report = new ContainerDiffReport();
     try {
       captureLatencyNs(metrics.getMerkleTreeDiffLatencyNS(), () -> {
-        Preconditions.assertNotNull(thisContainer, "Container data is null");
-        Preconditions.assertNotNull(peerChecksumInfo, "Peer checksum info is null");
-        Optional<ContainerProtos.ContainerChecksumInfo> thisContainerChecksumInfo = read(thisContainer);
-        if (!thisContainerChecksumInfo.isPresent()) {
-          throw new StorageContainerException("The container #" + thisContainer.getContainerID() +
-              " doesn't have container checksum", ContainerProtos.Result.IO_EXCEPTION);
-        }
-
-        if (thisContainer.getContainerID() != peerChecksumInfo.getContainerID()) {
-          throw new StorageContainerException("Container Id does not match for container "
-              + thisContainer.getContainerID(), ContainerProtos.Result.CONTAINER_ID_MISMATCH);
+        Preconditions.assertNotNull(thisChecksumInfo, "Datanode's checksum info is null.");
+        Preconditions.assertNotNull(peerChecksumInfo, "Peer checksum info is null.");
+        if (thisChecksumInfo.getContainerID() != peerChecksumInfo.getContainerID()) {
+          throw new StorageContainerException("Container ID does not match. Local container ID "
+              + thisChecksumInfo.getContainerID() + " , Peer container ID " + peerChecksumInfo.getContainerID(),
+              ContainerProtos.Result.CONTAINER_ID_MISMATCH);
         }
 
-        ContainerProtos.ContainerChecksumInfo thisChecksumInfo = thisContainerChecksumInfo.get();
         compareContainerMerkleTree(thisChecksumInfo, peerChecksumInfo, report);
       });
     } catch (IOException ex) {
       metrics.incrementMerkleTreeDiffFailures();
-      throw new StorageContainerException("Container Diff failed for container #" + thisContainer.getContainerID(), ex,
-          ContainerProtos.Result.IO_EXCEPTION);
+      throw new StorageContainerException("Container Diff failed for container #" + thisChecksumInfo.getContainerID(),
+          ex, ContainerProtos.Result.IO_EXCEPTION);
     }
 
     // Update Container Diff metrics based on the diff report.
@@ -314,7 +317,7 @@ private Lock getLock(long containerID) {
    * Callers are not required to hold a lock while calling this since writes are done to a tmp file and atomically
    * swapped into place.
    */
-  private Optional<ContainerProtos.ContainerChecksumInfo> read(ContainerData data) throws IOException {
+  public Optional<ContainerProtos.ContainerChecksumInfo> read(ContainerData data) throws IOException {
     long containerID = data.getContainerID();
     File checksumFile = getContainerChecksumFile(data);
     try {
@@ -361,6 +364,8 @@ private void write(ContainerData data, ContainerProtos.ContainerChecksumInfo che
       throw new IOException("Error occurred when writing container merkle tree for containerID "
           + data.getContainerID(), ex);
     }
+    // Set in-memory data checksum.
+    data.setDataChecksum(checksumInfo.getContainerMerkleTree().getDataChecksum());
   }
 
   /**

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/checksum/DNContainerOperationClient.java

@@ -84,6 +84,10 @@ public XceiverClientManager getXceiverClientManager() {
     return xceiverClientManager;
   }
 
+  public TokenHelper getTokenHelper() {
+    return tokenHelper;
+  }
+
   /**
    * Reads {@link ContainerProtos.ContainerChecksumInfo} for a specified container for the specified datanode.
    *

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerMetrics.java

@@ -56,6 +56,8 @@ public class ContainerMetrics implements Closeable {
   @Metric private MutableCounterLong containerForceDelete;
   @Metric private MutableCounterLong numReadStateMachine;
   @Metric private MutableCounterLong bytesReadStateMachine;
+  @Metric private MutableCounterLong numContainerReconciledWithoutChanges;
+  @Metric private MutableCounterLong numContainerReconciledWithChanges;
 
 
   private final EnumMap<ContainerProtos.Type, MutableCounterLong> numOpsArray;
@@ -172,4 +174,12 @@ public void incBytesReadStateMachine(long bytes) {
   public long getBytesReadStateMachine() {
     return bytesReadStateMachine.value();
   }
+
+  public void incContainerReconciledWithoutChanges() {
+    numContainerReconciledWithoutChanges.incr();
+  }
+
+  public void incContainerReconciledWithChanges() {
+    numContainerReconciledWithChanges.incr();
+  }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/ContainerLogger.java

@@ -17,7 +17,10 @@
 
 package org.apache.hadoop.ozone.container.common.utils;
 
+import static org.apache.hadoop.hdds.HddsUtils.checksumToString;
+
 import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.ozone.container.common.impl.ContainerData;
 import org.apache.hadoop.ozone.container.common.interfaces.ScanResult;
 import org.apache.logging.log4j.LogManager;
@@ -145,6 +148,23 @@ public static void logRecovered(ContainerData containerData) {
     LOG.info(getMessage(containerData));
   }
 
+  /**
+   * Logged when a container is reconciled.
+   *
+   * @param containerData The container that was reconciled on this datanode.
+   * @param oldDataChecksum The old data checksum.
+   */
+  public static void logReconciled(ContainerData containerData, long oldDataChecksum, DatanodeDetails peer) {
+    if (containerData.getDataChecksum() == oldDataChecksum) {
+      LOG.info(getMessage(containerData, "Container reconciled with peer " + peer.toString() +
+          ". No change in checksum."));
+    } else {
+      LOG.warn(getMessage(containerData, "Container reconciled with peer " + peer.toString() +
+          ". Checksum updated from " + checksumToString(oldDataChecksum) + " to "
+          + checksumToString(containerData.getDataChecksum())));
+    }
+  }
+
   private static String getMessage(ContainerData containerData,
                                    String message) {
     return String.join(FIELD_SEPARATOR, getMessage(containerData), message);
@@ -155,6 +175,7 @@ private static String getMessage(ContainerData containerData) {
         "ID=" + containerData.getContainerID(),
         "Index=" + containerData.getReplicaIndex(),
         "BCSID=" + containerData.getBlockCommitSequenceId(),
-        "State=" + containerData.getState());
+        "State=" + containerData.getState(),
+        "DataChecksum=" + checksumToString(containerData.getDataChecksum()));
   }
 }