HDDS-14684. Allow deletion of empty quasi-closed containers

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

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hdds.scm.container.replication.health;
 
 import java.util.Set;
+import java.util.stream.Collectors;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
 import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
@@ -31,7 +32,7 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * This handler deletes a container if it's closed and empty (0 key count)
+ * This handler deletes a container if it's closed or quasi-closed and empty (0 key count)
  * and all its replicas are empty.
  */
 public class EmptyContainerHandler extends AbstractCheck {
@@ -45,8 +46,8 @@ public EmptyContainerHandler(ReplicationManager replicationManager) {
   }
 
   /**
-   * Deletes a container if it's closed and empty (0 key count) and all its
-   * replicas are closed and empty.
+   * Deletes a container if it's closed or quasi-closed and empty (0 key count) and all its
+   * replicas are empty.
    * @param request ContainerCheckRequest object representing the container
    * @return true if the specified container is empty, otherwise false
    */
@@ -75,6 +76,33 @@ public boolean handle(ContainerCheckRequest request) {
             containerInfo.containerID(), HddsProtos.LifeCycleEvent.DELETE);
       }
       return true;
+    } else if (isContainerEmptyAndQuasiClosed(containerInfo, replicas)) {
+      request.getReport().incrementAndSample(ContainerHealthState.EMPTY, containerInfo);
+      if (!request.isReadOnly()) {
+        String originIds = replicas.stream()
+            .map(r -> r.getOriginDatanodeId().toString())
+            .collect(Collectors.joining(", "));
+        LOG.info("Deleting empty QUASI_CLOSED container {} with {} replicas from originIds: [{}]. " +
+                "If resurrected, container will transition to CLOSED but may have QUASI_CLOSED replicas.",
+                containerInfo.containerID(), replicas.size(), originIds);
+        // delete replicas if they are quasi-closed and empty
+        deleteContainerReplicas(containerInfo, replicas);
+
+        if (containerInfo.getReplicationType() == HddsProtos.ReplicationType.RATIS) {
+          if (replicas.stream().filter(r -> r.getSequenceId() != null)
+                  .noneMatch(r -> r.getSequenceId() == containerInfo.getSequenceId())) {
+            // don't update container state if replica seqid don't match with container seq id
+            return true;
+          }
+        }
+        // Update the container's state - transition to CLOSED first, then DELETE
+        // QUASI_CLOSED -> CLOSED requires FORCE_CLOSE event
+        replicationManager.updateContainerState(
+                containerInfo.containerID(), HddsProtos.LifeCycleEvent.FORCE_CLOSE);
+        replicationManager.updateContainerState(
+                containerInfo.containerID(), HddsProtos.LifeCycleEvent.DELETE);
+      }
+      return true;
     } else if (containerInfo.getState() == HddsProtos.LifeCycleState.CLOSED
         && containerInfo.getNumberOfKeys() == 0 && replicas.isEmpty()) {
       // If the container is empty and has no replicas, it is possible it was
@@ -114,19 +142,44 @@ private boolean isContainerEmptyAndClosed(final ContainerInfo container,
   }
 
   /**
-   * Deletes the specified container's replicas if they are closed and empty.
+   * Returns true if the container is empty and QUASI_CLOSED.
+   * For QUASI_CLOSED containers, replicas can be in QUASI_CLOSED, OPEN,
+   * CLOSING, or UNHEALTHY states. We check if all replicas are empty regardless
+   * of their state.
+   *
+   * @param container Container to check
+   * @param replicas Set of ContainerReplica
+   * @return true if the container is considered empty and quasi-closed, false otherwise
+   */
+  private boolean isContainerEmptyAndQuasiClosed(final ContainerInfo container,
+      final Set<ContainerReplica> replicas) {
+    return container.getState() == HddsProtos.LifeCycleState.QUASI_CLOSED &&
+            !replicas.isEmpty() &&
+            replicas.stream().allMatch(ContainerReplica::isEmpty);
+  }
+
+  /**
+   * Deletes the specified container's replicas if they are empty.
+   * For CLOSED containers, replicas must also be CLOSED.
+   * For QUASI_CLOSED containers, replicas can be in any state (QUASI_CLOSED, OPEN, CLOSING, UNHEALTHY).
    *
    * @param containerInfo ContainerInfo to delete
    * @param replicas Set of ContainerReplica
    */
   private void deleteContainerReplicas(final ContainerInfo containerInfo,
       final Set<ContainerReplica> replicas) {
-    Preconditions.assertSame(HddsProtos.LifeCycleState.CLOSED,
-        containerInfo.getState(), "container state");
+    boolean isQuasiClosed = containerInfo.getState() == HddsProtos.LifeCycleState.QUASI_CLOSED;
+
+    if (!isQuasiClosed) {
+      Preconditions.assertSame(HddsProtos.LifeCycleState.CLOSED,
+          containerInfo.getState(), "container state");
+    }
 
     for (ContainerReplica rp : replicas) {
-      Preconditions.assertSame(ContainerReplicaProto.State.CLOSED,
-          rp.getState(), "replica state");
+      if (!isQuasiClosed) {
+        Preconditions.assertSame(ContainerReplicaProto.State.CLOSED,
+            rp.getState(), "replica state");
+      }
       Preconditions.assertSame(true, rp.isEmpty(), "replica empty");
 
       try {

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/health/TestEmptyContainerHandler.java

@@ -19,6 +19,7 @@
 
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.CLOSED;
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.CLOSING;
+import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.QUASI_CLOSED;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.any;
@@ -284,6 +285,49 @@ public void testNoUpdateContainerStateWhenReplicaSequenceIdDoesNotMatch()
         any(HddsProtos.LifeCycleEvent.class));
   }
 
+  /**
+   * A QUASI_CLOSED container with all empty replicas should be deleted.
+   * Handler should return true and send delete commands to all replicas.
+   */
+  @Test
+  public void testEmptyQuasiClosedRatisContainerReturnsTrue()
+          throws IOException {
+    long keyCount = 0L;
+    long bytesUsed = 0L;
+    ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(
+            ratisReplicationConfig, 1, QUASI_CLOSED, keyCount, bytesUsed);
+    Set<ContainerReplica> containerReplicas = ReplicationTestUtil
+        .createReplicas(containerInfo.containerID(),
+            ContainerReplicaProto.State.QUASI_CLOSED, keyCount, bytesUsed,
+            0, 0, 0);
+
+    ContainerCheckRequest request = new ContainerCheckRequest.Builder()
+        .setPendingOps(Collections.emptyList())
+        .setReport(new ReplicationManagerReport(rmConf.getContainerSampleLimit()))
+        .setContainerInfo(containerInfo)
+        .setContainerReplicas(containerReplicas)
+        .build();
+
+    ContainerCheckRequest readRequest = new ContainerCheckRequest.Builder()
+        .setPendingOps(Collections.emptyList())
+        .setReport(new ReplicationManagerReport(rmConf.getContainerSampleLimit()))
+        .setContainerInfo(containerInfo)
+        .setContainerReplicas(containerReplicas)
+        .setReadOnly(true)
+        .build();
+
+    assertAndVerify(readRequest, true, 0, 1);
+
+    // Should delete all 3 replicas and update state twice (FORCE_CLOSE + DELETE)
+    assertTrue(emptyContainerHandler.handle(request));
+    verify(replicationManager, times(3)).sendDeleteCommand(any(ContainerInfo.class), anyInt(),
+        any(DatanodeDetails.class), eq(false));
+    assertEquals(1, request.getReport().getStat(ContainerHealthState.EMPTY));
+    // QUASI_CLOSED requires 2 state updates: FORCE_CLOSE + DELETE
+    verify(replicationManager, times(2)).updateContainerState(
+        any(ContainerID.class), any(HddsProtos.LifeCycleEvent.class));
+  }
+
   /**
    * Asserts that handler returns the specified assertion and delete command
    * to replicas is sent the specified number of times.

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManagerIntegration.java

@@ -29,6 +29,7 @@
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState.IN_MAINTENANCE;
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState.IN_SERVICE;
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.DEAD;
+import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State.QUASI_CLOSED;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DEADNODE_INTERVAL;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL;
@@ -362,4 +363,69 @@ public void testOneDeadMaintenanceNodeAndOneLiveMaintenanceNodeAndOneDecommissio
     assertEquals(0, report.getStat(ContainerHealthState.MIS_REPLICATED));
     assertEquals(0, report.getStat(ContainerHealthState.OVER_REPLICATED));
   }
+
+  /**
+   * Test for empty QUASI_CLOSED container deletion.
+   */
+  @Test
+  public void testEmptyQuasiClosedContainerDeletion() throws Exception {
+    ContainerInfo containerInfo = containerManager.allocateContainer(RATIS_REPLICATION_CONFIG, "TestOwner");
+    ContainerID cid = containerInfo.containerID();
+    containerManager.updateContainerState(cid, HddsProtos.LifeCycleEvent.FINALIZE);
+    containerManager.updateContainerState(cid, HddsProtos.LifeCycleEvent.QUASI_CLOSE);
+
+    // Wait for container to be QUASI_CLOSED
+    GenericTestUtils.waitFor(() -> {
+      try {
+        ContainerInfo info = containerManager.getContainer(cid);
+        return info.getState() == HddsProtos.LifeCycleState.QUASI_CLOSED;
+      } catch (ContainerNotFoundException e) {
+        return false;
+      }
+    }, 100, 5000);
+
+    containerInfo = containerManager.getContainer(cid);
+    assertEquals(HddsProtos.LifeCycleState.QUASI_CLOSED, containerInfo.getState());
+    assertEquals(0L, containerInfo.getNumberOfKeys());
+
+    // Add empty QUASI_CLOSED replicas
+    List<DatanodeDetails> datanodes = nodeManager.getAllNodes().stream()
+        .limit(3).collect(Collectors.toList());
+
+    for (int i = 0; i < 3; i++) {
+      ContainerReplica replica = ContainerReplica.newBuilder()
+          .setContainerID(cid)
+          .setContainerState(QUASI_CLOSED)
+          .setDatanodeDetails(datanodes.get(i))
+          .setOriginNodeId(datanodes.get(i).getID())
+          .setSequenceId(0L)
+          .setKeyCount(0L)
+          .setBytesUsed(0L)
+          .setEmpty(true)
+          .setReplicaIndex(i)
+          .build();
+      containerManager.updateContainerReplica(cid, replica);
+    }
+
+    Set<ContainerReplica> replicas = containerManager.getContainerReplicas(cid);
+    assertEquals(3, replicas.size());
+    assertTrue(replicas.stream().allMatch(ContainerReplica::isEmpty));
+
+    replicationManager.getConfig().setInterval(Duration.ofSeconds(1));
+    replicationManager.notifyStatusChanged();
+
+    // QUASI_CLOSED -> CLOSED -> DELETING
+    GenericTestUtils.waitFor(() -> {
+      try {
+        ContainerInfo info = containerManager.getContainer(cid);
+        HddsProtos.LifeCycleState state = info.getState();
+        return state == HddsProtos.LifeCycleState.DELETING;
+      } catch (ContainerNotFoundException e) {
+        return false;
+      }
+    }, 1000, 30000);
+
+    containerInfo = containerManager.getContainer(cid);
+    assertEquals(HddsProtos.LifeCycleState.DELETING, containerInfo.getState());
+  }
 }