HDDS-11136. Some containers affected by HDDS-8129 may still be in the DELETING state incorrectly

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

@@ -349,8 +349,13 @@ private boolean updateContainerState(final DatanodeDetails datanode,
       break;
     case DELETING:
       /*
-       * The container is under deleting. do nothing.
+      HDDS-11136: If a DELETING container has a non-empty CLOSED replica, the container should be moved back to CLOSED
+      state.
        */
+      if (replica.getState() == State.CLOSED && replica.hasIsEmpty() && !replica.getIsEmpty()) {
+        containerManager.transitionDeletingToClosedState(containerId);
+      }
+
       break;
     case DELETED:
       /*

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

@@ -131,6 +131,16 @@ void updateContainerState(ContainerID containerID,
                             LifeCycleEvent event)
       throws IOException, InvalidStateTransitionException;
 
+  /**
+   * Bypasses the container state machine to change a container's state from DELETING to CLOSED. This API was
+   * specially introduced to fix a bug (HDDS-11136), and should NOT be used in any other context.
+   *
+   * @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;
+
   /**
    * Returns the latest list of replicas for given containerId.
    *

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

@@ -283,6 +283,21 @@ public void updateContainerState(final ContainerID cid,
     }
   }
 
+  @Override
+  public void transitionDeletingToClosedState(ContainerID containerID) throws IOException {
+    HddsProtos.ContainerID proto = containerID.getProtobuf();
+    lock.lock();
+    try {
+      if (containerExist(containerID)) {
+        containerStateManager.transitionDeletingToClosedState(proto);
+      } else {
+        throwContainerNotFoundException(containerID);
+      }
+    } finally {
+      lock.unlock();
+    }
+  }
+
   @Override
   public Set<ContainerReplica> getContainerReplicas(final ContainerID id)
       throws ContainerNotFoundException {

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

@@ -151,6 +151,18 @@ void updateContainerState(HddsProtos.ContainerID id,
                             HddsProtos.LifeCycleEvent event)
       throws IOException, InvalidStateTransitionException;
 
+
+  /**
+   * Bypasses the container state machine to change a container's state from DELETING to CLOSED. This API was
+   * specially introduced to fix a bug (HDDS-11136), and should NOT be used in any other context.
+   *
+   * @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;
+
   /**
    *
    */

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

@@ -39,6 +39,7 @@
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
 import org.apache.hadoop.hdds.protocol.proto.SCMRatisProtocol.RequestType;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.scm.container.common.helpers.InvalidContainerStateException;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerReplicaPendingOps;
 import org.apache.hadoop.hdds.scm.container.states.ContainerState;
 import org.apache.hadoop.hdds.scm.container.states.ContainerStateMap;
@@ -367,6 +368,28 @@ public void updateContainerState(final HddsProtos.ContainerID containerID,
     }
   }
 
