HDFS-12703. Exceptions are fatal to decommissioning monitor. Contributed by He Xiaoqiao.

(cherry picked from commit 3d396786cf6eaab49c1c9b8b2a4652c2e440b9e3)
(cherry picked from commit 950aa74d5f)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminManager.java

@@ -485,6 +485,7 @@ public class DatanodeAdminManager {
 
     @Override
     public void run() {
+      LOG.debug("DatanodeAdminMonitor is running.");
       if (!namesystem.isRunning()) {
         LOG.info("Namesystem is not running, skipping " +
             "decommissioning/maintenance checks.");
@@ -499,6 +500,9 @@ public class DatanodeAdminManager {
       try {
         processPendingNodes();
         check();
+      } catch (Exception e) {
+        LOG.warn("DatanodeAdminMonitor caught exception when processing node.",
+            e);
       } finally {
         namesystem.writeUnlock();
       }
@@ -532,6 +536,7 @@ public class DatanodeAdminManager {
         final Map.Entry<DatanodeDescriptor, AbstractList<BlockInfo>>
             entry = it.next();
         final DatanodeDescriptor dn = entry.getKey();
+        try {
           AbstractList<BlockInfo> blocks = entry.getValue();
           boolean fullScan = false;
           if (dn.isMaintenance() && dn.maintenanceExpired()) {
@@ -587,7 +592,9 @@ public class DatanodeAdminManager {
                 setInMaintenance(dn);
               } else {
                 Preconditions.checkState(false,
-                  "A node is in an invalid state!");
+                    "Node %s is in an invalid state! "
+                      + "Invalid state: %s %s blocks are on this dn.",
+                        dn, dn.getAdminState(), blocks.size());
               }
               LOG.debug("Node {} is sufficiently replicated and healthy, "
                   + "marked as {}.", dn, dn.getAdminState());
@@ -602,13 +609,23 @@ public class DatanodeAdminManager {
                 + "before it is a candidate to finish {}.",
                 dn, blocks.size(), dn.getAdminState());
           }
+        } catch (Exception e) {
+          // Log and postpone to process node when meet exception since it is in
+          // an invalid state.
+          LOG.warn("DatanodeAdminMonitor caught exception when processing node "
+              + "{}.", dn, e);
+          pendingNodes.add(dn);
+          toRemove.add(dn);
+        } finally {
           iterkey = dn;
         }
+      }
       // Remove the datanodes that are DECOMMISSIONED or in service after
       // maintenance expiration.
       for (DatanodeDescriptor dn : toRemove) {
         Preconditions.checkState(dn.isDecommissioned() || dn.isInService(),
-            "Removing a node that is not yet decommissioned or in service!");
+            "Removing node %s that is not yet decommissioned or in service!",
+                dn);
         outOfServiceNodeBlocks.remove(dn);
       }
     }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommission.java

@@ -21,6 +21,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -35,6 +36,7 @@ import java.util.Map;
 import java.util.Scanner;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Pattern;
 
 import com.google.common.base.Supplier;
 import com.google.common.collect.Lists;
@@ -1181,6 +1183,56 @@ public class TestDecommission extends AdminStatesBaseTest {
     }
   }
 
+  /**
+   * Test DatanodeAdminManager#monitor can swallow any exceptions by default.
+   */
+  @Test(timeout=120000)
+  public void testPendingNodeButDecommissioned() throws Exception {
+    // Only allow one node to be decom'd at a time
+    getConf().setInt(
+        DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES,
+        1);
+    // Disable the normal monitor runs
+    getConf().setInt(DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_INTERVAL_KEY,
+        Integer.MAX_VALUE);
+    startCluster(1, 2);
+    final DatanodeManager datanodeManager =
+        getCluster().getNamesystem().getBlockManager().getDatanodeManager();
+    final DatanodeAdminManager decomManager =
+        datanodeManager.getDatanodeAdminManager();
+
+    ArrayList<DatanodeInfo> decommissionedNodes = Lists.newArrayList();
+    List<DataNode> dns = getCluster().getDataNodes();
+    // Try to decommission 2 datanodes
+    for (int i = 0; i < 2; i++) {
+      DataNode d = dns.get(i);
+      DatanodeInfo dn = takeNodeOutofService(0, d.getDatanodeUuid(), 0,
+          decommissionedNodes, AdminStates.DECOMMISSION_INPROGRESS);
+      decommissionedNodes.add(dn);
+    }
+
+    assertEquals(2, decomManager.getNumPendingNodes());
+
+    // Set one datanode state to Decommissioned after decommission ops.
+    DatanodeDescriptor dn = datanodeManager.getDatanode(dns.get(0)
+        .getDatanodeId());
+    dn.setDecommissioned();
+
+    try {
+      // Trigger DatanodeAdminManager#monitor
+      BlockManagerTestUtil.recheckDecommissionState(datanodeManager);
+
+      // Wait for OutOfServiceNodeBlocks to be 0
+      GenericTestUtils.waitFor(() -> decomManager.getNumTrackedNodes() == 0,
+          500, 30000);
+      assertTrue(GenericTestUtils.anyThreadMatching(
+          Pattern.compile("DatanodeAdminMonitor-.*")));
+    } catch (ExecutionException e) {
+      GenericTestUtils.assertExceptionContains("in an invalid state!", e);
+      fail("DatanodeAdminManager#monitor does not swallow exceptions.");
+    }
+  }
+
   @Test(timeout=120000)
   public void testPendingNodes() throws Exception {
     org.apache.log4j.Logger.getLogger(DatanodeAdminManager.class)