diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java
index 79cf324..5b8d6da 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java
@@ -27,6 +27,7 @@
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State;
 import org.apache.hadoop.hdds.scm.PlacementPolicy;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
@@ -42,7 +43,10 @@
 import org.apache.hadoop.hdds.scm.node.NodeManager;
 import org.apache.hadoop.hdds.scm.node.NodeStatus;
 import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
 import org.apache.hadoop.hdds.server.events.EventPublisher;
+import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand;
+import org.apache.hadoop.ozone.protocol.commands.CommandForDatanode;
 import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
 import org.apache.hadoop.util.ExitUtil;
 import org.apache.ratis.protocol.exceptions.NotLeaderException;
@@ -53,6 +57,7 @@
 import java.time.Clock;
 import java.time.Duration;
 import java.util.Comparator;
+import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -64,6 +69,7 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
 
 import static org.apache.hadoop.hdds.conf.ConfigTag.OZONE;
 import static org.apache.hadoop.hdds.conf.ConfigTag.SCM;
@@ -401,6 +407,16 @@
       return new ContainerHealthResult.HealthyResult(containerInfo);
     }
 
+    if (containerInfo.getState() == HddsProtos.LifeCycleState.CLOSED) {
+      List<ContainerReplica> unhealthyReplicas = replicas.stream()
+          .filter(r -> !compareState(containerInfo.getState(), r.getState()))
+          .collect(Collectors.toList());
+
+      if (unhealthyReplicas.size() > 0) {
+        handleUnhealthyReplicas(containerInfo, unhealthyReplicas);
+      }
+    }
+
     List<ContainerReplicaOp> pendingOps =
         containerReplicaPendingOps.getPendingOps(containerID);
     ContainerHealthResult health = ecContainerHealthCheck
@@ -433,6 +449,67 @@
   }
 
   /**
+   * Handles unhealthy container.
+   * A container is inconsistent if any of the replica state doesn't
+   * match the container state. We have to take appropriate action
+   * based on state of the replica.
+   *
+   * @param container ContainerInfo
+   * @param unhealthyReplicas List of ContainerReplica
+   */
+  private void handleUnhealthyReplicas(final ContainerInfo container,
+      List<ContainerReplica> unhealthyReplicas) {
+    Iterator<ContainerReplica> iterator = unhealthyReplicas.iterator();
+    while (iterator.hasNext()) {
+      final ContainerReplica replica = iterator.next();
+      final ContainerReplicaProto.State state = replica.getState();
+      if (state == State.OPEN || state == State.CLOSING) {
+        sendCloseCommand(container, replica.getDatanodeDetails(), true);
+        iterator.remove();
+      }
+    }
+  }
+
+  /**
+   * Sends close container command for the given container to the given
+   * datanode.
+   *
+   * @param container Container to be closed
+   * @param datanode The datanode on which the container
+   *                  has to be closed
+   * @param force Should be set to true if we want to force close.
+   */
+  private void sendCloseCommand(final ContainerInfo container,
+      final DatanodeDetails datanode, final boolean force) {
+
+    ContainerID containerID = container.containerID();
+    LOG.info("Sending close container command for container {}" +
+        " to datanode {}.", containerID, datanode);
+    CloseContainerCommand closeContainerCommand =
+        new CloseContainerCommand(container.getContainerID(),
+            container.getPipelineID(), force);
+    try {
+      closeContainerCommand.setTerm(scmContext.getTermOfLeader());
+    } catch (NotLeaderException nle) {
+      LOG.warn("Skip sending close container command,"
+          + " since current SCM is not leader.", nle);
+      return;
+    }
+    closeContainerCommand.setEncodedToken(getContainerToken(containerID));
+    eventPublisher.fireEvent(SCMEvents.DATANODE_COMMAND,
+        new CommandForDatanode<>(datanode.getUuid(), closeContainerCommand));
+  }
+
+  private String getContainerToken(ContainerID containerID) {
+    if (scmContext.getScm() instanceof StorageContainerManager) {
+      StorageContainerManager scm =
+          (StorageContainerManager) scmContext.getScm();
+      return scm.getContainerTokenGenerator().generateEncodedToken(containerID);
+    }
+    return ""; // unit test
+  }
+
+  /**
    * Creates a priority queue of UnderReplicatedHealthResult, where the elements
    * are ordered by the weighted redundancy of the container. This means that
    * containers with the least remaining redundancy are at the front of the
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/TestECContainerRecovery.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/TestECContainerRecovery.java
index da49bef..38ec64c 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/TestECContainerRecovery.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/TestECContainerRecovery.java
@@ -23,15 +23,17 @@
 import org.apache.hadoop.hdds.client.ReplicationType;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
 import org.apache.hadoop.hdds.scm.OzoneClientConfig;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerNotFoundException;
+import org.apache.hadoop.hdds.scm.container.ContainerReplica;
+import org.apache.hadoop.hdds.scm.container.replication.ReplicationManager;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
-import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol;
 import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
-import org.apache.hadoop.hdds.utils.HAUtils;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.client.BucketArgs;
@@ -43,9 +45,9 @@
 import org.apache.hadoop.ozone.client.io.ECKeyOutputStream;
 import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
 import org.apache.ozone.test.GenericTestUtils;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -54,6 +56,7 @@
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.HDDS_SCM_WATCHER_TIMEOUT;
@@ -78,7 +81,7 @@
   /**
    * Create a MiniDFSCluster for testing.
    */
