HDDS-557. DeadNodeHandler should handle exception from removeContainerHandler api. Contributed by Ajay Kumar.

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DeadNodeHandler.java

@@ -72,8 +72,13 @@ public void onMessage(DatanodeDetails datanodeDetails,
         datanodeDetails.getUuid());
     for (ContainerID container : containers) {
       try {
-        containerStateManager.removeContainerReplica(container,
+        try {
-            datanodeDetails);
+          containerStateManager.removeContainerReplica(container,
+              datanodeDetails);
+        } catch (SCMException ex) {
+          LOG.info("DataNode {} doesn't have replica for container {}.",
+              datanodeDetails.getUuid(), container.getId());
+        }
 
         if (!containerStateManager.isOpen(container)) {
           ReplicationRequest replicationRequest =
@@ -90,4 +95,11 @@ public void onMessage(DatanodeDetails datanodeDetails,
       }
     }
   }
+
+  /**
+   * Returns logger.
+   * */
+  public static Logger getLogger() {
+    return LOG;
+  }
 }

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestDeadNodeHandler.java

@@ -196,6 +196,29 @@ public void testStatisticsUpdate() throws Exception {
     Assert.assertTrue(nodeStat.get().getScmUsed().get() == 0);
   }
 
+  @Test
+  public void testOnMessageReplicaFailure() throws Exception {
+    DatanodeDetails dn1 = TestUtils.randomDatanodeDetails();
+    GenericTestUtils.LogCapturer logCapturer = GenericTestUtils.LogCapturer
+        .captureLogs(DeadNodeHandler.getLogger());
+    String storagePath1 = GenericTestUtils.getRandomizedTempPath()
+        .concat("/" + dn1.getUuidString());
+
+    StorageReportProto storageOne = TestUtils.createStorageReport(
+        dn1.getUuid(), storagePath1, 100, 10, 90, null);
+    nodeReportHandler.onMessage(getNodeReport(dn1, storageOne),
+        Mockito.mock(EventPublisher.class));
+
+    ContainerInfo container1 =
+        TestUtils.allocateContainer(containerStateManager);
+    registerReplicas(node2ContainerMap, dn1, container1);
+
+    deadNodeHandler.onMessage(dn1, eventQueue);
+    Assert.assertTrue(logCapturer.getOutput().contains(
+        "DataNode " + dn1.getUuid() + " doesn't have replica for container "
+            + container1.getContainerID()));
+  }
+
   private void registerReplicas(ContainerStateManager csm,
       ContainerInfo container, DatanodeDetails... datanodes) {
     csm.getContainerStateMap()