MAPREDUCE-5512. TaskTracker hung after failed reconnect to the JobTracker. Contributed by Ivan Mitic.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-1@1531200 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -151,6 +151,9 @@ Release 1.3.0 - unreleased
     MAPREDUCE-5569. FloatSplitter is not generating correct splits (Nathan
     Roberts via jlowe)
 
+    MAPREDUCE-5512. TaskTracker hung after failed reconnect to the JobTracker.
+    (ivanmi)
+
 Release 1.2.2 - unreleased
 
   INCOMPATIBLE CHANGES

src/core/org/apache/hadoop/http/HttpServer.java

@@ -160,7 +160,9 @@ public class HttpServer implements FilterContainer {
     
     webServer.addConnector(listener);
 
-    webServer.setThreadPool(new QueuedThreadPool());
+    QueuedThreadPool threadPool = new QueuedThreadPool();
+    threadPool.setName("httpServerThreadPool");
+    webServer.setThreadPool(threadPool);
 
     final String appDir = getWebAppsPath();
     ContextHandlerCollection contexts = new ContextHandlerCollection();

src/mapred/org/apache/hadoop/filecache/TrackerDistributedCacheManager.java

@@ -125,6 +125,8 @@ public class TrackerDistributedCacheManager {
 
     this.taskController = controller;
     this.cleanupThread = new CleanupThread(conf);
+    this.cleanupThread.setName("distCacheManagerCleanupThread");
+    this.cleanupThread.setDaemon(true);
   }
 
   /**

src/mapred/org/apache/hadoop/mapred/JobEndNotifier.java

@@ -83,7 +83,7 @@ public class JobEndNotifier {
               LOG.error("Notification failure [" + notification + "]", ex);
             }
           }
-        });
+        }, "jobEndNotifier");
     thread.start();
   }
 

src/test/org/apache/hadoop/mapred/MiniMRCluster.java

@@ -617,7 +617,7 @@ public class MiniMRCluster {
   void startJobTracker(boolean wait) {
     //  Create the JobTracker
     jobTracker = new JobTrackerRunner(conf);
-    jobTrackerThread = new Thread(jobTracker);
+    jobTrackerThread = new Thread(jobTracker, "jobTrackerMain");
         
     jobTrackerThread.start();
     
@@ -712,7 +712,7 @@ public class MiniMRCluster {
    * Add a tasktracker to the Mini-MR cluster.
    */
   void addTaskTracker(TaskTrackerRunner taskTracker) {
-    Thread taskTrackerThread = new Thread(taskTracker);
+    Thread taskTrackerThread = new Thread(taskTracker, "taskTrackerMain");
     taskTrackerList.add(taskTracker);
     taskTrackerThreadList.add(taskTrackerThread);
     taskTrackerThread.start();

src/test/org/apache/hadoop/mapred/TestMiniMRDaemonThreads.java

@@ -0,0 +1,79 @@
+/**
+ * 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.mapred;
+
+import java.io.IOException;
+import java.util.Set;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Validate the list of non-daemon threads running in the JobTracker
+ * and TaskTracker. Each non-daemon thread is a possible candidate for
+ * the service hang, so please make sure to validate all failure
+ * codepaths of a new non-daemon thread before making a change
+ * to this test.
+ */
+public class TestMiniMRDaemonThreads {
+
+  static String[] whitelistedDaemonKeys = {
+    "jobTrackerMain",
+    "httpServerThreadPool",
+    "taskTrackerMain",
+    "expireTrackers",
+    "jobEndNotifier",
+    "ReaderThread",
+    "Socket Reader",
+    "expireLaunchingTasks",
+    "retireJobs"
+    };
+
+  @Test
+  public void testValidateDaemonThreads() throws IOException, InterruptedException {
+    MiniMRCluster mr = null;
+    try {
+      JobConf jtConf = new JobConf();
+      mr = new MiniMRCluster(1, "file:///", 1, null, null, jtConf);
+      // Give it one second for all threads to start up
+      Thread.sleep(1000);
+
+      Set<Thread> threadSet = Thread.getAllStackTraces().keySet();
+      for (Thread t : threadSet) {
+        if (!t.isDaemon() && !Thread.currentThread().equals(t)) {
+          boolean found = false;
+          for(String key : whitelistedDaemonKeys) {
+            if (t.getName().contains(key)) {
+              found = true;
+              break;
+            }
+          }
+          if (!found) {
+            Assert.fail(
+                String.format("Thread '%s' is not a whitelisted daemon thread",
+                    t.getName()));
+          }
+        }
+      }
+
+    } finally {
+      if (mr != null) {
+        mr.shutdown();
+      }
+    }
+  }
+}