HDDS-11763. Implement container repair logic within datanodes.

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/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

@@ -314,7 +314,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 {

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/statemachine/DatanodeStateMachine.java

@@ -132,6 +132,7 @@ public class DatanodeStateMachine implements Closeable {
 
   private final DatanodeQueueMetrics queueMetrics;
   private final ReconfigurationHandler reconfigurationHandler;
+  private final DNContainerOperationClient dnClient;
   /**
    * Constructs a datanode state machine.
    * @param datanodeDetails - DatanodeDetails used to identify a datanode
@@ -229,7 +230,7 @@ public DatanodeStateMachine(HddsDatanodeService hddsDatanodeService,
 
     // TODO HDDS-11218 combine the clients used for reconstruction and reconciliation so they share the same cache of
     //  datanode clients.
-    DNContainerOperationClient dnClient = new DNContainerOperationClient(conf, certClient, secretKeyClient);
+    dnClient = new DNContainerOperationClient(conf, certClient, secretKeyClient);
 
     ThreadFactory threadFactory = new ThreadFactoryBuilder()
         .setNameFormat(threadNamePrefix + "PipelineCommandHandlerThread-%d")
@@ -752,4 +753,9 @@ public DatanodeQueueMetrics getQueueMetrics() {
   public ReconfigurationHandler getReconfigurationHandler() {
     return reconfigurationHandler;
   }
+
+  @VisibleForTesting
+  public DNContainerOperationClient getDnContainerOperationClientClient() {
+    return dnClient;
+  }
 }

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

@@ -17,6 +17,8 @@
 
 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.ozone.container.common.impl.ContainerData;
 import org.apache.hadoop.ozone.container.common.interfaces.ScanResult;
@@ -145,6 +147,16 @@ 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) {
+    LOG.info(getMessage(containerData, "Container reconciled. Old checksum is " + checksumToString(oldDataChecksum)));
+  }
+
   private static String getMessage(ContainerData containerData,
                                    String message) {
     return String.join(FIELD_SEPARATOR, getMessage(containerData), message);
@@ -155,6 +167,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()));
   }
 }