You are viewing a plain text version of this content. The canonical link for it is here.
Posted to mapreduce-commits@hadoop.apache.org by to...@apache.org on 2012/01/18 19:21:46 UTC
svn commit: r1232982 - in
/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project: ./
hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/
Author: tomwhite
Date: Wed Jan 18 18:21:45 2012
New Revision: 1232982
URL: http://svn.apache.org/viewvc?rev=1232982&view=rev
Log:
Merge -r 1232980:1232981 from trunk to branch-0.23. Fixes: MAPREDUCE-3684
Modified:
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalDistributedCacheManager.java
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalJobRunner.java
Modified: hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt?rev=1232982&r1=1232981&r2=1232982&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt (original)
+++ hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt Wed Jan 18 18:21:45 2012
@@ -453,7 +453,10 @@ Release 0.23.1 - Unreleased
leading to a long timeout in Task. (Rajesh Balamohan via acmurthy)
MAPREDUCE-3669. Allow clients to talk to MR HistoryServer using both
- delegation tokens and kerberos. (mahadev via acmurthy)
+ delegation tokens and kerberos. (mahadev via acmurthy)
+
+ MAPREDUCE-3684. LocalDistributedCacheManager does not shut down its thread
+ pool (tomwhite)
Release 0.23.0 - 2011-11-01
Modified: hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalDistributedCacheManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalDistributedCacheManager.java?rev=1232982&r1=1232981&r2=1232982&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalDistributedCacheManager.java (original)
+++ hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalDistributedCacheManager.java Wed Jan 18 18:21:45 2012
@@ -39,6 +39,7 @@ import java.util.concurrent.ExecutionExc
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -57,6 +58,8 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.FSDownload;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
/**
* A helper class for managing the distributed cache for {@link LocalJobRunner}.
*/
@@ -111,43 +114,52 @@ class LocalDistributedCacheManager {
FileContext localFSFileContext = FileContext.getLocalFSFileContext();
UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
- Map<LocalResource, Future<Path>> resourcesToPaths = Maps.newHashMap();
- ExecutorService exec = Executors.newCachedThreadPool();
- Path destPath = localDirAllocator.getLocalPathForWrite(".", conf);
- for (LocalResource resource : localResources.values()) {
- Callable<Path> download = new FSDownload(localFSFileContext, ugi, conf,
- destPath, resource, new Random());
- Future<Path> future = exec.submit(download);
- resourcesToPaths.put(resource, future);
- }
- for (LocalResource resource : localResources.values()) {
- Path path;
- try {
- path = resourcesToPaths.get(resource).get();
- } catch (InterruptedException e) {
- throw new IOException(e);
- } catch (ExecutionException e) {
- throw new IOException(e);
- }
- String pathString = path.toUri().toString();
- if (resource.getType() == LocalResourceType.ARCHIVE) {
- localArchives.add(pathString);
- } else if (resource.getType() == LocalResourceType.FILE) {
- localFiles.add(pathString);
- }
- Path resourcePath;
- try {
- resourcePath = ConverterUtils.getPathFromYarnURL(resource.getResource());
- } catch (URISyntaxException e) {
- throw new IOException(e);
- }
- LOG.info(String.format("Localized %s as %s", resourcePath, path));
- String cp = resourcePath.toUri().getPath();
- if (classpaths.keySet().contains(cp)) {
- localClasspaths.add(path.toUri().getPath().toString());
+ ExecutorService exec = null;
+ try {
+ ThreadFactory tf = new ThreadFactoryBuilder()
+ .setNameFormat("LocalDistributedCacheManager Downloader #%d")
+ .build();
+ exec = Executors.newCachedThreadPool(tf);
+ Path destPath = localDirAllocator.getLocalPathForWrite(".", conf);
+ Map<LocalResource, Future<Path>> resourcesToPaths = Maps.newHashMap();
+ for (LocalResource resource : localResources.values()) {
+ Callable<Path> download = new FSDownload(localFSFileContext, ugi, conf,
+ destPath, resource, new Random());
+ Future<Path> future = exec.submit(download);
+ resourcesToPaths.put(resource, future);
+ }
+ for (LocalResource resource : localResources.values()) {
+ Path path;
+ try {
+ path = resourcesToPaths.get(resource).get();
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ } catch (ExecutionException e) {
+ throw new IOException(e);
+ }
+ String pathString = path.toUri().toString();
+ if (resource.getType() == LocalResourceType.ARCHIVE) {
+ localArchives.add(pathString);
+ } else if (resource.getType() == LocalResourceType.FILE) {
+ localFiles.add(pathString);
+ }
+ Path resourcePath;
+ try {
+ resourcePath = ConverterUtils.getPathFromYarnURL(resource.getResource());
+ } catch (URISyntaxException e) {
+ throw new IOException(e);
+ }
+ LOG.info(String.format("Localized %s as %s", resourcePath, path));
+ String cp = resourcePath.toUri().getPath();
+ if (classpaths.keySet().contains(cp)) {
+ localClasspaths.add(path.toUri().getPath().toString());
+ }
+ }
+ } finally {
+ if (exec != null) {
+ exec.shutdown();
}
- }
-
+ }
// Update the configuration object with localized data.
if (!localArchives.isEmpty()) {
conf.set(MRJobConfig.CACHE_LOCALARCHIVES, StringUtils
@@ -171,7 +183,7 @@ class LocalDistributedCacheManager {
}
setupCalled = true;
}
-
+
/**
* Are the resources that should be added to the classpath?
* Should be called after setup().
Modified: hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalJobRunner.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalJobRunner.java?rev=1232982&r1=1232981&r2=1232982&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalJobRunner.java (original)
+++ hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalJobRunner.java Wed Jan 18 18:21:45 2012
@@ -28,6 +28,7 @@ import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -61,6 +62,8 @@ import org.apache.hadoop.security.author
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.util.ReflectionUtils;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
/** Implements MapReduce locally, in-process, for debugging. */
@InterfaceAudience.Private
@InterfaceStability.Unstable
@@ -302,7 +305,10 @@ public class LocalJobRunner implements C
LOG.debug("Map tasks to process: " + this.numMapTasks);
// Create a new executor service to drain the work queue.
- ExecutorService executor = Executors.newFixedThreadPool(maxMapThreads);
+ ThreadFactory tf = new ThreadFactoryBuilder()
+ .setNameFormat("LocalJobRunner Map Task Executor #%d")
+ .build();
+ ExecutorService executor = Executors.newFixedThreadPool(maxMapThreads, tf);
return executor;
}