HDDS-12865. Handle automatic container cache refresh for replica verification.

hadoop-ozone/cli-debug/src/main/java/org/apache/hadoop/ozone/debug/replicas/BlockExistenceVerifier.java

@@ -24,6 +24,7 @@
 import org.apache.hadoop.hdds.scm.XceiverClientManager;
 import org.apache.hadoop.hdds.scm.XceiverClientSpi;
 import org.apache.hadoop.hdds.scm.cli.ContainerOperationClient;
+import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
@@ -67,6 +68,11 @@ public BlockVerificationResult verifyBlock(DatanodeDetails datanode, OmKeyLocati
       } else {
         return BlockVerificationResult.failCheck("Block does not exist on this replica");
       }
+    } catch (StorageContainerException e) {
+      if (e.getResult() == ContainerProtos.Result.NO_SUCH_BLOCK) {
+        return BlockVerificationResult.failCheckAndRefreshKeyLocation(e.getMessage());
+      }
+      return BlockVerificationResult.failIncomplete(e.getMessage());
     } catch (IOException e) {
       return BlockVerificationResult.failIncomplete(e.getMessage());
     } finally {

hadoop-ozone/cli-debug/src/main/java/org/apache/hadoop/ozone/debug/replicas/BlockVerificationResult.java

@@ -28,11 +28,18 @@ public class BlockVerificationResult {
   private final boolean completed;
   private final boolean pass;
   private final List<String> failures;
+  private final boolean refreshKeyLocation;
 
   public BlockVerificationResult(boolean completed, boolean pass, List<String> failures) {
+    this(completed, pass, failures, false);
+  }
+
+  public BlockVerificationResult(boolean completed, boolean pass, List<String> failures,
+      boolean refreshKeyLocation) {
     this.completed = completed;
     this.pass = pass;
     this.failures = failures;
+    this.refreshKeyLocation = refreshKeyLocation;
   }
 
   public static BlockVerificationResult pass() {
@@ -43,6 +50,10 @@ public static BlockVerificationResult failCheck(String message) {
     return new BlockVerificationResult(true, false, Collections.singletonList(message));
   }
 
+  public static BlockVerificationResult failCheckAndRefreshKeyLocation(String message) {
+    return new BlockVerificationResult(true, false, Collections.singletonList(message), true);
+  }
+
   public static BlockVerificationResult failIncomplete(String message) {
     return new BlockVerificationResult(false, false, Collections.singletonList(message));
   }
@@ -59,4 +70,8 @@ public List<String> getFailures() {
     return failures;
   }
 
+  public boolean shouldRefreshKeyLocation() {
+    return refreshKeyLocation;
+  }
+
 }

hadoop-ozone/cli-debug/src/main/java/org/apache/hadoop/ozone/debug/replicas/ChecksumVerifier.java

@@ -23,9 +23,11 @@
 import org.apache.commons.io.output.NullOutputStream;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.scm.OzoneClientConfig;
 import org.apache.hadoop.hdds.scm.XceiverClientManager;
 import org.apache.hadoop.hdds.scm.cli.ContainerOperationClient;
+import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.ozone.client.io.BlockInputStreamFactoryImpl;
 import org.apache.hadoop.ozone.common.OzoneChecksumException;
@@ -68,11 +70,27 @@ public BlockVerificationResult verifyBlock(DatanodeDetails datanode, OmKeyLocati
       return BlockVerificationResult.pass();
     } catch (IOException e) {
       Throwable cause = e.getCause() != null ? e.getCause() : e;
+      StorageContainerException storageContainerException = findStorageContainerException(e);
+      if (storageContainerException != null
+          && storageContainerException.getResult() == ContainerProtos.Result.NO_SUCH_BLOCK) {
+        return BlockVerificationResult.failCheckAndRefreshKeyLocation(storageContainerException.getMessage());
+      }
       if (cause instanceof OzoneChecksumException) {
         return BlockVerificationResult.failCheck(cause.getMessage());
       } else {
         return BlockVerificationResult.failIncomplete(cause.getMessage());
       }
     }
   }
+
+  private static StorageContainerException findStorageContainerException(Throwable throwable) {
+    Throwable current = throwable;
+    while (current != null) {
+      if (current instanceof StorageContainerException) {
+        return (StorageContainerException) current;
+      }
+      current = current.getCause();
+    }
+    return null;
+  }
 }

hadoop-ozone/cli-debug/src/main/java/org/apache/hadoop/ozone/debug/replicas/ReplicasVerify.java

