Author: omalley
Date: Fri Mar 4 04:23:38 2011
New Revision: 1077521
URL: http://svn.apache.org/viewvc?rev=1077521&view=rev
Log:
commit 9512651992dab6db394fb10f7cf17e887e35f6e0
Author: Devaraj Das <[email protected]>
Date: Wed Jun 30 19:36:25 2010 -0700
MAPREDUCE:1900 from
https://issues.apache.org/jira/secure/attachment/12448230/mapred-fs-close.patch
+++ b/YAHOO-CHANGES.txt
+
+ MAPREDUCE-1900. Makes JobTracker/TaskTracker close filesystems, created
+ on behalf of users, when they are no longer needed. (ddas)
+
Modified:
hadoop/common/branches/branch-0.20-security-patches/src/core/org/apache/hadoop/fs/FileSystem.java
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/JobInProgress.java
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/JobTracker.java
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/TaskTracker.java
hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/fs/TestFileSystem.java
Modified:
hadoop/common/branches/branch-0.20-security-patches/src/core/org/apache/hadoop/fs/FileSystem.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/core/org/apache/hadoop/fs/FileSystem.java?rev=1077521&r1=1077520&r2=1077521&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.20-security-patches/src/core/org/apache/hadoop/fs/FileSystem.java
(original)
+++
hadoop/common/branches/branch-0.20-security-patches/src/core/org/apache/hadoop/fs/FileSystem.java
Fri Mar 4 04:23:38 2011
@@ -250,6 +250,17 @@ public abstract class FileSystem extends
LOG.debug("Done clearing cache");
}
+ /**
+ * Close all cached filesystems for a given UGI. Be sure those filesystems
+ * are not used anymore.
+ * @param ugi
+ * @throws IOException
+ */
+ public static void closeAllForUGI(UserGroupInformation ugi)
+ throws IOException {
+ CACHE.closeAll(ugi);
+ }
+
/** Make sure that a path specifies a FileSystem. */
public Path makeQualified(Path path) {
checkPath(path);
@@ -1368,6 +1379,32 @@ public abstract class FileSystem extends
}
}
+ synchronized void closeAll(UserGroupInformation ugi) throws IOException {
+ List<FileSystem> targetFSList = new ArrayList<FileSystem>();
+ //Make a pass over the list and collect the filesystems to close
+ //we cannot close inline since close() removes the entry from the Map
+ for (Map.Entry<Key, FileSystem> entry : map.entrySet()) {
+ final Key key = entry.getKey();
+ final FileSystem fs = entry.getValue();
+ if (ugi.equals(key.ugi) && fs != null) {
+ targetFSList.add(fs);
+ }
+ }
+ List<IOException> exceptions = new ArrayList<IOException>();
+ //now make a pass over the target list and close each
+ for (FileSystem fs : targetFSList) {
+ try {
+ fs.close();
+ }
+ catch(IOException ioe) {
+ exceptions.add(ioe);
+ }
+ }
+ if (!exceptions.isEmpty()) {
+ throw MultipleIOException.createIOException(exceptions);
+ }
+ }
+
/** FileSystem.Cache.Key */
static class Key {
final String scheme;
Modified:
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/JobInProgress.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/JobInProgress.java?rev=1077521&r1=1077520&r2=1077521&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/JobInProgress.java
(original)
+++
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/JobInProgress.java
Fri Mar 4 04:23:38 2011
@@ -413,7 +413,7 @@ public class JobInProgress {
// register job's tokens for renewal
DelegationTokenRenewal.registerDelegationTokensForRenewal(
- jobInfo.getJobID(), ts, this.conf);
+ jobInfo.getJobID(), ts, jobtracker.getConf());
}
/**
@@ -2980,6 +2980,14 @@ public class JobInProgress {
if(conf.getBoolean(JobContext.JOB_CANCEL_DELEGATION_TOKEN, true)) {
DelegationTokenRenewal.removeDelegationTokenRenewalForJob(jobId);
} // else don't remove it.May be used by spawned tasks
+
+ //close the user's FS
+ try {
+ FileSystem.closeAllForUGI(userUGI);
+ } catch (IOException ie) {
+ LOG.warn("Ignoring exception " + StringUtils.stringifyException(ie) +
+ " while closing FileSystem for " + userUGI);
+ }
}
/**
@@ -3256,7 +3264,7 @@ public class JobInProgress {
TokenCache.setJobToken(token, tokenStorage);
// write TokenStorage out
- tokenStorage.writeTokenStorageFile(keysFile, conf);
+ tokenStorage.writeTokenStorageFile(keysFile, jobtracker.getConf());
LOG.info("jobToken generated and stored with users keys in "
+ keysFile.toUri().getPath());
}
Modified:
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/JobTracker.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/JobTracker.java?rev=1077521&r1=1077520&r2=1077521&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/JobTracker.java
(original)
+++
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/JobTracker.java
Fri Mar 4 04:23:38 2011
@@ -3611,7 +3611,13 @@ public class JobTracker implements MRCon
return fs.getUri().toString();
}
-
+ /**
+ * Returns a handle to the JobTracker's Configuration
+ */
+ public JobConf getConf() {
+ return conf;
+ }
+
public void reportTaskTrackerError(String taskTracker,
String errorClass,
String errorMessage) throws IOException {
Modified:
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/TaskTracker.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/TaskTracker.java?rev=1077521&r1=1077520&r2=1077521&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/TaskTracker.java
(original)
+++
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/mapred/TaskTracker.java
Fri Mar 4 04:23:38 2011
@@ -1842,6 +1842,13 @@ public class TaskTracker
// Remove this job
rjob.tasks.clear();
+ // Close all FileSystems for this job
+ try {
+ FileSystem.closeAllForUGI(rjob.getUGI());
+ } catch (IOException ie) {
+ LOG.warn("Ignoring exception " + StringUtils.stringifyException(ie)
+
+ " while closing FileSystem for " + rjob.getUGI());
+ }
}
}
Modified:
hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/fs/TestFileSystem.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/fs/TestFileSystem.java?rev=1077521&r1=1077520&r2=1077521&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/fs/TestFileSystem.java
(original)
+++
hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/fs/TestFileSystem.java
Fri Mar 4 04:23:38 2011
@@ -669,4 +669,32 @@ public class TestFileSystem extends Test
// file system.
assertSame(fsA, fsA1);
}
+
+ public void testCloseAllForUGI() throws Exception {
+ final Configuration conf = new Configuration();
+ conf.set("fs.cachedfile.impl", conf.get("fs.file.impl"));
+ UserGroupInformation ugiA = UserGroupInformation.createRemoteUser("foo");
+ FileSystem fsA = ugiA.doAs(new PrivilegedExceptionAction<FileSystem>() {
+ public FileSystem run() throws Exception {
+ return FileSystem.get(new URI("cachedfile://a"), conf);
+ }
+ });
+ //Now we should get the cached filesystem
+ FileSystem fsA1 = ugiA.doAs(new PrivilegedExceptionAction<FileSystem>() {
+ public FileSystem run() throws Exception {
+ return FileSystem.get(new URI("cachedfile://a"), conf);
+ }
+ });
+ assertSame(fsA, fsA1);
+
+ FileSystem.closeAllForUGI(ugiA);
+
+ //Now we should get a different (newly created) filesystem
+ fsA1 = ugiA.doAs(new PrivilegedExceptionAction<FileSystem>() {
+ public FileSystem run() throws Exception {
+ return FileSystem.get(new URI("cachedfile://a"), conf);
+ }
+ });
+ assertNotSame(fsA, fsA1);
+ }
}
