HDDS-16405. ReplicationManagerReport should not be used for transient state

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

@@ -54,7 +54,6 @@ public class ReplicationManagerReport {
 
   private final Map<String, LongAdder> stats;
   private final Map<String, List<ContainerID>> containerSample = new ConcurrentHashMap<>();
-  private ContainerHealthState containerHealthState = ContainerHealthState.HEALTHY;
 
   public static ReplicationManagerReport fromProtobuf(
       HddsProtos.ReplicationManagerReportProto proto) {
@@ -87,7 +86,6 @@ public void increment(ContainerHealthState stat) {
 
   public void incrementAndSample(ContainerHealthState stat, ContainerInfo containerInfo) {
     incrementAndSample(stat.name(), containerInfo.containerID());
-    containerHealthState = stat;
   }
 
   public void increment(HddsProtos.LifeCycleState stat) {
@@ -106,14 +104,6 @@ public long getReportTimeStamp() {
     return reportTimeStamp;
   }
 
-  public ContainerHealthState getContainerHealthState() {
-    return containerHealthState;
-  }
-
-  public void resetContainerHealthState() {
-    this.containerHealthState = ContainerHealthState.HEALTHY;
-  }
-
   /**
    * Return a map of all stats and their value as a long.
    */

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ContainerCheckRequest.java

@@ -20,6 +20,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
+import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerReplica;
 import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
@@ -37,6 +38,7 @@ public final class ContainerCheckRequest {
   private final ReplicationManagerReport report;
   private final ReplicationQueue replicationQueue;
   private final boolean readOnly;
+  private ContainerHealthState healthState = ContainerHealthState.HEALTHY;
 
   private ContainerCheckRequest(Builder builder) {
     this.containerInfo = builder.containerInfo;
@@ -77,6 +79,15 @@ public boolean isReadOnly() {
     return readOnly;
   }
 
+  public ContainerHealthState getHealthState() {
+    return healthState;
+  }
+
+  public void setHealthState(ContainerHealthState state) {
+    this.healthState = state;
+    report.incrementAndSample(state, containerInfo);
+  }
+
   /**
    * Builder class for ContainerCheckRequest.
    */

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java

@@ -858,8 +858,6 @@ protected boolean processContainer(ContainerInfo containerInfo,
         return false;
       }
 
-      // Reset health state to HEALTHY before processing this container
-      report.resetContainerHealthState();
       final boolean isEC = isEC(containerInfo.getReplicationConfig());
 
       ContainerCheckRequest checkRequest = new ContainerCheckRequest.Builder()
@@ -878,10 +876,8 @@ protected boolean processContainer(ContainerInfo containerInfo,
       if (!handled) {
         LOG.debug("Container {} had no actions after passing through the " +
             "check chain", containerInfo.containerID());
-        // Container remains HEALTHY (set at start of loop)
       }
-      // Apply final health state from report to container
-      containerInfo.setHealthState(report.getContainerHealthState());
+      containerInfo.setHealthState(checkRequest.getHealthState());
       return handled;
     }
   }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/ClosedWithUnhealthyReplicasHandler.java

@@ -109,7 +109,7 @@ public boolean handle(ContainerCheckRequest request) {
     // some unhealthy replicas were found so the container must be
     // over replicated due to unhealthy replicas.
     if (foundUnhealthy) {
-      request.getReport().incrementAndSample(ContainerHealthState.UNHEALTHY_OVER_REPLICATED, containerInfo);
+      request.setHealthState(ContainerHealthState.UNHEALTHY_OVER_REPLICATED);
     }
     LOG.debug("Returning {} for container {}", foundUnhealthy, containerInfo);
     return foundUnhealthy;

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/ClosingContainerHandler.java

@@ -70,7 +70,7 @@ public boolean handle(ContainerCheckRequest request) {
 
     // TODO - review this logic - may need an empty check here
     if (request.getContainerReplicas().isEmpty()) {
-      request.getReport().incrementAndSample(ContainerHealthState.MISSING, containerInfo);
+      request.setHealthState(ContainerHealthState.MISSING);
     }
 
     if (request.isReadOnly()) {

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/ECMisReplicationCheckHandler.java

@@ -30,7 +30,6 @@
 import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerReplica;
-import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerCheckRequest;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerHealthResult;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerReplicaOp;
@@ -59,14 +58,13 @@ public boolean handle(ContainerCheckRequest request) {
       return false;
     }
 
-    ReplicationManagerReport report = request.getReport();
     ContainerInfo container = request.getContainerInfo();
     LOG.debug("Checking container {} for mis replication.", container);
 
     ContainerHealthResult health = checkMisReplication(request);
     if (health.getHealthState() ==
         ContainerHealthResult.HealthState.MIS_REPLICATED) {
-      report.incrementAndSample(ContainerHealthState.MIS_REPLICATED, container);
+      request.setHealthState(ContainerHealthState.MIS_REPLICATED);
       ContainerHealthResult.MisReplicatedHealthResult misRepHealth
           = ((ContainerHealthResult.MisReplicatedHealthResult) health);
       if (!misRepHealth.isReplicatedOkAfterPending()) {

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/ECReplicationCheckHandler.java

@@ -25,7 +25,6 @@
 import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerReplica;
-import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerCheckRequest;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerHealthResult;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerReplicaOp;
@@ -49,7 +48,6 @@ public boolean handle(ContainerCheckRequest request) {
       // This handler is only for EC containers.
       return false;
     }
-    ReplicationManagerReport report = request.getReport();
     ContainerInfo container = request.getContainerInfo();
     ContainerHealthResult health = checkHealth(request);
     LOG.debug("Checking container {} in ECReplicationCheckHandler", container);
@@ -84,16 +82,15 @@ public boolean handle(ContainerCheckRequest request) {
             healthState = ContainerHealthState.UNHEALTHY;
           }
         }
-        report.incrementAndSample(healthState, container);
       } else {
         healthState = ContainerHealthState.UNDER_REPLICATED;
-        report.incrementAndSample(healthState, container);
       }
       if (!underHealth.isReplicatedOkAfterPending() &&
           (!underHealth.isUnrecoverable()
               || (underHealth.hasUnreplicatedOfflineIndexes() && !underHealth.offlineIndexesOkAfterPending()))) {
         request.getReplicationQueue().enqueue(underHealth);
       }
+      request.setHealthState(healthState);
       LOG.debug("Container {} is Under Replicated. isReplicatedOkAfterPending "
           + "is [{}]. isUnrecoverable is [{}]. isMissing is [{}]. "
           + "hasUnreplicatedOfflineIndexes is [{}]",
@@ -103,7 +100,7 @@ public boolean handle(ContainerCheckRequest request) {
       return true;
     } else if (health.getHealthState()
         == ContainerHealthResult.HealthState.OVER_REPLICATED) {
-      report.incrementAndSample(ContainerHealthState.OVER_REPLICATED, container);
+      request.setHealthState(ContainerHealthState.OVER_REPLICATED);
       ContainerHealthResult.OverReplicatedHealthResult overHealth
           = ((ContainerHealthResult.OverReplicatedHealthResult) health);
       if (!overHealth.isReplicatedOkAfterPending()) {

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/EmptyContainerHandler.java

@@ -57,7 +57,7 @@ public boolean handle(ContainerCheckRequest request) {
     Set<ContainerReplica> replicas = request.getContainerReplicas();
 
     if (isContainerEmptyAndClosed(containerInfo, replicas)) {
-      request.getReport().incrementAndSample(ContainerHealthState.EMPTY, containerInfo);
+      request.setHealthState(ContainerHealthState.EMPTY);
       if (!request.isReadOnly()) {
         LOG.debug("Container {} is empty and closed, marking as DELETING",
             containerInfo);
@@ -77,7 +77,7 @@ public boolean handle(ContainerCheckRequest request) {
       }
       return true;
     } else if (isContainerEmptyAndQuasiClosed(containerInfo, replicas)) {
-      request.getReport().incrementAndSample(ContainerHealthState.EMPTY, containerInfo);
+      request.setHealthState(ContainerHealthState.EMPTY);
       if (!request.isReadOnly()) {
         String originIds = replicas.stream()
             .map(r -> r.getOriginDatanodeId().toString())
@@ -115,7 +115,7 @@ public boolean handle(ContainerCheckRequest request) {
       // information to delete the container, so we just log it as EMPTY,
       // leaving it as CLOSED and return true, otherwise, it will end up marked
       // as missing in the replication check handlers.
-      request.getReport().incrementAndSample(ContainerHealthState.EMPTY, containerInfo);
+      request.setHealthState(ContainerHealthState.EMPTY);
       LOG.debug("Container {} appears empty and is closed, but cannot be " +
               "deleted because it has no replicas. Marking as EMPTY.",
           containerInfo);

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/OpenContainerHandler.java

@@ -63,10 +63,10 @@ public boolean handle(ContainerCheckRequest request) {
         LOG.info("Container {} is open but {}. Triggering close.",
             containerInfo, noPipeline ? "has no Pipeline" : "unhealthy");
 
-        request.getReport().incrementAndSample(noPipeline ?
-                ContainerHealthState.OPEN_WITHOUT_PIPELINE :
-                ContainerHealthState.OPEN_UNHEALTHY,
-            containerInfo);
+        ContainerHealthState openHealthState = noPipeline ?
+            ContainerHealthState.OPEN_WITHOUT_PIPELINE :
+            ContainerHealthState.OPEN_UNHEALTHY;
+        request.setHealthState(openHealthState);
 
         if (!request.isReadOnly()) {
           replicationManager

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/QuasiClosedContainerHandler.java

@@ -76,7 +76,7 @@ public boolean handle(ContainerCheckRequest request) {
     } else {
       LOG.debug("Container {} cannot be force closed and is stuck in " +
               "QUASI_CLOSED", containerInfo);
-      request.getReport().incrementAndSample(ContainerHealthState.QUASI_CLOSED_STUCK, containerInfo);
+      request.setHealthState(ContainerHealthState.QUASI_CLOSED_STUCK);
     }
     // Always return false, even if commands were sent. That way, under and
     // over replication handlers can to check for other issues in the container.

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/QuasiClosedStuckReplicationCheck.java

@@ -73,8 +73,7 @@ public boolean handle(ContainerCheckRequest request) {
 
     if (request.getContainerReplicas().isEmpty()) {
       // If there are no replicas, mark as QUASI_CLOSED_STUCK + MISSING combination
-      request.getReport().incrementAndSample(ContainerHealthState.QUASI_CLOSED_STUCK_MISSING, 
-          request.getContainerInfo());
+      request.setHealthState(ContainerHealthState.QUASI_CLOSED_STUCK_MISSING);
       return true;
     }
 
@@ -101,8 +100,7 @@ public boolean handle(ContainerCheckRequest request) {
     if (replicaCount.isUnderReplicated()) {
       LOG.debug("Container {} is quasi-closed-stuck under-replicated", request.getContainerInfo());
       // Container is both QUASI_CLOSED_STUCK and UNDER_REPLICATED
-      request.getReport().incrementAndSample(ContainerHealthState.QUASI_CLOSED_STUCK_UNDER_REPLICATED, 
-          request.getContainerInfo());
+      request.setHealthState(ContainerHealthState.QUASI_CLOSED_STUCK_UNDER_REPLICATED);
       if (pendingAdd == 0) {
         // Only queue if there are no pending adds, as that could correct the under replication.
         LOG.debug("Queueing under-replicated health result for container {}", request.getContainerInfo());
@@ -117,8 +115,7 @@ public boolean handle(ContainerCheckRequest request) {
     if (replicaCount.isOverReplicated()) {
       LOG.debug("Container {} is quasi-closed-stuck over-replicated", request.getContainerInfo());
       // Container is both QUASI_CLOSED_STUCK and OVER_REPLICATED
-      request.getReport().incrementAndSample(ContainerHealthState.QUASI_CLOSED_STUCK_OVER_REPLICATED, 
-          request.getContainerInfo());
+      request.setHealthState(ContainerHealthState.QUASI_CLOSED_STUCK_OVER_REPLICATED);
       if (pendingDelete == 0) {
         // Only queue if there are no pending deletes which could correct the over replication
         LOG.debug("Queueing over-replicated health result for container {}", request.getContainerInfo());

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/RatisReplicationCheckHandler.java

@@ -31,7 +31,6 @@
 import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerReplica;
-import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerCheckRequest;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerHealthResult;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerReplicaOp;
@@ -84,7 +83,6 @@ public boolean handle(ContainerCheckRequest request) {
         .shouldHandleAsQuasiClosedStuck(request.getContainerInfo(), request.getContainerReplicas())) {
       return false;
     }
-    ReplicationManagerReport report = request.getReport();
     ContainerInfo container = request.getContainerInfo();
     ContainerHealthResult health = checkHealth(request);
     LOG.debug("Checking container {} in RatisReplicationCheckHandler",
@@ -117,10 +115,10 @@ public boolean handle(ContainerCheckRequest request) {
           underHealth.isUnrecoverable(), underHealth.hasHealthyReplicas());
 
       if (underHealth.isUnrecoverable()) {
-        report.incrementAndSample(ContainerHealthState.MISSING, container);
+        request.setHealthState(ContainerHealthState.MISSING);
         return true;
       }
-      report.incrementAndSample(ContainerHealthState.UNDER_REPLICATED, container);
+      request.setHealthState(ContainerHealthState.UNDER_REPLICATED);
 
       if (!underHealth.isReplicatedOkAfterPending() &&
           underHealth.hasHealthyReplicas()) {
@@ -138,7 +136,7 @@ public boolean handle(ContainerCheckRequest request) {
     */
     if (health.getHealthState()
         == ContainerHealthResult.HealthState.OVER_REPLICATED) {
-      report.incrementAndSample(ContainerHealthState.OVER_REPLICATED, container);
+      request.setHealthState(ContainerHealthState.OVER_REPLICATED);
       ContainerHealthResult.OverReplicatedHealthResult overHealth
           = ((ContainerHealthResult.OverReplicatedHealthResult) health);
       if (!overHealth.isReplicatedOkAfterPending() &&
@@ -166,7 +164,7 @@ public boolean handle(ContainerCheckRequest request) {
 
     if (health.getHealthState() ==
         ContainerHealthResult.HealthState.MIS_REPLICATED) {
-      report.incrementAndSample(ContainerHealthState.MIS_REPLICATED, container);
+      request.setHealthState(ContainerHealthState.MIS_REPLICATED);
       ContainerHealthResult.MisReplicatedHealthResult misRepHealth
           = ((ContainerHealthResult.MisReplicatedHealthResult) health);
       if (!misRepHealth.isReplicatedOkAfterPending()) {

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/RatisUnhealthyReplicationCheckHandler.java

@@ -22,7 +22,6 @@
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
-import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerCheckRequest;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerHealthResult;
 import org.apache.hadoop.hdds.scm.container.replication.RatisContainerReplicaCount;
@@ -52,7 +51,6 @@ public boolean handle(ContainerCheckRequest request) {
       // This handler is only for Ratis containers.
       return false;
     }
-    ReplicationManagerReport report = request.getReport();
     ContainerInfo container = request.getContainerInfo();
 
     RatisContainerReplicaCount replicaCount = getReplicaCount(request);
@@ -75,7 +73,7 @@ public boolean handle(ContainerCheckRequest request) {
       ContainerHealthResult.UnderReplicatedHealthResult underHealth
           = ((ContainerHealthResult.UnderReplicatedHealthResult) health);
       // Container is UNHEALTHY + UNDER_REPLICATED
-      report.incrementAndSample(ContainerHealthState.UNHEALTHY_UNDER_REPLICATED, container);
+      request.setHealthState(ContainerHealthState.UNHEALTHY_UNDER_REPLICATED);
       LOG.debug("Container {} is Under Replicated. isReplicatedOkAfterPending" +
               " is [{}]. isUnrecoverable is [{}]. hasHealthyReplicas is [{}].",
           container,
@@ -91,7 +89,7 @@ public boolean handle(ContainerCheckRequest request) {
     if (health.getHealthState()
         == ContainerHealthResult.HealthState.OVER_REPLICATED) {
       // Container is UNHEALTHY + OVER_REPLICATED
-      report.incrementAndSample(ContainerHealthState.UNHEALTHY_OVER_REPLICATED, container);
+      request.setHealthState(ContainerHealthState.UNHEALTHY_OVER_REPLICATED);
       ContainerHealthResult.OverReplicatedHealthResult overHealth
           = ((ContainerHealthResult.OverReplicatedHealthResult) health);
       LOG.debug("Container {} is Over Replicated. isReplicatedOkAfterPending" +
@@ -107,7 +105,7 @@ public boolean handle(ContainerCheckRequest request) {
 
     if (health.getHealthState() == ContainerHealthResult.HealthState.UNHEALTHY) {
       // Container is UNHEALTHY + SUFFICIENTLY REPLICATED
-      report.incrementAndSample(ContainerHealthState.UNHEALTHY, container);
+      request.setHealthState(ContainerHealthState.UNHEALTHY);
       LOG.debug("Container {} is sufficiently replicated with all unhealthy replicas", container);
     }
 

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/VulnerableUnhealthyReplicasHandler.java

@@ -25,7 +25,6 @@
 import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerReplica;
-import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerCheckRequest;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerHealthResult;
 import org.apache.hadoop.hdds.scm.container.replication.RatisContainerReplicaCount;
@@ -83,8 +82,7 @@ public boolean handle(ContainerCheckRequest request) {
 
     if (!vulnerableUnhealthy.isEmpty()) {
       LOG.info("Found vulnerable UNHEALTHY replicas {} for container {}.", vulnerableUnhealthy, container);
-      ReplicationManagerReport report = request.getReport();
-      report.incrementAndSample(ContainerHealthState.UNHEALTHY_UNDER_REPLICATED, container);
+      request.setHealthState(ContainerHealthState.UNHEALTHY_UNDER_REPLICATED);
       if (!request.isReadOnly()) {
         ContainerHealthResult.UnderReplicatedHealthResult underRepResult =
             replicaCount.toUnderHealthResult();