HDDS-12421. ContainerReportHandler should not make the call to delete replicas

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

@@ -245,7 +245,7 @@ private boolean updateContainerState(final DatanodeDetails datanode,
 
     final ContainerID containerId = container.containerID();
     boolean ignored = false;
-
+    boolean replicaIsEmpty = replica.hasIsEmpty() && replica.getIsEmpty();
     switch (container.getState()) {
     case OPEN:
       /*
@@ -351,27 +351,29 @@ private boolean updateContainerState(final DatanodeDetails datanode,
        * The container is already in closed state. do nothing.
        */
       break;
+    case DELETED:
+      // If container is in DELETED state and the reported replica is empty, delete the empty replica.
+      // We should also do this for DELETING containers and currently DeletingContainerHandler does that
+      if (replicaIsEmpty) {
+        deleteReplica(containerId, datanode, publisher, "DELETED", false);
+        break;
+      }
     case DELETING:
       /*
-      HDDS-11136: If a DELETING container has a non-empty CLOSED replica, the container should be moved back to CLOSED
-      state.
+       * HDDS-11136: If a DELETING container has a non-empty CLOSED replica, the container should be moved back to
+       * CLOSED state.
+       *
+       * HDDS-12421: If a DELETING or DELETED container has a non-empty replica, the container should also be moved
+       * back to CLOSED state.
        */
-      boolean replicaStateAllowed = replica.getState() == State.CLOSED;
-      boolean replicaNotEmpty = replica.hasIsEmpty() && !replica.getIsEmpty();
-      if (replicaStateAllowed && replicaNotEmpty) {
-        logger.info("Moving DELETING container {} to CLOSED state, datanode {} reported replica with state={}, " +
-            "isEmpty={} and keyCount={}.", containerId, datanode.getHostName(), replica.getState(), false,
-            replica.getKeyCount());
-        containerManager.transitionDeletingToClosedState(containerId);
+      boolean replicaStateAllowed = (replica.getState() != State.INVALID && replica.getState() != State.DELETED);
+      if (!replicaIsEmpty && replicaStateAllowed) {
+        logger.info("Moving container {} from {} to CLOSED state, datanode {} reported replica with state={}, " +
+            "isEmpty={}, bcsId={}, keyCount={}, and origin={}",
+            container, container.getState(), datanode.getHostName(), replica.getState(),
+            replica.getIsEmpty(), replica.getBlockCommitSequenceId(), replica.getKeyCount(), replica.getOriginNodeId());
+        containerManager.transitionDeletingOrDeletedToClosedState(containerId);
       }
-
-      break;
-    case DELETED:
-      /*
-       * The container is deleted. delete the replica.
-       */
-      deleteReplica(containerId, datanode, publisher, "DELETED");
-      ignored = true;
       break;
     default:
       break;
@@ -453,9 +455,9 @@ protected ContainerManager getContainerManager() {
   }
 
   protected void deleteReplica(ContainerID containerID, DatanodeDetails dn,
-      EventPublisher publisher, String reason) {
+      EventPublisher publisher, String reason, boolean force) {
     SCMCommand<?> command = new DeleteContainerCommand(
-        containerID.getId(), true);
+        containerID.getId(), force);
     try {
       command.setTerm(scmContext.getTermOfLeader());
     } catch (NotLeaderException nle) {

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

@@ -135,14 +135,14 @@ void updateContainerState(ContainerID containerID,
       throws IOException, InvalidStateTransitionException;
 
   /**
-   * Bypasses the container state machine to change a container's state from DELETING to CLOSED. This API was
+   * Bypasses the container state machine to change a container's state from DELETING or DELETED to CLOSED. This API was
    * introduced to fix a bug (HDDS-11136), and should be used with care otherwise.
    *
    * @see <a href="https://issues.apache.org/jira/browse/HDDS-11136">HDDS-11136</a>
    * @param containerID id of the container to transition
    * @throws IOException
    */
-  void transitionDeletingToClosedState(ContainerID containerID) throws IOException;
+  void transitionDeletingOrDeletedToClosedState(ContainerID containerID) throws IOException;
 
   /**
    * Returns the latest list of replicas for given containerId.

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

@@ -290,12 +290,12 @@ public void updateContainerState(final ContainerID cid,
   }
 
   @Override
-  public void transitionDeletingToClosedState(ContainerID containerID) throws IOException {
+  public void transitionDeletingOrDeletedToClosedState(ContainerID containerID) throws IOException {
     HddsProtos.ContainerID proto = containerID.getProtobuf();
     lock.lock();
     try {
       if (containerExist(containerID)) {
-        containerStateManager.transitionDeletingToClosedState(proto);
+        containerStateManager.transitionDeletingOrDeletedToClosedState(proto);
       } else {
         throwContainerNotFoundException(containerID);
       }

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

@@ -234,7 +234,7 @@ private void processSingleReplica(final DatanodeDetails datanodeDetails,
           UNKNOWN_CONTAINER_ACTION_DELETE)) {
         final ContainerID containerId = ContainerID
             .valueOf(replicaProto.getContainerID());
-        deleteReplica(containerId, datanodeDetails, publisher, "unknown");
+        deleteReplica(containerId, datanodeDetails, publisher, "unknown", true);
       }
       return;
     }

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

@@ -159,15 +159,15 @@ void updateContainerState(HddsProtos.ContainerID id,
 
 
   /**
-   * Bypasses the container state machine to change a container's state from DELETING to CLOSED. This API was
+   * Bypasses the container state machine to change a container's state from DELETING or DELETED to CLOSED. This API was
    * introduced to fix a bug (HDDS-11136), and should be used with care otherwise.
    *
    * @see <a href="https://issues.apache.org/jira/browse/HDDS-11136">HDDS-11136</a>
    * @param id id of the container to transition
    * @throws IOException
    */
   @Replicate
-  void transitionDeletingToClosedState(HddsProtos.ContainerID id) throws IOException;
+  void transitionDeletingOrDeletedToClosedState(HddsProtos.ContainerID id) throws IOException;
 
   /**
    *

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

@@ -370,16 +370,16 @@ public void updateContainerState(final HddsProtos.ContainerID containerID,
   }
 
   @Override
-  public void transitionDeletingToClosedState(HddsProtos.ContainerID containerID) throws IOException {
+  public void transitionDeletingOrDeletedToClosedState(HddsProtos.ContainerID containerID) throws IOException {
     final ContainerID id = ContainerID.getFromProtobuf(containerID);
 
     try (AutoCloseableLock ignored = writeLock(id)) {
       if (containers.contains(id)) {
         final ContainerInfo oldInfo = containers.getContainerInfo(id);
         final LifeCycleState oldState = oldInfo.getState();
-        if (oldState != DELETING) {
+        if (oldState != DELETING && oldState != DELETED) {
           throw new InvalidContainerStateException("Cannot transition container " + id + " from " + oldState +
-              " back to CLOSED. The container must be in the DELETING state.");
+              " back to CLOSED. The container must be in the DELETING or DELETED state.");
         }
         ExecutionUtil.create(() -> {
           containers.updateState(id, oldState, CLOSED);

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

@@ -23,6 +23,7 @@
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerReplica;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerCheckRequest;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerReplicaOp;
 import org.apache.hadoop.hdds.scm.container.replication.ReplicationManager;
@@ -88,6 +89,7 @@ public boolean handle(ContainerCheckRequest request) {
     //resend deleteCommand if needed
     request.getContainerReplicas().stream()
         .filter(r -> !pendingDelete.contains(r.getDatanodeDetails()))
+        .filter(ContainerReplica::isEmpty)
         .forEach(rp -> {
           try {
             replicationManager.sendDeleteCommand(

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java

@@ -19,6 +19,7 @@
 
 import static org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes.FAILED_TO_CHANGE_CONTAINER_STATE;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableSet;
 import java.util.Collections;
@@ -93,6 +94,11 @@ public ContainerStateMap() {
     this.replicaMap = new ConcurrentHashMap<>();
   }
 
+  @VisibleForTesting
+  public static Logger getLogger() {
+    return LOG;
+  }
+
   /**
    * Adds a ContainerInfo Entry in the ContainerStateMap.
    *

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/TestContainerManagerImpl.java

@@ -42,6 +42,7 @@
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerReplicaPendingOps;
+import org.apache.hadoop.hdds.scm.container.states.ContainerStateMap;
 import org.apache.hadoop.hdds.scm.ha.SCMHAManager;
 import org.apache.hadoop.hdds.scm.ha.SCMHAManagerStub;
 import org.apache.hadoop.hdds.scm.ha.SequenceIdGenerator;
@@ -53,10 +54,15 @@
 import org.apache.hadoop.hdds.utils.db.DBStoreBuilder;
 import org.apache.hadoop.ozone.common.statemachine.InvalidStateTransitionException;
 import org.apache.hadoop.ozone.container.common.SCMTestUtils;
+import org.apache.ozone.test.GenericTestUtils;
 import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.slf4j.event.Level;
 
 /**
  * Tests to verify the functionality of ContainerManager.
@@ -72,6 +78,12 @@ public class TestContainerManagerImpl {
   private NodeManager nodeManager;
   private ContainerReplicaPendingOps pendingOpsMock;
 
+  @BeforeAll
+  static void init() {
+    // Print container state transition logs
+    GenericTestUtils.setLogLevel(ContainerStateMap.getLogger(), Level.TRACE);
+  }
+
   @BeforeEach
   void setUp() throws Exception {
     final OzoneConfiguration conf = SCMTestUtils.getConf(testDir);
@@ -131,9 +143,12 @@ void testUpdateContainerState() throws Exception {
     assertEquals(LifeCycleState.CLOSED, containerManager.getContainer(cid).getState());
   }
 
-  @Test
-  void testTransitionDeletingToClosedState() throws IOException, InvalidStateTransitionException {
-    // allocate OPEN Ratis and Ec containers, and do a series of state changes to transition them to DELETING
+  @ParameterizedTest
+  @EnumSource(value = HddsProtos.LifeCycleState.class,
+      names = {"DELETING", "DELETED"})
+  void testTransitionDeletingOrDeletedToClosedState(HddsProtos.LifeCycleState desiredState)
+      throws IOException, InvalidStateTransitionException {
+    // Allocate OPEN Ratis and Ec containers, and do a series of state changes to transition them to DELETING / DELETED
     final ContainerInfo container = containerManager.allocateContainer(
         RatisReplicationConfig.getInstance(
             ReplicationFactor.THREE), "admin");
@@ -162,29 +177,39 @@ void testTransitionDeletingToClosedState() throws IOException, InvalidStateTrans
     assertEquals(LifeCycleState.DELETING, containerManager.getContainer(cid).getState());
     assertEquals(LifeCycleState.DELETING, containerManager.getContainer(ecCid).getState());
 
-    // DELETING -> CLOSED
-    containerManager.transitionDeletingToClosedState(cid);
-    containerManager.transitionDeletingToClosedState(ecCid);
+    if (desiredState == LifeCycleState.DELETED) {
+      // DELETING -> DELETED
+      containerManager.updateContainerState(cid, HddsProtos.LifeCycleEvent.CLEANUP);
+      containerManager.updateContainerState(ecCid, HddsProtos.LifeCycleEvent.CLEANUP);
+      assertEquals(LifeCycleState.DELETED, containerManager.getContainer(cid).getState());
+      assertEquals(LifeCycleState.DELETED, containerManager.getContainer(ecCid).getState());
+    }
+
+    // DELETING / DELETED -> CLOSED
+    containerManager.transitionDeletingOrDeletedToClosedState(cid);
+    containerManager.transitionDeletingOrDeletedToClosedState(ecCid);
     // the containers should be back in CLOSED state now
     assertEquals(LifeCycleState.CLOSED, containerManager.getContainer(cid).getState());
     assertEquals(LifeCycleState.CLOSED, containerManager.getContainer(ecCid).getState());
   }
 
   @Test
-  void testTransitionDeletingToClosedStateAllowsOnlyDeletingContainers() throws IOException {
+  void testTransitionContainerToClosedStateAllowOnlyDeletingOrDeletedContainers() throws IOException {
+    // Negative test for Ratis/EC container OPEN -> CLOSED transition
+
     // test for RATIS container
     final ContainerInfo container = containerManager.allocateContainer(
         RatisReplicationConfig.getInstance(
             ReplicationFactor.THREE), "admin");
     final ContainerID cid = container.containerID();
     assertEquals(LifeCycleState.OPEN, containerManager.getContainer(cid).getState());
-    assertThrows(IOException.class, () -> containerManager.transitionDeletingToClosedState(cid));
+    assertThrows(IOException.class, () -> containerManager.transitionDeletingOrDeletedToClosedState(cid));
 
     // test for EC container
     final ContainerInfo ecContainer = containerManager.allocateContainer(new ECReplicationConfig(3, 2), "admin");
     final ContainerID ecCid = ecContainer.containerID();
     assertEquals(LifeCycleState.OPEN, containerManager.getContainer(ecCid).getState());
-    assertThrows(IOException.class, () -> containerManager.transitionDeletingToClosedState(ecCid));
+    assertThrows(IOException.class, () -> containerManager.transitionDeletingOrDeletedToClosedState(ecCid));
   }
 
   @Test