@@ -338,22 +338,79 @@ void checkBucket(OzoneClient ozoneClient, OzoneBucket bucket, ArrayNode keysArra
   void processKey(OzoneClient ozoneClient, String volumeName, String bucketName, String keyName,
       ArrayNode keysArray, AtomicBoolean allKeysPassed) throws IOException {
     keysProcessed.incrementAndGet();
-    OmKeyInfo keyInfo = ozoneClient.getProxy().getKeyInfo(
-        volumeName, bucketName, keyName, refreshContainerLocationsFromScm);
+    OmKeyInfo keyInfo;
+    try {
+      keyInfo = ozoneClient.getProxy().getKeyInfo(
+          volumeName, bucketName, keyName, refreshContainerLocationsFromScm);
+    } catch (IOException e) {
+      LOG.warn("Unable to fetch key info from OM for key {}/{}/{}; marking verification incomplete.",
+          volumeName, bucketName, keyName, e);
+      markKeyFetchFailure(volumeName, bucketName, keyName, e.getMessage(), keysArray, allKeysPassed);
+      return;
+    }
 
     // Check if key should be processed based on replication config
     if (!shouldProcessKeyByReplicationType(keyInfo)) {
       return;
     }
 
+    KeyVerificationResult keyVerificationResult = verifyKey(keyInfo, volumeName, bucketName, keyName);
+    if (keyVerificationResult.shouldRefreshKeyLocation()) {
+      try {
+        OmKeyInfo refreshedKeyInfo = ozoneClient.getProxy().getKeyInfo(
+            volumeName, bucketName, keyName, true);
+        keyVerificationResult = verifyKey(refreshedKeyInfo, volumeName, bucketName, keyName);
+      } catch (IOException e) {
+        LOG.warn("Unable to refresh key location from OM for key {}/{}/{}; marking verification incomplete.",
+            volumeName, bucketName, keyName, e);
+        keyVerificationResult.markRefreshFailure(e.getMessage());
+      }
+    }
+
+    keyVerificationResult.getKeyNode().put("pass", keyVerificationResult.passed());
+    if (keyVerificationResult.passed()) {
+      keysPassed.incrementAndGet();
+    } else {
+      keysFailed.incrementAndGet();
+      allKeysPassed.set(false);
+      keyVerificationResult.getFailedVerificationTypes().forEach(failedType -> failuresByType
+          .computeIfAbsent(failedType, k -> new AtomicInteger(0))
+          .incrementAndGet()
+      );
+    }
+
+    if (!keyVerificationResult.passed() || allResults) {
+      keysArray.add(keyVerificationResult.getKeyNode());
+    }
+  }
+
+  private void markKeyFetchFailure(String volumeName, String bucketName, String keyName, String message,
+      ArrayNode keysArray, AtomicBoolean allKeysPassed) {
+    ObjectNode keyNode = JsonUtils.createObjectNode(null);
+    keyNode.put("volumeName", volumeName);
+    keyNode.put("bucketName", bucketName);
+    keyNode.put("name", keyName);
+    keyNode.putArray("blocks");
+    keyNode.put("completed", false);
+    keyNode.put("pass", false);
+    keyNode.putArray("failures").addObject().put("message", "Failed to fetch key info from OM: " + message);
+    keysFailed.incrementAndGet();
+    allKeysPassed.set(false);
+    keysArray.add(keyNode);
+  }
+
+  private KeyVerificationResult verifyKey(
+      OmKeyInfo keyInfo, String volumeName, String bucketName, String keyName) {
     ObjectNode keyNode = JsonUtils.createObjectNode(null);
     keyNode.put("volumeName", volumeName);
     keyNode.put("bucketName", bucketName);
     keyNode.put("name", keyName);
 
     ArrayNode blocksArray = keyNode.putArray("blocks");
     boolean keyPass = true;
+    boolean shouldRefreshKeyLocation = false;
     Set<String> failedVerificationTypes = new HashSet<>();
+    List<ObjectNode> refreshChecks = new ArrayList<>();
 
     for (OmKeyLocationInfo keyLocation : keyInfo.getLatestVersionLocations().getBlocksLatestVersionOnly()) {
       long containerID = keyLocation.getContainerID();
@@ -383,6 +440,10 @@ void processKey(OzoneClient ozoneClient, String volumeName, String bucketName, S
           checkNode.put("type", verifier.getType());
           checkNode.put("completed", result.isCompleted());
           checkNode.put("pass", result.passed());
+          if (result.shouldRefreshKeyLocation()) {
+            shouldRefreshKeyLocation = true;
+            refreshChecks.add(checkNode);
+          }
 
           ArrayNode failuresArray = checkNode.putArray("failures");
           for (String failure : result.getFailures()) {
@@ -406,20 +467,48 @@ void processKey(OzoneClient ozoneClient, String volumeName, String bucketName, S
       }
     }
 
-    keyNode.put("pass", keyPass);
-    if (keyPass) {
-      keysPassed.incrementAndGet();
-    } else {
-      keysFailed.incrementAndGet();
-      allKeysPassed.set(false);
-      failedVerificationTypes.forEach(failedType -> failuresByType
-          .computeIfAbsent(failedType, k -> new AtomicInteger(0))
-          .incrementAndGet()
-      );
+    return new KeyVerificationResult(
+        keyNode, keyPass, failedVerificationTypes, shouldRefreshKeyLocation, refreshChecks);
+  }
+
+  private static final class KeyVerificationResult {
+    private final ObjectNode keyNode;
+    private final boolean pass;
+    private final Set<String> failedVerificationTypes;
+    private final boolean refreshKeyLocation;
+    private final List<ObjectNode> refreshChecks;
+
+    private KeyVerificationResult(ObjectNode keyNode, boolean pass, Set<String> failedVerificationTypes,
+        boolean refreshKeyLocation, List<ObjectNode> refreshChecks) {
+      this.keyNode = keyNode;
+      this.pass = pass;
+      this.failedVerificationTypes = failedVerificationTypes;
+      this.refreshKeyLocation = refreshKeyLocation;
+      this.refreshChecks = refreshChecks;
+    }
+
+    private ObjectNode getKeyNode() {
+      return keyNode;
+    }
+
+    private boolean passed() {
+      return pass;
     }
 
-    if (!keyPass || allResults) {
-      keysArray.add(keyNode);
+    private Set<String> getFailedVerificationTypes() {
+      return failedVerificationTypes;
+    }
+
+    private boolean shouldRefreshKeyLocation() {
+      return refreshKeyLocation;
+    }
+
+    private void markRefreshFailure(String message) {
+      String refreshFailure = "Failed to refresh key location from OM: " + message;
+      for (ObjectNode checkNode : refreshChecks) {
+        checkNode.put("completed", false);
+        checkNode.withArray("failures").addObject().put("message", refreshFailure);
+      }
     }
   }