HDDS-11389. Incorrect number of deleted containers shown in Recon UI.

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/fsck/ContainerHealthStatus.java

@@ -160,6 +160,10 @@ public boolean isEmpty() {
     return numKeys == 0;
   }
 
+  public boolean isEmptyMissing() {
+    return numKeys == 0 && numReplicas == 0;
+  }
+
   private ContainerPlacementStatus getPlacementStatus(
       PlacementPolicy policy, int repFactor) {
     List<DatanodeDetails> dns = healthyReplicas.stream()

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/fsck/ContainerHealthTask.java

@@ -256,6 +256,11 @@ private void completeProcessingContainer(
    * completeProcessingContainer is called. This will check to see if any
    * additional records need to be added to the database.
    *
+   * If a container is identified as missing, empty-missing, under-replicated,
+   * over-replicated or mis-replicated, the method checks with SCM to determine
+   * if it has been deleted, using {@code containerDeletedInSCM}. If the container is
+   * deleted in SCM, the corresponding record is removed from Recon.
+   *
    * @param currentTime Timestamp to place on all records generated by this run
    * @param unhealthyContainerStateCountMap
    * @return Count of records processed
@@ -273,34 +278,44 @@ private long processExistingDBRecords(long currentTime,
         recordCount++;
         UnhealthyContainersRecord rec = cursor.fetchNext();
         try {
+          // Set the current container if it's not already set
           if (currentContainer == null) {
             currentContainer = setCurrentContainer(rec.getContainerId());
           }
+          // If the container ID has changed, finish processing the previous one
           if (currentContainer.getContainerID() != rec.getContainerId()) {
             completeProcessingContainer(
                 currentContainer, existingRecords, currentTime,
                 unhealthyContainerStateCountMap);
             existingRecords.clear();
             currentContainer = setCurrentContainer(rec.getContainerId());
           }
-          if (ContainerHealthRecords
-              .retainOrUpdateRecord(currentContainer, rec
-              )) {
-            // Check if the missing container is deleted in SCM
-            if (currentContainer.isMissing() &&
-                containerDeletedInSCM(currentContainer.getContainer())) {
-              rec.delete();
-            }
-            existingRecords.add(rec.getContainerState());
-            if (rec.changed()) {
-              rec.update();
-            }
-          } else {
+
+          //  Unhealthy Containers such as MISSING, EMPTY_MISSING, UNDER_REPLICATED,
+          // OVER_REPLICATED, MIS_REPLICATED can have their unhealthy states changed or retained.
+          if (!ContainerHealthRecords.retainOrUpdateRecord(currentContainer, rec)) {
             LOG.info("DELETED existing unhealthy container record...for Container: {}",
                 currentContainer.getContainerID());
             rec.delete();
+            continue;
+          }
+
+          // If the container is marked as MISSING and it's deleted in SCM, remove the record
+          if (currentContainer.isMissing() && containerDeletedInSCM(currentContainer.getContainer())) {
+            rec.delete();
+          }
+
+          // If the container is in the EMPTY_MISSING state, delete the record
+          if (currentContainer.isEmptyMissing()) {
+            rec.delete();
+          }
+          existingRecords.add(rec.getContainerState());
+          // If the record was changed, update it
+          if (rec.changed()) {
+            rec.update();
           }
         } catch (ContainerNotFoundException cnf) {
+          // If the container is not found, delete the record and reset currentContainer
           rec.delete();
           currentContainer = null;
         }
@@ -349,6 +364,18 @@ private void processContainer(ContainerInfo container, long currentTime,
     }
   }
 
+  /**
+   * Ensures the container's state in Recon is updated to match its state in SCM.
+   *
+   * If SCM reports the container as DELETED, this method attempts to transition
+   * the container's state in Recon from CLOSED to DELETING, or from DELETING to
+   * DELETED, based on the current state in Recon. It logs each transition attempt
+   * and handles any exceptions that may occur.
+   *
+   * @param containerInfo the container whose state is being checked and potentially updated.
+   * @return {@code true} if the container was found to be DELETED in SCM and the
+   *         state transition was attempted in Recon; {@code false} otherwise.
+   */
   private boolean containerDeletedInSCM(ContainerInfo containerInfo) {
     try {
       ContainerWithPipeline containerWithPipeline =
@@ -358,13 +385,16 @@ private boolean containerDeletedInSCM(ContainerInfo containerInfo) {
         if (containerInfo.getState() == HddsProtos.LifeCycleState.CLOSED) {
           containerManager.updateContainerState(containerInfo.containerID(),
               HddsProtos.LifeCycleEvent.DELETE);
+          LOG.debug("Successfully changed container {} state from CLOSED to DELETING.",
+              containerInfo.containerID());
         }
         if (containerInfo.getState() == HddsProtos.LifeCycleState.DELETING &&
             containerManager.getContainerReplicas(containerInfo.containerID())
                 .size() == 0
         ) {
           containerManager.updateContainerState(containerInfo.containerID(),
               HddsProtos.LifeCycleEvent.CLEANUP);
+          LOG.info("Successfully Deleted container {} from Recon.", containerInfo.containerID());
         }
         return true;
       }
@@ -435,6 +465,9 @@ public static boolean retainOrUpdateRecord(
       case MISSING:
         returnValue = container.isMissing() && !container.isEmpty();
         break;
+      case EMPTY_MISSING:
+        returnValue = container.isMissing() && container.isEmpty();
+        break;
       case MIS_REPLICATED:
         returnValue = keepMisReplicatedRecord(container, rec);
         break;
@@ -495,7 +528,7 @@ public static List<UnhealthyContainers> generateUnhealthyRecords(
 
           LOG.debug("Empty container {} is missing. Kindly check the " +
               "consolidated container stats per UNHEALTHY state logged as " +
-              "starting with **Container State Stats:**");
+              "starting with **Container State Stats:**", container.getContainerID());
 
           records.add(
               recordForState(container, EMPTY_MISSING,

hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/fsck/TestContainerHealthTask.java

@@ -21,6 +21,8 @@
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.THREE;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.hadoop.ozone.recon.schema.ContainerSchemaDefinition.UnHealthyContainerStates.ALL_REPLICAS_BAD;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.times;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
@@ -31,6 +33,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+
 import java.io.IOException;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -530,6 +533,106 @@ public void testNegativeSizeContainers() throws Exception {
     assertThat(negativeSizeContainers).hasSize(3);
   }
 
+  @Test
+  public void testMissingAndEmptyMissingContainerDeletion() throws Exception {
+    // Setup mock DAOs and managers
+    UnhealthyContainersDao unHealthyContainersTableHandle =
+        getDao(UnhealthyContainersDao.class);
+    ContainerHealthSchemaManager containerHealthSchemaManager =
+        new ContainerHealthSchemaManager(
+            getSchemaDefinition(ContainerSchemaDefinition.class),
+            unHealthyContainersTableHandle);
+    ReconStorageContainerManagerFacade scmMock =
+        mock(ReconStorageContainerManagerFacade.class);
+    MockPlacementPolicy placementMock = new MockPlacementPolicy();
+    ContainerManager containerManagerMock = mock(ContainerManager.class);
+    StorageContainerServiceProvider scmClientMock =
+        mock(StorageContainerServiceProvider.class);
+    ReconContainerMetadataManager reconContainerMetadataManager =
+        mock(ReconContainerMetadataManager.class);
+    mock(ReconContainerMetadataManager.class);
+
+    // Create 2 containers. They start in CLOSED state in Recon.
+    List<ContainerInfo> mockContainers = getMockContainers(2);
+    when(scmMock.getScmServiceProvider()).thenReturn(scmClientMock);
+    when(scmMock.getContainerManager()).thenReturn(containerManagerMock);
+    when(containerManagerMock.getContainers(any(ContainerID.class),
+        anyInt())).thenReturn(mockContainers);
+
+    // Mark both containers as initially CLOSED in Recon
+    for (ContainerInfo c : mockContainers) {
+      when(containerManagerMock.getContainer(c.containerID())).thenReturn(c);
+    }
+
+    // Simulate SCM reporting the containers as DELETED
+    ContainerInfo deletedContainer1 = getMockDeletedContainer(1);
+    ContainerInfo deletedContainer2 = getMockDeletedContainer(2);
+
+    when(scmClientMock.getContainerWithPipeline(1))
+        .thenReturn(new ContainerWithPipeline(deletedContainer1, null));
+    when(scmClientMock.getContainerWithPipeline(2))
+        .thenReturn(new ContainerWithPipeline(deletedContainer2, null));
+
+    // Both containers start as CLOSED in Recon (MISSING or EMPTY_MISSING)
+    when(containerManagerMock.getContainer(ContainerID.valueOf(1L)).getState())
+        .thenReturn(HddsProtos.LifeCycleState.CLOSED);
+    when(containerManagerMock.getContainer(ContainerID.valueOf(2L)).getState())
+        .thenReturn(HddsProtos.LifeCycleState.CLOSED);
+
+    // Replicas are empty, so both containers should be considered for deletion
+    when(containerManagerMock.getContainerReplicas(ContainerID.valueOf(1L)))
+        .thenReturn(Collections.emptySet());
+    when(containerManagerMock.getContainerReplicas(ContainerID.valueOf(2L)))
+        .thenReturn(Collections.emptySet());
+
+    // Initialize UnhealthyContainers in DB (MISSING and EMPTY_MISSING)
+    // Create and set up the first UnhealthyContainer for a MISSING container
+    UnhealthyContainers container1 = new UnhealthyContainers();
+    container1.setContainerId(1L);
+    container1.setContainerState("MISSING");
+    container1.setExpectedReplicaCount(3);
+    container1.setActualReplicaCount(0);
+    container1.setReplicaDelta(3);
+    container1.setInStateSince(System.currentTimeMillis());
+
+    // Create and set up the second UnhealthyContainer for an EMPTY_MISSING container
+    UnhealthyContainers container2 = new UnhealthyContainers();
+    container2.setContainerId(2L);
+    container2.setContainerState("EMPTY_MISSING");
+    container2.setExpectedReplicaCount(3);
+    container2.setActualReplicaCount(0);
+    container2.setReplicaDelta(3);
+    container2.setInStateSince(System.currentTimeMillis());
+
+    unHealthyContainersTableHandle.insert(container1);
+    unHealthyContainersTableHandle.insert(container2);
+
+    when(reconContainerMetadataManager.getKeyCountForContainer(1L)).thenReturn(5L);
+    when(reconContainerMetadataManager.getKeyCountForContainer(2L)).thenReturn(0L);
+
+    // Start the container health task
+    ReconTaskStatusDao reconTaskStatusDao = getDao(ReconTaskStatusDao.class);
+    ReconTaskConfig reconTaskConfig = new ReconTaskConfig();
+    reconTaskConfig.setMissingContainerTaskInterval(Duration.ofSeconds(2));
+    ContainerHealthTask containerHealthTask =
+        new ContainerHealthTask(scmMock.getContainerManager(),
+            scmMock.getScmServiceProvider(),
+            reconTaskStatusDao, containerHealthSchemaManager,
+            placementMock, reconTaskConfig,
+            reconContainerMetadataManager, new OzoneConfiguration());
+
+    containerHealthTask.start();
+
+    // Wait for the task to complete and ensure that updateContainerState is invoked for
+    // container IDs 1 and 2 to mark the containers as DELETED, since they are DELETED in SCM.
+    LambdaTestUtils.await(60000, 1000, () -> {
+      verify(containerManagerMock, times(1))
+          .updateContainerState(ContainerID.valueOf(1L), HddsProtos.LifeCycleEvent.DELETE);
+      verify(containerManagerMock, times(1))
+          .updateContainerState(ContainerID.valueOf(2L), HddsProtos.LifeCycleEvent.DELETE);
+      return true;
+    });
+  }
 
   private Set<ContainerReplica> getMockReplicas(
       long containerId, State...states) {