-  @BeforeClass
+  @BeforeAll
   public static void init() throws Exception {
     chunkSize = 1024;
     flushSize = 2 * chunkSize;
@@ -125,7 +128,7 @@
   /**
    * Shutdown MiniDFSCluster.
    */
-  @AfterClass
+  @AfterAll
   public static void shutdown() {
     if (cluster != null) {
       cluster.shutdown();
@@ -180,30 +183,30 @@
     List<ContainerInfo> containers =
         cluster.getStorageContainerManager().getContainerManager()
             .getContainers();
-    long containerID = 0;
+    ContainerInfo container = null;
     for (ContainerInfo info : containers) {
       if (info.getPipelineID().getId().equals(pipeline.getId().getId())) {
-        containerID = info.containerID().getProtobuf().getId();
+        container = info;
       }
     }
     StorageContainerManager scm = cluster.getStorageContainerManager();
-    StorageContainerLocationProtocol scmContainerClient =
-        HAUtils.getScmContainerClient(cluster.getConf());
-    scmContainerClient.closeContainer(containerID);
-    // Make sure replica closed.
-    waitForDNContainerState(ContainerProtos.ContainerDataProto.State.CLOSED,
-        pipeline, containerID);
+
+    // Shutting sown DN triggers close pipeline and close container.
+    cluster.shutdownHddsDatanode(pipeline.getFirstNode());
+
+    // Make sure container closed.
+    waitForSCMContainerState(StorageContainerDatanodeProtocolProtos
+        .ContainerReplicaProto.State.CLOSED, container.containerID());
     //Temporarily stop the RM process.
     scm.getReplicationManager().stop();
 
-    // Stop the DN and wait for the lower replication.
-    cluster.shutdownHddsDatanode(pipeline.getFirstNode());
-    waitForContainerCount(4, containerID, scm);
+    // Wait for the lower replication.
+    waitForContainerCount(4, container.containerID(), scm);
 
     // Start the RM to resume the replication process and wait for the
     // reconstruction.
     scm.getReplicationManager().start();
-    waitForContainerCount(5, containerID, scm);
+    waitForContainerCount(5, container.containerID(), scm);
 
     // Let's verify for Over replications now.
     //Temporarily stop the RM process.
@@ -213,35 +216,56 @@
     // increased.
     cluster.restartHddsDatanode(pipeline.getFirstNode(), true);
     // Check container is over replicated.
-    waitForContainerCount(6, containerID, scm);
+    waitForContainerCount(6, container.containerID(), scm);
+    // Wait for all the replicas to be closed.
+    container = scm.getContainerInfo(container.getContainerID());
+    waitForDNContainerState(container, scm);
 
     // Resume RM and wait the over replicated replica deleted.
     scm.getReplicationManager().start();
-    waitForContainerCount(5, containerID, scm);
+    waitForContainerCount(5, container.containerID(), scm);
   }
 
-  private void waitForDNContainerState(
-      ContainerProtos.ContainerDataProto.State state, Pipeline pipeline,
-      long containerID) throws TimeoutException, InterruptedException {
-    //Wait until container closed at DN
+  private void waitForDNContainerState(ContainerInfo container,
+      StorageContainerManager scm) throws InterruptedException,
+      TimeoutException {
     GenericTestUtils.waitFor(() -> {
       try {
-        return cluster.getHddsDatanode(pipeline.getFirstNode())
-            .getDatanodeStateMachine().getContainer().getContainerSet()
-            .getContainer(containerID).getContainerState() == state;
+        List<ContainerReplica> unhealthyReplicas = scm.getContainerManager()
+            .getContainerReplicas(container.containerID()).stream()
+            .filter(r -> !ReplicationManager
+                .compareState(container.getState(), r.getState()))
+            .collect(Collectors.toList());
+        return unhealthyReplicas.size() == 0;
+      } catch (ContainerNotFoundException e) {
+        return false;
+      }
+    }, 100, 100000);
+  }
+
+  private void waitForSCMContainerState(
+      StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State state,
+      ContainerID containerID) throws TimeoutException, InterruptedException {
+    //Wait until container closed at SCM
+    GenericTestUtils.waitFor(() -> {
+      try {
+        HddsProtos.LifeCycleState containerState = cluster
+            .getStorageContainerManager().getContainerManager()
+            .getContainer(containerID).getState();
+        return ReplicationManager.compareState(containerState, state);
       } catch (IOException e) {
         return false;
       }
     }, 100, 100000);
   }
 
-  private void waitForContainerCount(int count, long containerID,
+  private void waitForContainerCount(int count, ContainerID containerID,
       StorageContainerManager scm)
       throws TimeoutException, InterruptedException {
     GenericTestUtils.waitFor(() -> {
       try {
         return scm.getContainerManager()
-            .getContainerReplicas(ContainerID.valueOf(containerID))
+            .getContainerReplicas(containerID)
             .size() == count;
       } catch (ContainerNotFoundException e) {
         return false;