HDFS-15641. DataNode could meet deadlock if invoke refreshNameNode. Contributed by Hongbing Wang.

(cherry picked from commit 12c908c827c80ea37ed5207a65831d0b699c8381)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPOfferService.java

@@ -206,6 +206,7 @@ class BPOfferService {
     if (id != null) {
       return id;
     }
+    DataNodeFaultInjector.get().delayWhenOfferServiceHoldLock();
     readLock();
     try {
       if (bpNSInfo != null) {
@@ -382,6 +383,7 @@ class BPOfferService {
     }
 
     try {
+      DataNodeFaultInjector.get().delayWhenOfferServiceHoldLock();
       if (setNamespaceInfo(nsInfo) == null) {
         boolean success = false;
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java

@@ -572,11 +572,11 @@ class BPServiceActor implements Runnable {
     }
     bpThread = new Thread(this);
     bpThread.setDaemon(true); // needed for JUnit testing
-    bpThread.start();
 
     if (lifelineSender != null) {
       lifelineSender.start();
     }
+    bpThread.start();
   }
 
   private String formatThreadName(

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestRefreshNamenodes.java

@@ -26,6 +26,8 @@ import java.net.InetSocketAddress;
 import java.util.Set;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.MiniDFSNNTopology;
 import org.apache.hadoop.hdfs.MiniDFSNNTopology.NNConf;
@@ -93,4 +95,36 @@ public class TestRefreshNamenodes {
       }
     }
   }
+
+  @Test(timeout=10000)
+  public void testRefreshNameNodeDeadLock() throws Exception {
+    Configuration conf = new HdfsConfiguration();
+    MiniDFSCluster cluster = null;
+    try {
+      cluster = new MiniDFSCluster.Builder(conf).numDataNodes(3).build();
+      cluster.waitActive();
+
+      DataNodeFaultInjector.set(new DataNodeFaultInjector() {
+        @Override
+        public void delayWhenOfferServiceHoldLock() {
+          try {
+            Thread.sleep(1000);
+          } catch (InterruptedException e) {
+            e.printStackTrace();
+          }
+        }
+      });
+
+      DataNode dn = cluster.getDataNodes().get(0);
+      Configuration dnConf = dn.getConf();
+      dnConf.set(DFSConfigKeys.DFS_NAMESERVICES, "ns1");
+      dnConf.set(DFSConfigKeys.DFS_NAMENODE_LIFELINE_RPC_ADDRESS_KEY + ".ns1",
+          "mock:8022");
+      dn.refreshNamenodes(dnConf);
+    } finally {
+      if (cluster != null) {
+        cluster.shutdown();
+      }
+    }
+  }
 }