HDDS-7384. EC: ReplicationManager - implement deleting container handler

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

@@ -38,6 +38,7 @@
 import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.apache.hadoop.hdds.scm.container.replication.health.ClosedWithMismatchedReplicasHandler;
 import org.apache.hadoop.hdds.scm.container.replication.health.ClosingContainerHandler;
+import org.apache.hadoop.hdds.scm.container.replication.health.DeletingContainerHandler;
 import org.apache.hadoop.hdds.scm.container.replication.health.ECReplicationCheckHandler;
 import org.apache.hadoop.hdds.scm.container.replication.health.HealthCheck;
 import org.apache.hadoop.hdds.scm.container.replication.health.OpenContainerHandler;
@@ -220,6 +221,7 @@ public ReplicationManager(final ConfigurationSource conf,
         .addNext(new ClosingContainerHandler(this))
         .addNext(new QuasiClosedContainerHandler(this))
         .addNext(new ClosedWithMismatchedReplicasHandler(this))
+        .addNext(new DeletingContainerHandler(this, containerManager))
         .addNext(ecReplicationCheckHandler)
         .addNext(ratisReplicationCheckHandler);
     start();

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

@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm.container.replication.health;
+
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+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.ContainerManager;
+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;
+import org.apache.hadoop.ozone.common.statemachine.InvalidStateTransitionException;
+import org.apache.ratis.protocol.exceptions.NotLeaderException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+
+/**
+ * Class used in Replication Manager to handle the
+ * replicas of containers in DELETING State.
+ */
+public class DeletingContainerHandler extends AbstractCheck {
+  private final ReplicationManager replicationManager;
+  private final ContainerManager containerManager;
+
+  public static final Logger LOG =
+      LoggerFactory.getLogger(DeletingContainerHandler.class);
+
+  public DeletingContainerHandler(ReplicationManager replicationManager,
+                                  ContainerManager containerManager) {
+    this.replicationManager = replicationManager;
+    this.containerManager = containerManager;
+  }
+
+  /**
+   * If the replica size of the container is 0, change the state
+   * of the container to Deleted, otherwise resend delete command if needed.
+   * @param request ContainerCheckRequest object representing the container
+   * @return false if the specified container is not in DELETING state,
+   * otherwise true.
+   */
+  @Override
+  public boolean handle(ContainerCheckRequest request) {
+    ContainerInfo containerInfo = request.getContainerInfo();
+    ContainerID cID = containerInfo.containerID();
+
+    if (containerInfo.getState() != HddsProtos.LifeCycleState.DELETING) {
+      return false;
+    }
+
+    if (request.getContainerReplicas().size() == 0) {
+      try {
+        containerManager.updateContainerState(
+            cID, HddsProtos.LifeCycleEvent.CLEANUP);
+        LOG.debug("Container {} state changes to DELETED", cID);
+      } catch (IOException | InvalidStateTransitionException |
+               TimeoutException e) {
+        LOG.error("Failed to cleanup empty container {}",
+            request.getContainerInfo(), e);
+      }
+      return true;
+    }
+
+    Set<DatanodeDetails> pendingDelete = request.getPendingOps().stream()
+        .filter(o -> o.getOpType() == ContainerReplicaOp.PendingOpType.DELETE)
+        .map(o -> o.getTarget()).collect(Collectors.toSet());
+    //resend deleteCommand if needed
+    request.getContainerReplicas().stream()
+        .filter(r -> !pendingDelete.contains(r.getDatanodeDetails()))
+        .forEach(rp -> {
+          try {
+            replicationManager.sendDeleteCommand(
+                containerInfo, rp.getReplicaIndex(), rp.getDatanodeDetails());
+          } catch (NotLeaderException e) {
+            LOG.warn("Failed to delete empty replica with index {} for " +
+                    "container {} on datanode {}", rp.getReplicaIndex(),
+                cID, rp.getDatanodeDetails().getUuidString(), e);
+          }
+        });
+    return true;
+  }
+}