MAPREDUCE-3872. Fix an event handling races in ContainerLauncherImpl. (Contributed by Robert Kanter)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1459547 13f79535-47bb-0310-9956-ffa450edef68

hadoop-mapreduce-project/CHANGES.txt

@@ -244,6 +244,9 @@ Release 2.0.5-beta - UNRELEASED
     MAPREDUCE-5075. DistCp leaks input file handles since ThrottledInputStream
     does not close the wrapped InputStream.  (Chris Nauroth via szetszwo)
 
+    MAPREDUCE-3872. Fix an event handling races in ContainerLauncherImpl.
+    (Robert Kanter via sseth)
+
 Release 2.0.4-alpha - UNRELEASED
 
   INCOMPATIBLE CHANGES

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/launcher/ContainerLauncherImpl.java

@@ -230,9 +230,6 @@ public class ContainerLauncherImpl extends AbstractService implements
     }
   }
 
-  // To track numNodes.
-  Set<String> allNodes = new HashSet<String>();
-
   public ContainerLauncherImpl(AppContext context) {
     super(ContainerLauncherImpl.class.getName());
     this.context = context;
@@ -271,6 +268,8 @@ public class ContainerLauncherImpl extends AbstractService implements
       @Override
       public void run() {
         ContainerLauncherEvent event = null;
+        Set<String> allNodes = new HashSet<String>();
+
         while (!stopped.get() && !Thread.currentThread().isInterrupted()) {
           try {
             event = eventQueue.take();
@@ -280,6 +279,8 @@ public class ContainerLauncherImpl extends AbstractService implements
             }
             return;
           }
+          allNodes.add(event.getContainerMgrAddress());
+
           int poolSize = launcherPool.getCorePoolSize();
 
           // See if we need up the pool size only if haven't reached the
@@ -419,7 +420,6 @@ public class ContainerLauncherImpl extends AbstractService implements
   public void handle(ContainerLauncherEvent event) {
     try {
       eventQueue.put(event);
-      this.allNodes.add(event.getContainerMgrAddress());
     } catch (InterruptedException e) {
       throw new YarnException(e);
     }