Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

merge MAPREDUCE-3872 from trunk. Fix an event handling races in ContainerLauncherImpl. Contributed by Robert Kanter. 

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1459548 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Siddharth Seth 2013-03-21 21:57:12 +00:00
parent 92af1b7d8d
commit 7892e1aabd
2 changed files with 7 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hadoop-mapreduce-project/CHANGES.txt
View File

@ -61,6 +61,9 @@ Release 2.0.5-beta - UNRELEASED
MAPREDUCE-5075. DistCp leaks input file handles since ThrottledInputStream MAPREDUCE-5075. DistCp leaks input file handles since ThrottledInputStream
does not close the wrapped InputStream. (Chris Nauroth via szetszwo) 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 Release 2.0.4-alpha - UNRELEASED
INCOMPATIBLE CHANGES INCOMPATIBLE CHANGES

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

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