Author: vinodkv
Date: Fri May 6 10:45:08 2011
New Revision: 1100165
URL: http://svn.apache.org/viewvc?rev=1100165&view=rev
Log:
Replacing FileContext usage with FileSystem to work around security
authentication issues with FileContext against a secure DFS. Contributed by
Vinod Kumar Vavilapalli.
Modified:
hadoop/mapreduce/branches/MR-279/CHANGES.txt
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/MRAppMaster.java
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRApp.java
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ContainerLocalizer.java
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/FSDownload.java
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestContainerLocalizer.java
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestFSDownload.java
Modified: hadoop/mapreduce/branches/MR-279/CHANGES.txt
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/CHANGES.txt?rev=1100165&r1=1100164&r2=1100165&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/CHANGES.txt (original)
+++ hadoop/mapreduce/branches/MR-279/CHANGES.txt Fri May 6 10:45:08 2011
@@ -4,6 +4,11 @@ Trunk (unreleased changes)
MAPREDUCE-279
+ Replacing FileContext usage with FileSystem to work around security
authentication
+ issues with FileContext against a secure DFS. (vinodkv)
+
+ Fixing three tight-loops in RM that are causing high cpu-usage. (vinodkv)
+
Adding user log handling for YARN. Making NM put the user-logs on DFS and
providing log-dump tools. (vinodkv)
MAPREDUCE-2468. Add metrics for NM Shuffle. (Luke Lu via cdouglas)
Modified:
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java?rev=1100165&r1=1100164&r2=1100165&view=diff
==============================================================================
---
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java
(original)
+++
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java
Fri May 6 10:45:08 2011
@@ -33,7 +33,8 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.fs.CreateFlag;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileAlreadyExistsException;
-import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.mapreduce.MRJobConfig;
@@ -59,8 +60,10 @@ public class JobHistoryEventHandler exte
private final AppContext context;
private final int startCount;
- private FileContext logDirFc; // log Dir FileContext
- private FileContext doneDirFc; // done Dir FileContext
+ private FileSystem logDirFS; // log Dir FileSystem
+ private FileSystem doneDirFS; // done Dir FileSystem
+
+ private Configuration conf;
private Path logDirPath = null;
private Path doneDirPrefixPath = null; // folder for completed jobs
@@ -84,23 +87,25 @@ public class JobHistoryEventHandler exte
/* (non-Javadoc)
* @see
org.apache.hadoop.yarn.service.AbstractService#init(org.apache.hadoop.conf.Configuration)
- * Initializes the FileContext and Path objects for the log and done
directories.
+ * Initializes the FileSystem and Path objects for the log and done
directories.
* Creates these directories if they do not already exist.
*/
@Override
public void init(Configuration conf) {
-
+
+ this.conf = conf;
+
String logDir = JobHistoryUtils.getConfiguredHistoryLogDirPrefix(conf);
String doneDirPrefix =
JobHistoryUtils.getConfiguredHistoryDoneDirPrefix(conf);
try {
- doneDirPrefixPath = FileContext.getFileContext(conf).makeQualified(
+ doneDirPrefixPath = FileSystem.get(conf).makeQualified(
new Path(doneDirPrefix));
- doneDirFc = FileContext.getFileContext(doneDirPrefixPath.toUri(), conf);
- if (!doneDirFc.util().exists(doneDirPrefixPath)) {
+ doneDirFS = FileSystem.get(doneDirPrefixPath.toUri(), conf);
+ if (!doneDirFS.exists(doneDirPrefixPath)) {
try {
- doneDirFc.mkdir(doneDirPrefixPath, new FsPermission(
- JobHistoryUtils.HISTORY_DIR_PERMISSION), true);
+ doneDirFS.mkdirs(doneDirPrefixPath, new FsPermission(
+ JobHistoryUtils.HISTORY_DIR_PERMISSION));
} catch (FileAlreadyExistsException e) {
LOG.info("JobHistory Done Directory: [" + doneDirPrefixPath
+ "] already exists.");
@@ -111,13 +116,12 @@ public class JobHistoryEventHandler exte
throw new YarnException(e);
}
try {
- logDirPath = FileContext.getFileContext(conf).makeQualified(
+ logDirPath = FileSystem.get(conf).makeQualified(
new Path(logDir));
- logDirFc = FileContext.getFileContext(logDirPath.toUri(), conf);
- if (!logDirFc.util().exists(logDirPath)) {
+ logDirFS = FileSystem.get(logDirPath.toUri(), conf);
+ if (!logDirFS.exists(logDirPath)) {
try {
- logDirFc.mkdir(logDirPath, new
FsPermission(JobHistoryUtils.HISTORY_DIR_PERMISSION),
- true);
+ logDirFS.mkdirs(logDirPath, new
FsPermission(JobHistoryUtils.HISTORY_DIR_PERMISSION));
} catch (FileAlreadyExistsException e) {
LOG.info("JobHistory Log Directory: [" + doneDirPrefixPath
+ "] already exists.");
@@ -216,8 +220,7 @@ public class JobHistoryEventHandler exte
if (writer == null) {
try {
- FSDataOutputStream out = logDirFc.create(logFile,
- EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE));
+ FSDataOutputStream out = logDirFS.create(logFile, true);
//TODO Permissions for the history file?
writer = new EventWriter(out);
} catch (IOException ioe) {
@@ -234,8 +237,7 @@ public class JobHistoryEventHandler exte
FSDataOutputStream jobFileOut = null;
try {
if (logDirConfPath != null) {
- jobFileOut = logDirFc.create(logDirConfPath,
- EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE));
+ jobFileOut = logDirFS.create(logDirConfPath, true);
conf.writeXml(jobFileOut);
jobFileOut.close();
}
@@ -323,27 +325,27 @@ public class JobHistoryEventHandler exte
//
String doneDir =
JobHistoryUtils.getCurrentDoneDir(doneDirPrefixPath.toString());
Path doneDirPath =
- doneDirFc.makeQualified(new Path(doneDir));
- if (!pathExists(doneDirFc, doneDirPath)) {
+ doneDirFS.makeQualified(new Path(doneDir));
+ if (!pathExists(doneDirFS, doneDirPath)) {
try {
- doneDirFc.mkdir(doneDirPath, new
FsPermission(JobHistoryUtils.HISTORY_DIR_PERMISSION), true);
+ doneDirFS.mkdirs(doneDirPath, new
FsPermission(JobHistoryUtils.HISTORY_DIR_PERMISSION));
} catch (FileAlreadyExistsException e) {
LOG.info("Done directory: [" + doneDirPath + "] already exists.");
}
}
Path logFile = mi.getHistoryFile();
- Path qualifiedLogFile = logDirFc.makeQualified(logFile);
- Path qualifiedDoneFile = doneDirFc.makeQualified(new Path(doneDirPath,
+ Path qualifiedLogFile = logDirFS.makeQualified(logFile);
+ Path qualifiedDoneFile = doneDirFS.makeQualified(new Path(doneDirPath,
getDoneJobHistoryFileName(jobId)));
moveToDoneNow(qualifiedLogFile, qualifiedDoneFile);
Path confFile = mi.getConfFile();
- Path qualifiedConfFile = logDirFc.makeQualified(confFile);
- Path qualifiedConfDoneFile = doneDirFc.makeQualified(new
Path(doneDirPath, getDoneConfFileName(jobId)));
+ Path qualifiedConfFile = logDirFS.makeQualified(confFile);
+ Path qualifiedConfDoneFile = doneDirFS.makeQualified(new
Path(doneDirPath, getDoneConfFileName(jobId)));
moveToDoneNow(qualifiedConfFile, qualifiedConfDoneFile);
- logDirFc.delete(qualifiedLogFile, true);
- logDirFc.delete(qualifiedConfFile, true);
+ logDirFS.delete(qualifiedLogFile, true);
+ logDirFS.delete(qualifiedConfFile, true);
} catch (IOException e) {
LOG.info("Error closing writer for JobID: " + jobId);
throw e;
@@ -413,35 +415,35 @@ public class JobHistoryEventHandler exte
// Currently JHEventHandler is moving files to the final location.
private void moveToDoneNow(Path fromPath, Path toPath) throws IOException {
//check if path exists, in case of retries it may not exist
- if (logDirFc.util().exists(fromPath)) {
+ if (logDirFS.exists(fromPath)) {
LOG.info("Moving " + fromPath.toString() + " to " +
toPath.toString());
//TODO temporarily removing the existing dst
- if (logDirFc.util().exists(toPath)) {
- logDirFc.delete(toPath, true);
+ if (doneDirFS.exists(toPath)) {
+ doneDirFS.delete(toPath, true);
}
- boolean copied = logDirFc.util().copy(fromPath, toPath);
+ boolean copied =
+ FileUtil.copy(logDirFS, fromPath, doneDirFS, toPath, false, conf);
if (copied)
LOG.info("Copied to done location: "+ toPath);
else
LOG.info("copy failed");
- doneDirFc.setPermission(toPath,
+ doneDirFS.setPermission(toPath,
new FsPermission(JobHistoryUtils.HISTORY_FILE_PERMISSION));
}
}
- boolean pathExists(FileContext fc, Path path) throws IOException {
- return fc.util().exists(path);
+ boolean pathExists(FileSystem fileSys, Path path) throws IOException {
+ return fileSys.exists(path);
}
private void writeStatus(String statusstoredir, HistoryEvent event) throws
IOException {
try {
- Path statusstorepath = doneDirFc.makeQualified(new Path(statusstoredir));
- doneDirFc.mkdir(statusstorepath,
- new FsPermission(JobHistoryUtils.HISTORY_DIR_PERMISSION), true);
+ Path statusstorepath = doneDirFS.makeQualified(new Path(statusstoredir));
+ doneDirFS.mkdirs(statusstorepath,
+ new FsPermission(JobHistoryUtils.HISTORY_DIR_PERMISSION));
Path toPath = new Path(statusstoredir, "jobstats");
- FSDataOutputStream out = doneDirFc.create(toPath, EnumSet
- .of(CreateFlag.CREATE, CreateFlag.OVERWRITE));
+ FSDataOutputStream out = doneDirFS.create(toPath, true);
EventWriter writer = new EventWriter(out);
writer.write(event);
writer.close();
Modified:
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/MRAppMaster.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/MRAppMaster.java?rev=1100165&r1=1100164&r2=1100165&view=diff
==============================================================================
---
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/MRAppMaster.java
(original)
+++
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/MRAppMaster.java
Fri May 6 10:45:08 2011
@@ -191,13 +191,40 @@ public class MRAppMaster extends Composi
(Speculator.EventType.class, new NullSpeculatorEventHandler());
}
+ Credentials fsTokens = new Credentials();
+ if (UserGroupInformation.isSecurityEnabled()) {
+ // Read the file-system tokens from the localized tokens-file.
+ try {
+ Path jobSubmitDir =
+ FileContext.getLocalFSFileContext().makeQualified(
+ new Path(new File(YARNApplicationConstants.JOB_SUBMIT_DIR)
+ .getAbsolutePath()));
+ Path jobTokenFile =
+ new Path(jobSubmitDir, YarnConfiguration.APPLICATION_TOKENS_FILE);
+ fsTokens.addAll(Credentials.readTokenStorageFile(jobTokenFile, conf));
+ LOG.info("jobSubmitDir=" + jobSubmitDir + " jobTokenFile="
+ + jobTokenFile);
+
+ UserGroupInformation currentUser =
+ UserGroupInformation.getCurrentUser();
+ for (Token<? extends TokenIdentifier> tk : fsTokens.getAllTokens()) {
+ LOG.info(" --- DEBUG: Token of kind " + tk.getKind()
+ + "in current ugi in the AppMaster for service "
+ + tk.getService());
+ currentUser.addToken(tk); // For use by AppMaster itself.
+ }
+ } catch (IOException e) {
+ throw new YarnException(e);
+ }
+ }
+
super.init(conf);
//---- start of what used to be startJobs() code:
Configuration config = getConfig();
- job = createJob(config);
+ job = createJob(config, fsTokens);
/** create a job event for job intialization */
JobEvent initJobEvent = new JobEvent(job.getID(), JobEventType.JOB_INIT);
@@ -236,35 +263,9 @@ public class MRAppMaster extends Composi
} // end of init()
- /** Create and initialize (but don't start) a single job. */
- public Job createJob(Configuration conf) {
- Credentials fsTokens = new Credentials();
-
- if (UserGroupInformation.isSecurityEnabled()) {
- // Read the file-system tokens from the localized tokens-file.
- try {
- Path jobSubmitDir =
- FileContext.getLocalFSFileContext().makeQualified(
- new Path(new File(YARNApplicationConstants.JOB_SUBMIT_DIR)
- .getAbsolutePath()));
- Path jobTokenFile =
- new Path(jobSubmitDir, YarnConfiguration.APPLICATION_TOKENS_FILE);
- fsTokens.addAll(Credentials.readTokenStorageFile(jobTokenFile, conf));
- LOG.info("jobSubmitDir=" + jobSubmitDir + " jobTokenFile="
- + jobTokenFile);
-
- UserGroupInformation currentUser =
- UserGroupInformation.getCurrentUser();
- for (Token<? extends TokenIdentifier> tk : fsTokens.getAllTokens()) {
- LOG.info(" --- DEBUG: Token of kind " + tk.getKind()
- + "in current ugi in the AppMaster for service "
- + tk.getService());
- currentUser.addToken(tk); // For use by AppMaster itself.
- }
- } catch (IOException e) {
- throw new YarnException(e);
- }
- }
+ /** Create and initialize (but don't start) a single job.
+ * @param fsTokens */
+ protected Job createJob(Configuration conf, Credentials fsTokens) {
// create single job
Job newJob = new JobImpl(appID, conf, dispatcher.getEventHandler(),
Modified:
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRApp.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRApp.java?rev=1100165&r1=1100164&r2=1100165&view=diff
==============================================================================
---
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRApp.java
(original)
+++
hadoop/mapreduce/branches/MR-279/mr-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRApp.java
Fri May 6 10:45:08 2011
@@ -192,7 +192,8 @@ public class MRApp extends MRAppMaster {
}
}
- public Job createJob(Configuration conf) {
+ @Override
+ protected Job createJob(Configuration conf, Credentials fsTokens) {
Job newJob = new TestJob(getAppID(), getDispatcher().getEventHandler(),
getTaskAttemptListener(),
getContext().getClock());
((AppContext) getContext()).getAllJobs().put(newJob.getID(), newJob);
Modified:
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ContainerLocalizer.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ContainerLocalizer.java?rev=1100165&r1=1100164&r2=1100165&view=diff
==============================================================================
---
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ContainerLocalizer.java
(original)
+++
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ContainerLocalizer.java
Fri May 6 10:45:08 2011
@@ -185,38 +185,36 @@ public class ContainerLocalizer {
ugi.addToken(token);
}
- return ugi.doAs(new PrivilegedExceptionAction<Integer>() {
- public Integer run() {
- ExecutorService exec = null;
- try {
- exec = createDownloadThreadPool();
- localizeFiles(nodeManager, exec);
- return 0;
- } catch (Throwable e) {
- e.printStackTrace(System.out);
- return -1;
- } finally {
- if (exec != null) {
- exec.shutdownNow();
- }
- }
+ ExecutorService exec = null;
+ try {
+ exec = createDownloadThreadPool();
+ localizeFiles(nodeManager, exec, ugi);
+ return 0;
+ } catch (Throwable e) {
+ e.printStackTrace(System.out);
+ return -1;
+ } finally {
+ if (exec != null) {
+ exec.shutdownNow();
}
- });
+ }
}
ExecutorService createDownloadThreadPool() {
return Executors.newSingleThreadExecutor();
}
- Callable<Path> download(LocalDirAllocator lda, LocalResource rsrc) {
- return new FSDownload(lfs, conf, lda, rsrc, new Random());
+ Callable<Path> download(LocalDirAllocator lda, LocalResource rsrc,
+ UserGroupInformation ugi) {
+ return new FSDownload(lfs, ugi, conf, lda, rsrc, new Random());
}
void sleep(int duration) throws InterruptedException {
TimeUnit.SECONDS.sleep(duration);
}
- void localizeFiles(LocalizationProtocol nodemanager, ExecutorService exec) {
+ void localizeFiles(LocalizationProtocol nodemanager, ExecutorService exec,
+ UserGroupInformation ugi) {
while (true) {
try {
LocalizerStatus status = createStatus();
@@ -238,7 +236,7 @@ public class ContainerLocalizer {
lda = appDirs;
break;
}
- pendingResources.put(r, exec.submit(download(lda, r)));
+ pendingResources.put(r, exec.submit(download(lda, r, ugi)));
}
}
break;
Modified:
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/FSDownload.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/FSDownload.java?rev=1100165&r1=1100164&r2=1100165&view=diff
==============================================================================
---
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/FSDownload.java
(original)
+++
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/FSDownload.java
Fri May 6 10:45:08 2011
@@ -22,25 +22,25 @@ import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URISyntaxException;
+import java.security.PrivilegedExceptionAction;
import java.util.Random;
import java.util.concurrent.Callable;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.LocalDirAllocator;
+import org.apache.hadoop.fs.Options.Rename;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.RunJar;
-import org.apache.hadoop.yarn.util.ConverterUtils;
-
-import static org.apache.hadoop.fs.Options.*;
-
import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.util.ConverterUtils;
/**
* Download a single URL to the local disk.
@@ -52,22 +52,18 @@ public class FSDownload implements Calla
private Random rand;
private FileContext files;
+ private final UserGroupInformation userUgi;
private Configuration conf;
private LocalResource resource;
private LocalDirAllocator dirs;
private FsPermission cachePerms = new FsPermission((short) 0755);
- public FSDownload(Configuration conf, LocalDirAllocator dirs,
- LocalResource resource) throws IOException {
- this(FileContext.getLocalFSFileContext(conf), conf, dirs, resource,
- new Random());
- }
-
- FSDownload(FileContext files, Configuration conf, LocalDirAllocator dirs,
- LocalResource resource, Random rand) {
+ FSDownload(FileContext files, UserGroupInformation ugi, Configuration conf,
+ LocalDirAllocator dirs, LocalResource resource, Random rand) {
this.conf = conf;
this.dirs = dirs;
this.files = files;
+ this.userUgi = ugi;
this.resource = resource;
this.rand = rand;
}
@@ -78,13 +74,14 @@ public class FSDownload implements Calla
private Path copy(Path sCopy, Path dstdir) throws IOException {
Path dCopy = new Path(dstdir, sCopy.getName() + ".tmp");
- FileStatus sStat = files.getFileStatus(sCopy);
+ FileSystem fs = FileSystem.get(new Configuration());
+ FileStatus sStat = fs.getFileStatus(sCopy);
if (sStat.getModificationTime() != resource.getTimestamp()) {
throw new IOException("Resource " + sCopy +
" changed on src filesystem (expected " + resource.getTimestamp() +
", was " + sStat.getModificationTime());
}
- files.util().copy(sCopy, dCopy);
+ fs.copyToLocalFile(sCopy, dCopy);
return dCopy;
}
@@ -115,8 +112,8 @@ public class FSDownload implements Calla
}
@Override
- public Path call() throws IOException {
- Path sCopy;
+ public Path call() throws Exception {
+ final Path sCopy;
try {
sCopy = ConverterUtils.getPathFromYarnURL(resource.getResource());
} catch (URISyntaxException e) {
@@ -132,15 +129,19 @@ public class FSDownload implements Calla
} while (files.util().exists(tmp));
dst = tmp;
files.mkdir(dst, cachePerms, false);
- Path dst_work = new Path(dst + "_tmp");
+ final Path dst_work = new Path(dst + "_tmp");
files.mkdir(dst_work, cachePerms, false);
Path dFinal = files.makeQualified(new Path(dst_work, sCopy.getName()));
try {
- Path dTmp = files.makeQualified(copy(sCopy, dst_work));
+ Path dTmp = this.userUgi.doAs(new PrivilegedExceptionAction<Path>() {
+ public Path run() throws Exception {
+ return files.makeQualified(copy(sCopy, dst_work));
+ };
+ });
unpack(new File(dTmp.toUri()), new File(dFinal.toUri()));
files.rename(dst_work, dst, Rename.OVERWRITE);
- } catch (IOException e) {
+ } catch (Exception e) {
try { files.delete(dst, true); } catch (IOException ignore) { }
throw e;
} finally {
Modified:
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestContainerLocalizer.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestContainerLocalizer.java?rev=1100165&r1=1100164&r2=1100165&view=diff
==============================================================================
---
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestContainerLocalizer.java
(original)
+++
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestContainerLocalizer.java
Fri May 6 10:45:08 2011
@@ -55,6 +55,7 @@ import org.apache.hadoop.io.DataInputBuf
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
@@ -131,14 +132,18 @@ public class TestContainerLocalizer {
Collections.<LocalResource>emptyList()))
.thenReturn(new MockLocalizerHeartbeatResponse(LocalizerAction.DIE,
null));
- doReturn(new FakeDownload(rsrcA.getResource().getFile(), true))
- .when(localizer).download(isA(LocalDirAllocator.class), eq(rsrcA));
- doReturn(new FakeDownload(rsrcB.getResource().getFile(), true))
- .when(localizer).download(isA(LocalDirAllocator.class), eq(rsrcB));
- doReturn(new FakeDownload(rsrcC.getResource().getFile(), true))
- .when(localizer).download(isA(LocalDirAllocator.class), eq(rsrcC));
- doReturn(new FakeDownload(rsrcD.getResource().getFile(), true))
- .when(localizer).download(isA(LocalDirAllocator.class), eq(rsrcD));
+ doReturn(new FakeDownload(rsrcA.getResource().getFile(), true)).when(
+ localizer).download(isA(LocalDirAllocator.class), eq(rsrcA),
+ isA(UserGroupInformation.class));
+ doReturn(new FakeDownload(rsrcB.getResource().getFile(), true)).when(
+ localizer).download(isA(LocalDirAllocator.class), eq(rsrcB),
+ isA(UserGroupInformation.class));
+ doReturn(new FakeDownload(rsrcC.getResource().getFile(), true)).when(
+ localizer).download(isA(LocalDirAllocator.class), eq(rsrcC),
+ isA(UserGroupInformation.class));
+ doReturn(new FakeDownload(rsrcD.getResource().getFile(), true)).when(
+ localizer).download(isA(LocalDirAllocator.class), eq(rsrcD),
+ isA(UserGroupInformation.class));
doReturn(nmProxy).when(localizer).getProxy(nmAddr);
doNothing().when(localizer).sleep(anyInt());
Modified:
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestFSDownload.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestFSDownload.java?rev=1100165&r1=1100164&r2=1100165&view=diff
==============================================================================
---
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestFSDownload.java
(original)
+++
hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestFSDownload.java
Fri May 6 10:45:08 2011
@@ -34,6 +34,7 @@ import org.apache.hadoop.fs.FSDataOutput
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.LocalDirAllocator;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.factories.RecordFactory;
@@ -104,16 +105,16 @@ public class TestFSDownload {
LocalResource rsrc = createFile(files, new Path(basedir, "" + i),
sizes[i], rand);
FSDownload fsd =
- new FSDownload(files, conf, dirs, rsrc, new Random(sharedSeed));
+ new FSDownload(files, UserGroupInformation.getCurrentUser(), conf,
+ dirs, rsrc, new Random(sharedSeed));
pending.put(rsrc, exec.submit(fsd));
}
try {
for (Map.Entry<LocalResource,Future<Path>> p : pending.entrySet()) {
Path localized = p.getValue().get();
- assertEquals(
- sizes[Integer.valueOf(localized.getName())],
- p.getKey().getSize() - 4096 - 16); // bad DU impl + .crc ; sigh
+ assertEquals(sizes[Integer.valueOf(localized.getName())], p.getKey()
+ .getSize());
}
} catch (ExecutionException e) {
throw new IOException("Failed exec", e);
