HDDS-14551. Don't update container state when replica sequence doesn't match with container seq id.

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

@@ -63,6 +63,13 @@ public boolean handle(ContainerCheckRequest request) {
         // delete replicas if they are 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
         replicationManager.updateContainerState(
             containerInfo.containerID(), HddsProtos.LifeCycleEvent.DELETE);

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

@@ -20,6 +20,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.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyInt;
 import static org.mockito.Mockito.eq;
@@ -247,6 +248,42 @@ public void testEmptyECContainerWithNonEmptyReplicaReturnsFalse()
     assertAndVerify(request, false, 0, 0);
   }
 
+  /**
+   * Tests that container state is NOT updated to DELETE when no replica
+   * has a matching sequence ID with the container.
+   */
+  @Test
+  public void testNoUpdateContainerStateWhenReplicaSequenceIdDoesNotMatch()
+      throws IOException {
+    long keyCount = 0L;
+    long bytesUsed = 0L;
+    long containerSequenceId = 100L;
+    // Create container with specific sequence ID
+    ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(
+        ratisReplicationConfig, 1, CLOSED, containerSequenceId);
+    // Create replicas - all with 0 sequence IDs (none matching container)
+    Set<ContainerReplica> containerReplicas = ReplicationTestUtil
+        .createReplicas(containerInfo.containerID(),
+            ContainerReplicaProto.State.CLOSED, keyCount, bytesUsed,
+            0, 0, 0);
+    ContainerCheckRequest request = new ContainerCheckRequest.Builder()
+        .setPendingOps(Collections.emptyList())
+        .setReport(new ReplicationManagerReport(rmConf.getContainerSampleLimit()))
+        .setContainerInfo(containerInfo)
+        .setContainerReplicas(containerReplicas)
+        .build();
+    // Handler should return true but NOT update container state
+    // because no replica has matching sequence ID
+    assertTrue(emptyContainerHandler.handle(request));
+    verify(replicationManager, times(3)).sendDeleteCommand(
+        any(ContainerInfo.class), anyInt(),
+        any(DatanodeDetails.class), eq(false));
+    // updateContainerState should NOT be called when sequence IDs don't match
+    verify(replicationManager, times(0)).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.