+  @Override
+  public void transitionDeletingToClosedState(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) {
+          throw new InvalidContainerStateException("Container " + oldInfo + "must be in DELETING state to " +
+            "transition to CLOSED.");
+        }
+        ExecutionUtil.create(() -> {
+          containers.updateState(id, oldState, CLOSED);
+          transactionBuffer.addToBuffer(containerStore, id, containers.getContainerInfo(id));
+        }).onException(() -> {
+          transactionBuffer.addToBuffer(containerStore, id, oldInfo);
+          containers.updateState(id, CLOSED, oldState);
+        }).execute();
+      }
+    }
+  }
 
   @Override
   public Set<ContainerReplica> getContainerReplicas(final ContainerID id) {

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

@@ -41,12 +41,14 @@
 import org.apache.hadoop.hdds.scm.pipeline.PipelineManager;
 import org.apache.hadoop.hdds.utils.db.DBStore;
 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.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -132,6 +134,62 @@ 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
+    final ContainerInfo container = containerManager.allocateContainer(
+        RatisReplicationConfig.getInstance(
+            ReplicationFactor.THREE), "admin");
+    ContainerInfo ecContainer = containerManager.allocateContainer(new ECReplicationConfig(3, 2), "admin");
+    final ContainerID cid = container.containerID();
+    final ContainerID ecCid = ecContainer.containerID();
+    assertEquals(LifeCycleState.OPEN, containerManager.getContainer(cid).getState());
+    assertEquals(LifeCycleState.OPEN, containerManager.getContainer(ecCid).getState());
+
+    // OPEN -> CLOSING
+    containerManager.updateContainerState(cid,
+        HddsProtos.LifeCycleEvent.FINALIZE);
+    containerManager.updateContainerState(ecCid, HddsProtos.LifeCycleEvent.FINALIZE);
+    assertEquals(LifeCycleState.CLOSING, containerManager.getContainer(cid).getState());
+    assertEquals(LifeCycleState.CLOSING, containerManager.getContainer(ecCid).getState());
+
+    // CLOSING -> CLOSED
+    containerManager.updateContainerState(cid, HddsProtos.LifeCycleEvent.CLOSE);
+    containerManager.updateContainerState(ecCid, HddsProtos.LifeCycleEvent.CLOSE);
+    assertEquals(LifeCycleState.CLOSED, containerManager.getContainer(cid).getState());
+    assertEquals(LifeCycleState.CLOSED, containerManager.getContainer(ecCid).getState());
+
+    // CLOSED -> DELETING
+    containerManager.updateContainerState(cid, HddsProtos.LifeCycleEvent.DELETE);
+    containerManager.updateContainerState(ecCid, HddsProtos.LifeCycleEvent.DELETE);
+    assertEquals(LifeCycleState.DELETING, containerManager.getContainer(cid).getState());
+    assertEquals(LifeCycleState.DELETING, containerManager.getContainer(ecCid).getState());
+
+    // DELETING -> CLOSED
+    containerManager.transitionDeletingToClosedState(cid);
+    containerManager.transitionDeletingToClosedState(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 {
+    // 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));
+
+    // 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));
+  }
+
   @Test
   void testGetContainers() throws Exception {
     assertEquals(emptyList(), containerManager.getContainers());

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

@@ -66,6 +66,7 @@
 import java.util.stream.Stream;
 
 import static org.apache.hadoop.hdds.protocol.MockDatanodeDetails.randomDatanodeDetails;
+import static org.apache.hadoop.hdds.scm.HddsTestUtils.getContainerReports;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.doAnswer;
@@ -100,7 +101,6 @@ void setup() throws IOException, InvalidStateTransitionException {
     dbStore = DBStoreBuilder.createDBStore(
         conf, new SCMDBDefinition());
     scmhaManager = SCMHAManagerStub.getInstance(true);
-    nodeManager = new MockNodeManager(true, 10);
     pipelineManager =
         new MockPipelineManager(dbStore, scmhaManager, nodeManager);
     containerStateManager = ContainerStateManagerImpl.newBuilder()
@@ -151,6 +151,10 @@ void setup() throws IOException, InvalidStateTransitionException {
     }).when(containerManager).removeContainerReplica(
         any(ContainerID.class), any(ContainerReplica.class));
 
+    doAnswer(invocation -> {
+      containerStateManager.transitionDeletingToClosedState(((ContainerID) invocation.getArgument(0)).getProtobuf());
+      return null;
+    }).when(containerManager).transitionDeletingToClosedState(any(ContainerID.class));
   }
 
   @AfterEach
@@ -442,6 +446,85 @@ public void testClosingToClosedForECContainer()
     assertEquals(LifeCycleState.CLOSED, containerManager.getContainer(container2.containerID()).getState());
   }
 
+  /**
+   * Tests that a DELETING container transitions to CLOSED if a non-empty CLOSED replica is reported. It does not
+   * transition if a non-empty CLOSING (or any other state) replica is reported.
+   */
+  @Test
+  public void shouldTransitionFromDeletingToClosedWhenNonEmptyClosedReplica() throws IOException {
+    ContainerInfo container = getContainer(LifeCycleState.DELETING);
+    containerStateManager.addContainer(container.getProtobuf());
+
+    // set up a non-empty CLOSED replica
+    DatanodeDetails dnWithClosedReplica = nodeManager.getNodes(NodeStatus.inServiceHealthy()).get(0);
+    ContainerReplicaProto.Builder builder = ContainerReplicaProto.newBuilder();
+    ContainerReplicaProto closedReplica = builder.setContainerID(container.getContainerID())
+        .setIsEmpty(false)
+        .setState(ContainerReplicaProto.State.CLOSED)
+        .setKeyCount(0)
+        .setBlockCommitSequenceId(123)
+        .setOriginNodeId(dnWithClosedReplica.getUuidString()).build();
+
+    // set up a non-empty CLOSING replica
+    DatanodeDetails dnWithClosingReplica = nodeManager.getNodes(NodeStatus.inServiceHealthy()).get(1);
+    ContainerReplicaProto closingReplica = builder.setState(ContainerReplicaProto.State.CLOSING)
+        .setOriginNodeId(dnWithClosingReplica.getUuidString()).build();
+
+    // should not transition on processing the CLOSING replica's report
+    ContainerReportHandler containerReportHandler = new ContainerReportHandler(nodeManager, containerManager);
+    ContainerReportsProto closingContainerReport = getContainerReports(closingReplica);
+    containerReportHandler
+        .onMessage(new ContainerReportFromDatanode(dnWithClosingReplica, closingContainerReport), publisher);
+
+    assertEquals(LifeCycleState.DELETING, containerStateManager.getContainer(container.containerID()).getState());
+
+    // should transition on processing the CLOSED replica's report
+    ContainerReportsProto closedContainerReport = getContainerReports(closedReplica);
+    containerReportHandler
+        .onMessage(new ContainerReportFromDatanode(dnWithClosedReplica, closedContainerReport), publisher);
+    assertEquals(LifeCycleState.CLOSED, containerStateManager.getContainer(container.containerID()).getState());
+  }
+
+  /**
+   * Tests that a DELETING EC container transitions to CLOSED if a non-empty CLOSED replica is reported. It does not
+   * transition if a non-empty CLOSING (or any other state) replica is reported.
+   */
+  @Test
+  public void ecContainerShouldTransitionFromDeletingToClosedWhenNonEmptyClosedReplica() throws IOException {
+    ContainerInfo container = getECContainer(LifeCycleState.DELETING, PipelineID.randomId(),
+        new ECReplicationConfig(6, 3));
+    containerStateManager.addContainer(container.getProtobuf());
+
+    // set up a non-empty CLOSED replica
+    DatanodeDetails dnWithClosedReplica = nodeManager.getNodes(NodeStatus.inServiceHealthy()).get(0);
+    ContainerReplicaProto.Builder builder = ContainerReplicaProto.newBuilder();
+    ContainerReplicaProto closedReplica = builder.setContainerID(container.getContainerID())
+        .setIsEmpty(false)
+        .setState(ContainerReplicaProto.State.CLOSED)
+        .setKeyCount(0)
+        .setBlockCommitSequenceId(0)
+        .setReplicaIndex(1)
+        .setOriginNodeId(dnWithClosedReplica.getUuidString()).build();
+
+    // set up a non-empty CLOSING replica
+    DatanodeDetails dnWithClosingReplica = nodeManager.getNodes(NodeStatus.inServiceHealthy()).get(1);
+    ContainerReplicaProto closingReplica = builder.setState(ContainerReplicaProto.State.CLOSING).setReplicaIndex(2)
+        .setOriginNodeId(dnWithClosingReplica.getUuidString()).build();
+
+    // should not transition on processing the CLOSING replica's report
+    ContainerReportHandler containerReportHandler = new ContainerReportHandler(nodeManager, containerManager);
+    ContainerReportsProto closingContainerReport = getContainerReports(closingReplica);
+    containerReportHandler
+        .onMessage(new ContainerReportFromDatanode(dnWithClosingReplica, closingContainerReport), publisher);
+    assertEquals(LifeCycleState.DELETING, containerStateManager.getContainer(container.containerID()).getState());
+
+    // should transition on processing the CLOSED replica's report
+    ContainerReportsProto closedContainerReport = getContainerReports(closedReplica);
+    containerReportHandler
+        .onMessage(new ContainerReportFromDatanode(dnWithClosedReplica, closedContainerReport), publisher);
+    assertEquals(LifeCycleState.CLOSED, containerStateManager.getContainer(container.containerID()).getState());
+  }
+
   /**
    * Creates the required number of DNs that will hold a replica each for the
    * specified EC container. Registers these DNs with the NodeManager, adds this

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

@@ -50,6 +50,7 @@
 import org.junit.jupiter.api.io.TempDir;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.when;
@@ -148,6 +149,42 @@ public void checkReplicationStateMissingReplica()
     assertEquals(3, c1.getReplicationConfig().getRequiredNodes());
   }
 
+  @Test
+  public void testTransitionDeletingToClosedState() throws IOException {
+    HddsProtos.ContainerInfoProto.Builder builder = HddsProtos.ContainerInfoProto.newBuilder();
+    builder.setContainerID(1)
+        .setState(HddsProtos.LifeCycleState.DELETING)
+        .setUsedBytes(0)
+        .setNumberOfKeys(0)
+        .setOwner("root")
+        .setReplicationType(HddsProtos.ReplicationType.RATIS)
+        .setReplicationFactor(ReplicationFactor.THREE);
+
+    HddsProtos.ContainerInfoProto container = builder.build();
+    HddsProtos.ContainerID cid = HddsProtos.ContainerID.newBuilder().setId(container.getContainerID()).build();
+    containerStateManager.addContainer(container);
+    containerStateManager.transitionDeletingToClosedState(cid);
+    assertEquals(HddsProtos.LifeCycleState.CLOSED, containerStateManager.getContainer(ContainerID.getFromProtobuf(cid))
+        .getState());
+  }
+
+  @Test
+  public void testTransitionDeletingToClosedStateAllowsOnlyDeletingContainer() throws IOException {
+    HddsProtos.ContainerInfoProto.Builder builder = HddsProtos.ContainerInfoProto.newBuilder();
+    builder.setContainerID(1)
+        .setState(HddsProtos.LifeCycleState.QUASI_CLOSED)
+        .setUsedBytes(0)
+        .setNumberOfKeys(0)
+        .setOwner("root")
+        .setReplicationType(HddsProtos.ReplicationType.RATIS)
+        .setReplicationFactor(ReplicationFactor.THREE);
+
+    HddsProtos.ContainerInfoProto container = builder.build();
+    HddsProtos.ContainerID cid = HddsProtos.ContainerID.newBuilder().setId(container.getContainerID()).build();
+    containerStateManager.addContainer(container);
+    assertThrows(IOException.class, () -> containerStateManager.transitionDeletingToClosedState(cid));
+  }
+
   private void addReplica(ContainerInfo cont, DatanodeDetails node) {
     ContainerReplica replica = ContainerReplica.newBuilder()
         .setContainerID(cont.containerID())