Author: tomwhite
Date: Fri May 6 22:47:17 2011
New Revision: 1100409
URL: http://svn.apache.org/viewvc?rev=1100409&view=rev
Log:
MAPREDUCE-2422. Removed unused internal methods from DistributedCache.
Added:
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/ClientDistributedCacheManager.java
(with props)
Modified:
hadoop/mapreduce/trunk/CHANGES.txt
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/filecache/DistributedCache.java
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/JobSubmitter.java
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/DistributedCache.java
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/TrackerDistributedCacheManager.java
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestSetupWorkDir.java
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/filecache/TestTrackerDistributedCacheManager.java
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/filecache/TestURIFragments.java
Modified: hadoop/mapreduce/trunk/CHANGES.txt
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/CHANGES.txt?rev=1100409&r1=1100408&r2=1100409&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/CHANGES.txt (original)
+++ hadoop/mapreduce/trunk/CHANGES.txt Fri May 6 22:47:17 2011
@@ -79,6 +79,9 @@ Trunk (unreleased changes)
Job Configuration. (Harsh J Chouraria via todd)
MAPREDUCE-2475. Disable IPV6 for junit tests. (suresh srinivas via mahadev)
+
+ MAPREDUCE-2422. Removed unused internal methods from DistributedCache.
+ (tomwhite)
OPTIMIZATIONS
Modified:
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/filecache/DistributedCache.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/filecache/DistributedCache.java?rev=1100409&r1=1100408&r2=1100409&view=diff
==============================================================================
---
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/filecache/DistributedCache.java
(original)
+++
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/filecache/DistributedCache.java
Fri May 6 22:47:17 2011
@@ -24,8 +24,6 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.filecache.TaskDistributedCacheManager;
-import org.apache.hadoop.mapreduce.filecache.TrackerDistributedCacheManager;
/**
* Distribute application-specific large, read-only files efficiently.
@@ -116,9 +114,7 @@ import org.apache.hadoop.mapreduce.filec
* (specifically those mentioned in the example above, as well
* as {@link DistributedCache#addArchiveToClassPath(Path, Configuration)}),
* as well as methods intended for use by the MapReduce framework
- * (e.g., {@link org.apache.hadoop.mapred.JobClient}). For implementation
- * details, see {@link TrackerDistributedCacheManager} and
- * {@link TaskDistributedCacheManager}.
+ * (e.g., {@link org.apache.hadoop.mapred.JobClient}).
*
* @see org.apache.hadoop.mapred.JobConf
* @see org.apache.hadoop.mapred.JobClient
Modified:
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/JobSubmitter.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/JobSubmitter.java?rev=1100409&r1=1100408&r2=1100409&view=diff
==============================================================================
---
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/JobSubmitter.java
(original)
+++
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/JobSubmitter.java
Fri May 6 22:47:17 2011
@@ -43,8 +43,9 @@ import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.QueueACL;
import static org.apache.hadoop.mapred.QueueManager.toFullPropertyName;
+
+import org.apache.hadoop.mapreduce.filecache.ClientDistributedCacheManager;
import org.apache.hadoop.mapreduce.filecache.DistributedCache;
-import org.apache.hadoop.mapreduce.filecache.TrackerDistributedCacheManager;
import org.apache.hadoop.mapreduce.protocol.ClientProtocol;
import org.apache.hadoop.mapreduce.security.TokenCache;
import org.apache.hadoop.mapreduce.split.JobSplitWriter;
@@ -242,11 +243,11 @@ class JobSubmitter {
}
// set the timestamps of the archives and files
- TrackerDistributedCacheManager.determineTimestamps(conf);
+ ClientDistributedCacheManager.determineTimestamps(conf);
// set the public/private visibility of the archives and files
- TrackerDistributedCacheManager.determineCacheVisibilities(conf);
+ ClientDistributedCacheManager.determineCacheVisibilities(conf);
// get DelegationToken for each cached file
- TrackerDistributedCacheManager.getDelegationTokens(conf, job
+ ClientDistributedCacheManager.getDelegationTokens(conf, job
.getCredentials());
}
Added:
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/ClientDistributedCacheManager.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/ClientDistributedCacheManager.java?rev=1100409&view=auto
==============================================================================
---
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/ClientDistributedCacheManager.java
(added)
+++
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/ClientDistributedCacheManager.java
Fri May 6 22:47:17 2011
@@ -0,0 +1,266 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.mapreduce.filecache;
+
+import java.io.IOException;
+import java.net.URI;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.mapreduce.MRJobConfig;
+import org.apache.hadoop.mapreduce.security.TokenCache;
+import org.apache.hadoop.security.Credentials;
+
+/**
+ * Manages internal configuration of the cache by the client for job
submission.
+ */
[email protected]
+public class ClientDistributedCacheManager {
+
+ /**
+ * Determines timestamps of files to be cached, and stores those
+ * in the configuration. This is intended to be used internally by JobClient
+ * after all cache files have been added.
+ *
+ * This is an internal method!
+ *
+ * @param job Configuration of a job.
+ * @throws IOException
+ */
+ public static void determineTimestamps(Configuration job) throws IOException
{
+ URI[] tarchives = DistributedCache.getCacheArchives(job);
+ if (tarchives != null) {
+ FileStatus status = getFileStatus(job, tarchives[0]);
+ StringBuilder archiveFileSizes =
+ new StringBuilder(String.valueOf(status.getLen()));
+ StringBuilder archiveTimestamps =
+ new StringBuilder(String.valueOf(status.getModificationTime()));
+ for (int i = 1; i < tarchives.length; i++) {
+ status = getFileStatus(job, tarchives[i]);
+ archiveFileSizes.append(",");
+ archiveFileSizes.append(String.valueOf(status.getLen()));
+ archiveTimestamps.append(",");
+ archiveTimestamps.append(String.valueOf(status.getModificationTime()));
+ }
+ job.set(MRJobConfig.CACHE_ARCHIVES_SIZES, archiveFileSizes.toString());
+ setArchiveTimestamps(job, archiveTimestamps.toString());
+ }
+
+ URI[] tfiles = DistributedCache.getCacheFiles(job);
+ if (tfiles != null) {
+ FileStatus status = getFileStatus(job, tfiles[0]);
+ StringBuilder fileSizes =
+ new StringBuilder(String.valueOf(status.getLen()));
+ StringBuilder fileTimestamps = new StringBuilder(String.valueOf(
+ status.getModificationTime()));
+ for (int i = 1; i < tfiles.length; i++) {
+ status = getFileStatus(job, tfiles[i]);
+ fileSizes.append(",");
+ fileSizes.append(String.valueOf(status.getLen()));
+ fileTimestamps.append(",");
+ fileTimestamps.append(String.valueOf(status.getModificationTime()));
+ }
+ job.set(MRJobConfig.CACHE_FILES_SIZES, fileSizes.toString());
+ setFileTimestamps(job, fileTimestamps.toString());
+ }
+ }
+
+ /**
+ * For each archive or cache file - get the corresponding delegation token
+ * @param job
+ * @param credentials
+ * @throws IOException
+ */
+ public static void getDelegationTokens(Configuration job,
+ Credentials credentials) throws IOException {
+ URI[] tarchives = DistributedCache.getCacheArchives(job);
+ URI[] tfiles = DistributedCache.getCacheFiles(job);
+
+ int size = (tarchives!=null? tarchives.length : 0) + (tfiles!=null ?
tfiles.length :0);
+ Path[] ps = new Path[size];
+
+ int i = 0;
+ if (tarchives != null) {
+ for (i=0; i < tarchives.length; i++) {
+ ps[i] = new Path(tarchives[i].toString());
+ }
+ }
+
+ if (tfiles != null) {
+ for(int j=0; j< tfiles.length; j++) {
+ ps[i+j] = new Path(tfiles[j].toString());
+ }
+ }
+
+ TokenCache.obtainTokensForNamenodes(credentials, ps, job);
+ }
+
+ /**
+ * Determines the visibilities of the distributed cache files and
+ * archives. The visibility of a cache path is "public" if the leaf component
+ * has READ permissions for others, and the parent subdirs have
+ * EXECUTE permissions for others
+ * @param job
+ * @throws IOException
+ */
+ public static void determineCacheVisibilities(Configuration job)
+ throws IOException {
+ URI[] tarchives = DistributedCache.getCacheArchives(job);
+ if (tarchives != null) {
+ StringBuilder archiveVisibilities =
+ new StringBuilder(String.valueOf(isPublic(job, tarchives[0])));
+ for (int i = 1; i < tarchives.length; i++) {
+ archiveVisibilities.append(",");
+ archiveVisibilities.append(String.valueOf(isPublic(job,
tarchives[i])));
+ }
+ setArchiveVisibilities(job, archiveVisibilities.toString());
+ }
+ URI[] tfiles = DistributedCache.getCacheFiles(job);
+ if (tfiles != null) {
+ StringBuilder fileVisibilities =
+ new StringBuilder(String.valueOf(isPublic(job, tfiles[0])));
+ for (int i = 1; i < tfiles.length; i++) {
+ fileVisibilities.append(",");
+ fileVisibilities.append(String.valueOf(isPublic(job, tfiles[i])));
+ }
+ setFileVisibilities(job, fileVisibilities.toString());
+ }
+ }
+
+ /**
+ * This is to check the public/private visibility of the archives to be
+ * localized.
+ *
+ * @param conf Configuration which stores the timestamp's
+ * @param booleans comma separated list of booleans (true - public)
+ * The order should be the same as the order in which the archives are added.
+ */
+ static void setArchiveVisibilities(Configuration conf, String booleans) {
+ conf.set(MRJobConfig.CACHE_ARCHIVES_VISIBILITIES, booleans);
+ }
+
+ /**
+ * This is to check the public/private visibility of the files to be
localized
+ *
+ * @param conf Configuration which stores the timestamp's
+ * @param booleans comma separated list of booleans (true - public)
+ * The order should be the same as the order in which the files are added.
+ */
+ static void setFileVisibilities(Configuration conf, String booleans) {
+ conf.set(MRJobConfig.CACHE_FILE_VISIBILITIES, booleans);
+ }
+
+ /**
+ * This is to check the timestamp of the archives to be localized.
+ *
+ * @param conf Configuration which stores the timestamp's
+ * @param timestamps comma separated list of timestamps of archives.
+ * The order should be the same as the order in which the archives are added.
+ */
+ static void setArchiveTimestamps(Configuration conf, String timestamps) {
+ conf.set(MRJobConfig.CACHE_ARCHIVES_TIMESTAMPS, timestamps);
+ }
+
+ /**
+ * This is to check the timestamp of the files to be localized.
+ *
+ * @param conf Configuration which stores the timestamp's
+ * @param timestamps comma separated list of timestamps of files.
+ * The order should be the same as the order in which the files are added.
+ */
+ static void setFileTimestamps(Configuration conf, String timestamps) {
+ conf.set(MRJobConfig.CACHE_FILE_TIMESTAMPS, timestamps);
+ }
+
+ /**
+ * Returns {@link FileStatus} of a given cache file on hdfs.
+ *
+ * @param conf configuration
+ * @param cache cache file
+ * @return {@link FileStatus} of a given cache file on hdfs
+ * @throws IOException
+ */
+ static FileStatus getFileStatus(Configuration conf, URI cache)
+ throws IOException {
+ FileSystem fileSystem = FileSystem.get(cache, conf);
+ Path filePath = new Path(cache.getPath());
+
+ return fileSystem.getFileStatus(filePath);
+ }
+
+ /**
+ * Returns a boolean to denote whether a cache file is visible to all(public)
+ * or not
+ * @param conf
+ * @param uri
+ * @return true if the path in the uri is visible to all, false otherwise
+ * @throws IOException
+ */
+ static boolean isPublic(Configuration conf, URI uri) throws IOException {
+ FileSystem fs = FileSystem.get(uri, conf);
+ Path current = new Path(uri.getPath());
+ //the leaf level file should be readable by others
+ if (!checkPermissionOfOther(fs, current, FsAction.READ)) {
+ return false;
+ }
+ return ancestorsHaveExecutePermissions(fs, current.getParent());
+ }
+
+ /**
+ * Returns true if all ancestors of the specified path have the 'execute'
+ * permission set for all users (i.e. that other users can traverse
+ * the directory heirarchy to the given path)
+ */
+ static boolean ancestorsHaveExecutePermissions(FileSystem fs, Path path)
+ throws IOException {
+ Path current = path;
+ while (current != null) {
+ //the subdirs in the path should have execute permissions for others
+ if (!checkPermissionOfOther(fs, current, FsAction.EXECUTE)) {
+ return false;
+ }
+ current = current.getParent();
+ }
+ return true;
+ }
+
+ /**
+ * Checks for a given path whether the Other permissions on it
+ * imply the permission in the passed FsAction
+ * @param fs
+ * @param path
+ * @param action
+ * @return true if the path in the uri is visible to all, false otherwise
+ * @throws IOException
+ */
+ private static boolean checkPermissionOfOther(FileSystem fs, Path path,
+ FsAction action) throws IOException {
+ FileStatus status = fs.getFileStatus(path);
+ FsPermission perms = status.getPermission();
+ FsAction otherAction = perms.getOtherAction();
+ if (otherAction.implies(action)) {
+ return true;
+ }
+ return false;
+ }
+}
Propchange:
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/ClientDistributedCacheManager.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified:
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/DistributedCache.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/DistributedCache.java?rev=1100409&r1=1100408&r2=1100409&view=diff
==============================================================================
---
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/DistributedCache.java
(original)
+++
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/DistributedCache.java
Fri May 6 22:47:17 2011
@@ -26,7 +26,6 @@ import org.apache.hadoop.conf.*;
import org.apache.hadoop.util.*;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.mapred.DefaultTaskController;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.MRJobConfig;
@@ -122,218 +121,14 @@ import java.net.URI;
* (specifically those mentioned in the example above, as well
* as {@link DistributedCache#addArchiveToClassPath(Path, Configuration)}),
* as well as methods intended for use by the MapReduce framework
- * (e.g., {@link org.apache.hadoop.mapred.JobClient}). For implementation
- * details, see {@link TrackerDistributedCacheManager} and
- * {@link TaskDistributedCacheManager}.
+ * (e.g., {@link org.apache.hadoop.mapred.JobClient}).
*
- * @see TrackerDistributedCacheManager
- * @see TaskDistributedCacheManager
* @see org.apache.hadoop.mapred.JobConf
* @see org.apache.hadoop.mapred.JobClient
*/
@Deprecated
@InterfaceAudience.Private
public class DistributedCache {
- /**
- * Get the locally cached file or archive; it could either be
- * previously cached (and valid) or copy it from the {@link FileSystem} now.
- *
- * @param cache the cache to be localized, this should be specified as
- * new URI(scheme://scheme-specific-part/absolute_path_to_file#LINKNAME).
- * @param conf The Confguration file which contains the filesystem
- * @param baseDir The base cache Dir where you wnat to localize the
files/archives
- * @param fileStatus The file status on the dfs.
- * @param isArchive if the cache is an archive or a file. In case it is an
- * archive with a .zip or .jar or .tar or .tgz or .tar.gz extension it will
- * be unzipped/unjarred/untarred automatically
- * and the directory where the archive is unzipped/unjarred/untarred is
- * returned as the Path.
- * In case of a file, the path to the file is returned
- * @param confFileStamp this is the hdfs file modification timestamp to
verify that the
- * file to be cached hasn't changed since the job started
- * @param currentWorkDir this is the directory where you would want to
create symlinks
- * for the locally cached files/archives
- * @return the path to directory where the archives are unjarred in case of
archives,
- * the path to the file where the file is copied locally
- * @throws IOException
- * @deprecated Internal to MapReduce framework.
- * Use TrackerDistributedCacheManager instead.
- */
- @Deprecated
- public static Path getLocalCache(URI cache, Configuration conf,
- Path baseDir, FileStatus fileStatus,
- boolean isArchive, long confFileStamp,
- Path currentWorkDir)
- throws IOException {
- return getLocalCache(cache, conf, baseDir, fileStatus, isArchive,
- confFileStamp, currentWorkDir, true);
- }
-
- /**
- * Get the locally cached file or archive; it could either be
- * previously cached (and valid) or copy it from the {@link FileSystem} now.
- *
- * @param cache the cache to be localized, this should be specified as
- * new URI(scheme://scheme-specific-part/absolute_path_to_file#LINKNAME).
- * @param conf The Confguration file which contains the filesystem
- * @param baseDir The base cache Dir where you wnat to localize the
files/archives
- * @param fileStatus The file status on the dfs.
- * @param isArchive if the cache is an archive or a file. In case it is an
- * archive with a .zip or .jar or .tar or .tgz or .tar.gz extension it will
- * be unzipped/unjarred/untarred automatically
- * and the directory where the archive is unzipped/unjarred/untarred is
- * returned as the Path.
- * In case of a file, the path to the file is returned
- * @param confFileStamp this is the hdfs file modification timestamp to
verify that the
- * file to be cached hasn't changed since the job started
- * @param currentWorkDir this is the directory where you would want to
create symlinks
- * for the locally cached files/archives
- * @param honorSymLinkConf if this is false, then the symlinks are not
- * created even if conf says so (this is required for an optimization in task
- * launches
- * @return the path to directory where the archives are unjarred in case of
archives,
- * the path to the file where the file is copied locally
- * @throws IOException
- * @deprecated Internal to MapReduce framework.
- * Use TrackerDistributedCacheManager instead.
- */
- @Deprecated
- public static Path getLocalCache(URI cache, Configuration conf,
- Path baseDir, FileStatus fileStatus,
- boolean isArchive, long confFileStamp,
- Path currentWorkDir, boolean honorSymLinkConf) throws IOException {
-
- return new TrackerDistributedCacheManager(conf, new
DefaultTaskController())
- .getLocalCache(cache, conf, baseDir.toString(), fileStatus, isArchive,
- confFileStamp, currentWorkDir, honorSymLinkConf, false);
- }
-
- /**
- * Get the locally cached file or archive; it could either be
- * previously cached (and valid) or copy it from the {@link FileSystem} now.
- *
- * @param cache the cache to be localized, this should be specified as
- * new URI(scheme://scheme-specific-part/absolute_path_to_file#LINKNAME).
- * @param conf The Confguration file which contains the filesystem
- * @param baseDir The base cache Dir where you wnat to localize the
files/archives
- * @param isArchive if the cache is an archive or a file. In case it is an
- * archive with a .zip or .jar or .tar or .tgz or .tar.gz extension it will
- * be unzipped/unjarred/untarred automatically
- * and the directory where the archive is unzipped/unjarred/untarred
- * is returned as the Path.
- * In case of a file, the path to the file is returned
- * @param confFileStamp this is the hdfs file modification timestamp to
verify that the
- * file to be cached hasn't changed since the job started
- * @param currentWorkDir this is the directory where you would want to
create symlinks
- * for the locally cached files/archives
- * @return the path to directory where the archives are unjarred in case of
archives,
- * the path to the file where the file is copied locally
- * @throws IOException
- * @deprecated Internal to MapReduce framework.
- * Use TrackerDistributedCacheManager instead.
- */
- @Deprecated
- public static Path getLocalCache(URI cache, Configuration conf,
- Path baseDir, boolean isArchive,
- long confFileStamp, Path currentWorkDir)
- throws IOException {
- return getLocalCache(cache, conf,
- baseDir, null, isArchive,
- confFileStamp, currentWorkDir);
- }
-
- /**
- * This is the opposite of getlocalcache. When you are done with
- * using the cache, you need to release the cache
- * @param cache The cache URI to be released
- * @param conf configuration which contains the filesystem the cache
- * is contained in.
- * @throws IOException
- * @deprecated Internal to MapReduce framework.
- * Use TrackerDistributedCacheManager instead.
- */
- @Deprecated
- public static void releaseCache(URI cache, Configuration conf)
- throws IOException {
- // find the timestamp of the uri
- URI[] archives = DistributedCache.getCacheArchives(conf);
- URI[] files = DistributedCache.getCacheFiles(conf);
- String[] archivesTimestamps =
- DistributedCache.getArchiveTimestamps(conf);
- String[] filesTimestamps =
- DistributedCache.getFileTimestamps(conf);
- String timestamp = null;
- if (archives != null) {
- for (int i = 0; i < archives.length; i++) {
- if (archives[i].equals(cache)) {
- timestamp = archivesTimestamps[i];
- break;
- }
- }
- }
- if (timestamp == null && files != null) {
- for (int i = 0; i < files.length; i++) {
- if (files[i].equals(cache)) {
- timestamp = filesTimestamps[i];
- break;
- }
- }
- }
- if (timestamp == null) {
- throw new IOException("TimeStamp of the uri couldnot be found");
- }
- new TrackerDistributedCacheManager(conf, new DefaultTaskController())
- .releaseCache(cache, conf, Long.parseLong(timestamp),
- TrackerDistributedCacheManager.getLocalizedCacheOwner(false));
- }
-
- /**
- * Returns the relative path of the dir this cache will be localized in
- * relative path that this cache will be localized in. For
- * hdfs://hostname:port/absolute_path -- the relative path is
- * hostname/absolute path -- if it is just /absolute_path -- then the
- * relative path is hostname of DFS this mapred cluster is running
- * on/absolute_path
- * @deprecated Internal to MapReduce framework. Use DistributedCacheManager
- * instead.
- */
- @Deprecated
- public static String makeRelative(URI cache, Configuration conf)
- throws IOException {
- return new TrackerDistributedCacheManager(conf, new
DefaultTaskController())
- .makeRelative(cache, conf);
- }
-
- /**
- * Returns mtime of a given cache file on hdfs.
- *
- * @param conf configuration
- * @param cache cache file
- * @return mtime of a given cache file on hdfs
- * @throws IOException
- * @deprecated Internal to MapReduce framework.
- * Use {@link TrackerDistributedCacheManager} instead.
- */
- @Deprecated
- public static long getTimestamp(Configuration conf, URI cache)
- throws IOException {
- return TrackerDistributedCacheManager.getTimestamp(conf, cache);
- }
-
- /**
- * This method create symlinks for all files in a given dir in another
directory
- * @param conf the configuration
- * @param jobCacheDir the target directory for creating symlinks
- * @param workDir the directory in which the symlinks are created
- * @throws IOException
- * @deprecated Internal to MapReduce framework. Use DistributedCacheManager
- * instead.
- */
- @Deprecated
- public static void createAllSymlink(Configuration conf, File jobCacheDir,
File workDir)
- throws IOException{
- TrackerDistributedCacheManager.createAllSymlink(conf, jobCacheDir,
workDir);
- }
/**
* Set the configuration with the given set of archives. Intended
@@ -444,64 +239,6 @@ public class DistributedCache {
}
/**
- * This is to check the timestamp of the archives to be localized.
- * Used by internal MapReduce code.
- * @param conf Configuration which stores the timestamp's
- * @param timestamps comma separated list of timestamps of archives.
- * The order should be the same as the order in which the archives are added.
- * @deprecated Use
- * {@link TrackerDistributedCacheManager#setArchiveTimestamps(Configuration,
String)}
- * instead
- */
- @Deprecated
- public static void setArchiveTimestamps(Configuration conf, String
timestamps) {
- TrackerDistributedCacheManager.setArchiveTimestamps(conf, timestamps);
- }
-
- /**
- * This is to check the timestamp of the files to be localized.
- * Used by internal MapReduce code.
- * @param conf Configuration which stores the timestamp's
- * @param timestamps comma separated list of timestamps of files.
- * The order should be the same as the order in which the files are added.
- * @deprecated Use
- * {@link TrackerDistributedCacheManager#setFileTimestamps(Configuration,
String)}
- * instead
- */
- @Deprecated
- public static void setFileTimestamps(Configuration conf, String timestamps) {
- TrackerDistributedCacheManager.setFileTimestamps(conf, timestamps);
- }
-
- /**
- * Set the conf to contain the location for localized archives. Used
- * by internal DistributedCache code.
- * @param conf The conf to modify to contain the localized caches
- * @param str a comma separated list of local archives
- * @deprecated Use
- * {@link TrackerDistributedCacheManager#setLocalArchives(Configuration,
String)}
- * instead
- */
- @Deprecated
- public static void setLocalArchives(Configuration conf, String str) {
- TrackerDistributedCacheManager.setLocalArchives(conf, str);
- }
-
- /**
- * Set the conf to contain the location for localized files. Used
- * by internal DistributedCache code.
- * @param conf The conf to modify to contain the localized caches
- * @param str a comma separated list of local files
- * @deprecated Use
- * {@link TrackerDistributedCacheManager#setLocalFiles(Configuration,
String)}
- * instead
- */
- @Deprecated
- public static void setLocalFiles(Configuration conf, String str) {
- TrackerDistributedCacheManager.setLocalFiles(conf, str);
- }
-
- /**
* Add a archives to be localized to the conf. Intended to
* be used by user code.
* @param uri The uri of the cache to be localized
@@ -647,24 +384,45 @@ public class DistributedCache {
* the various archives and files. May be used by user code.
* @param uriFiles The uri array of urifiles
* @param uriArchives the uri array of uri archives
- * @deprecated Use
- * {@link TrackerDistributedCacheManager#checkURIs(URI[], URI[])} instead
*/
- @Deprecated
- public static boolean checkURIs(URI[] uriFiles, URI[] uriArchives){
- return TrackerDistributedCacheManager.checkURIs(uriFiles, uriArchives);
+ public static boolean checkURIs(URI[] uriFiles, URI[] uriArchives) {
+ if ((uriFiles == null) && (uriArchives == null)) {
+ return true;
+ }
+ // check if fragment is null for any uri
+ // also check if there are any conflicts in fragment names
+ Set<String> fragments = new HashSet<String>();
+
+ // iterate over file uris
+ if (uriFiles != null) {
+ for (int i = 0; i < uriFiles.length; i++) {
+ String fragment = uriFiles[i].getFragment();
+ if (fragment == null) {
+ return false;
+ }
+ String lowerCaseFragment = fragment.toLowerCase();
+ if (fragments.contains(lowerCaseFragment)) {
+ return false;
+ }
+ fragments.add(lowerCaseFragment);
+ }
+ }
+
+ // iterate over archive uris
+ if (uriArchives != null) {
+ for (int i = 0; i < uriArchives.length; i++) {
+ String fragment = uriArchives[i].getFragment();
+ if (fragment == null) {
+ return false;
+ }
+ String lowerCaseFragment = fragment.toLowerCase();
+ if (fragments.contains(lowerCaseFragment)) {
+ return false;
+ }
+ fragments.add(lowerCaseFragment);
+ }
+ }
+ return true;
}
- /**
- * Clear the entire contents of the cache and delete the backing files. This
- * should only be used when the server is reinitializing, because the users
- * are going to lose their files.
- * @deprecated Internal to MapReduce framework.
- * Use TrackerDistributedCacheManager instead.
- */
- @Deprecated
- public static void purgeCache(Configuration conf) throws IOException {
- new TrackerDistributedCacheManager(conf, new DefaultTaskController())
- .purgeCache();
- }
}
Modified:
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/TrackerDistributedCacheManager.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/TrackerDistributedCacheManager.java?rev=1100409&r1=1100408&r2=1100409&view=diff
==============================================================================
---
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/TrackerDistributedCacheManager.java
(original)
+++
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/filecache/TrackerDistributedCacheManager.java
Fri May 6 22:47:17 2011
@@ -35,7 +35,6 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.mapred.TaskController;
import org.apache.hadoop.mapred.TaskController.DistributedCacheFileContext;
import org.apache.hadoop.mapreduce.MRJobConfig;
-import org.apache.hadoop.mapreduce.security.TokenCache;
import org.apache.hadoop.mapreduce.server.tasktracker.TTConfig;
import org.apache.hadoop.mapreduce.util.MRAsyncDiskService;
import org.apache.hadoop.fs.FileStatus;
@@ -44,9 +43,6 @@ import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.LocalDirAllocator;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.permission.FsAction;
-import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.RunJar;
import org.apache.hadoop.classification.InterfaceAudience;
@@ -325,10 +321,7 @@ public class TrackerDistributedCacheMana
*/
static FileStatus getFileStatus(Configuration conf, URI cache)
throws IOException {
- FileSystem fileSystem = FileSystem.get(cache, conf);
- Path filePath = new Path(cache.getPath());
-
- return fileSystem.getFileStatus(filePath);
+ return ClientDistributedCacheManager.getFileStatus(conf, cache);
}
/**
@@ -344,62 +337,6 @@ public class TrackerDistributedCacheMana
return getFileStatus(conf, cache).getModificationTime();
}
- /**
- * Returns a boolean to denote whether a cache file is visible to all(public)
- * or not
- * @param conf
- * @param uri
- * @return true if the path in the uri is visible to all, false otherwise
- * @throws IOException
- */
- static boolean isPublic(Configuration conf, URI uri) throws IOException {
- FileSystem fs = FileSystem.get(uri, conf);
- Path current = new Path(uri.getPath());
- //the leaf level file should be readable by others
- if (!checkPermissionOfOther(fs, current, FsAction.READ)) {
- return false;
- }
- return ancestorsHaveExecutePermissions(fs, current.getParent());
- }
-
- /**
- * Returns true if all ancestors of the specified path have the 'execute'
- * permission set for all users (i.e. that other users can traverse
- * the directory heirarchy to the given path)
- */
- static boolean ancestorsHaveExecutePermissions(FileSystem fs, Path path)
- throws IOException {
- Path current = path;
- while (current != null) {
- //the subdirs in the path should have execute permissions for others
- if (!checkPermissionOfOther(fs, current, FsAction.EXECUTE)) {
- return false;
- }
- current = current.getParent();
- }
- return true;
- }
-
- /**
- * Checks for a given path whether the Other permissions on it
- * imply the permission in the passed FsAction
- * @param fs
- * @param path
- * @param action
- * @return true if the path in the uri is visible to all, false otherwise
- * @throws IOException
- */
- private static boolean checkPermissionOfOther(FileSystem fs, Path path,
- FsAction action) throws IOException {
- FileStatus status = fs.getFileStatus(path);
- FsPermission perms = status.getPermission();
- FsAction otherAction = perms.getOtherAction();
- if (otherAction.implies(action)) {
- return true;
- }
- return false;
- }
-
private Path checkCacheStatusValidity(Configuration conf,
URI cache, long confFileStamp,
CacheStatus cacheStatus,
@@ -676,116 +613,6 @@ public class TrackerDistributedCacheMana
Configuration taskConf) throws IOException {
return new TaskDistributedCacheManager(this, taskConf);
}
-
- /**
- * Determines timestamps of files to be cached, and stores those
- * in the configuration. This is intended to be used internally by JobClient
- * after all cache files have been added.
- *
- * This is an internal method!
- *
- * @param job Configuration of a job.
- * @throws IOException
- */
- public static void determineTimestamps(Configuration job) throws IOException
{
- URI[] tarchives = DistributedCache.getCacheArchives(job);
- if (tarchives != null) {
- FileStatus status = getFileStatus(job, tarchives[0]);
- StringBuilder archiveFileSizes =
- new StringBuilder(String.valueOf(status.getLen()));
- StringBuilder archiveTimestamps =
- new StringBuilder(String.valueOf(status.getModificationTime()));
- for (int i = 1; i < tarchives.length; i++) {
- status = getFileStatus(job, tarchives[i]);
- archiveFileSizes.append(",");
- archiveFileSizes.append(String.valueOf(status.getLen()));
- archiveTimestamps.append(",");
- archiveTimestamps.append(String.valueOf(status.getModificationTime()));
- }
- job.set(MRJobConfig.CACHE_ARCHIVES_SIZES, archiveFileSizes.toString());
- setArchiveTimestamps(job, archiveTimestamps.toString());
- }
-
- URI[] tfiles = DistributedCache.getCacheFiles(job);
- if (tfiles != null) {
- FileStatus status = getFileStatus(job, tfiles[0]);
- StringBuilder fileSizes =
- new StringBuilder(String.valueOf(status.getLen()));
- StringBuilder fileTimestamps = new StringBuilder(String.valueOf(
- status.getModificationTime()));
- for (int i = 1; i < tfiles.length; i++) {
- status = getFileStatus(job, tfiles[i]);
- fileSizes.append(",");
- fileSizes.append(String.valueOf(status.getLen()));
- fileTimestamps.append(",");
- fileTimestamps.append(String.valueOf(status.getModificationTime()));
- }
- job.set(MRJobConfig.CACHE_FILES_SIZES, fileSizes.toString());
- setFileTimestamps(job, fileTimestamps.toString());
- }
- }
-
- /**
- * For each archive or cache file - get the corresponding delegation token
- * @param job
- * @param credentials
- * @throws IOException
- */
- public static void getDelegationTokens(Configuration job,
- Credentials credentials) throws IOException {
- URI[] tarchives = DistributedCache.getCacheArchives(job);
- URI[] tfiles = DistributedCache.getCacheFiles(job);
-
- int size = (tarchives!=null? tarchives.length : 0) + (tfiles!=null ?
tfiles.length :0);
- Path[] ps = new Path[size];
-
- int i = 0;
- if (tarchives != null) {
- for (i=0; i < tarchives.length; i++) {
- ps[i] = new Path(tarchives[i].toString());
- }
- }
-
- if (tfiles != null) {
- for(int j=0; j< tfiles.length; j++) {
- ps[i+j] = new Path(tfiles[j].toString());
- }
- }
-
- TokenCache.obtainTokensForNamenodes(credentials, ps, job);
- }
-
- /**
- * Determines the visibilities of the distributed cache files and
- * archives. The visibility of a cache path is "public" if the leaf component
- * has READ permissions for others, and the parent subdirs have
- * EXECUTE permissions for others
- * @param job
- * @throws IOException
- */
- public static void determineCacheVisibilities(Configuration job)
- throws IOException {
- URI[] tarchives = DistributedCache.getCacheArchives(job);
- if (tarchives != null) {
- StringBuilder archiveVisibilities =
- new StringBuilder(String.valueOf(isPublic(job, tarchives[0])));
- for (int i = 1; i < tarchives.length; i++) {
- archiveVisibilities.append(",");
- archiveVisibilities.append(String.valueOf(isPublic(job,
tarchives[i])));
- }
- setArchiveVisibilities(job, archiveVisibilities.toString());
- }
- URI[] tfiles = DistributedCache.getCacheFiles(job);
- if (tfiles != null) {
- StringBuilder fileVisibilities =
- new StringBuilder(String.valueOf(isPublic(job, tfiles[0])));
- for (int i = 1; i < tfiles.length; i++) {
- fileVisibilities.append(",");
- fileVisibilities.append(String.valueOf(isPublic(job, tfiles[i])));
- }
- setFileVisibilities(job, fileVisibilities.toString());
- }
- }
/**
* Get the booleans on whether the files are public or not. Used by
@@ -809,99 +636,6 @@ public class TrackerDistributedCacheMana
}
/**
- * This method checks if there is a conflict in the fragment names
- * of the uris. Also makes sure that each uri has a fragment. It
- * is only to be called if you want to create symlinks for
- * the various archives and files. May be used by user code.
- * @param uriFiles The uri array of urifiles
- * @param uriArchives the uri array of uri archives
- */
- public static boolean checkURIs(URI[] uriFiles, URI[] uriArchives) {
- if ((uriFiles == null) && (uriArchives == null)) {
- return true;
- }
- // check if fragment is null for any uri
- // also check if there are any conflicts in fragment names
- Set<String> fragments = new HashSet<String>();
-
- // iterate over file uris
- if (uriFiles != null) {
- for (int i = 0; i < uriFiles.length; i++) {
- String fragment = uriFiles[i].getFragment();
- if (fragment == null) {
- return false;
- }
- String lowerCaseFragment = fragment.toLowerCase();
- if (fragments.contains(lowerCaseFragment)) {
- return false;
- }
- fragments.add(lowerCaseFragment);
- }
- }
-
- // iterate over archive uris
- if (uriArchives != null) {
- for (int i = 0; i < uriArchives.length; i++) {
- String fragment = uriArchives[i].getFragment();
- if (fragment == null) {
- return false;
- }
- String lowerCaseFragment = fragment.toLowerCase();
- if (fragments.contains(lowerCaseFragment)) {
- return false;
- }
- fragments.add(lowerCaseFragment);
- }
- }
- return true;
- }
-
- /**
- * This is to check the public/private visibility of the archives to be
- * localized.
- *
- * @param conf Configuration which stores the timestamp's
- * @param booleans comma separated list of booleans (true - public)
- * The order should be the same as the order in which the archives are added.
- */
- static void setArchiveVisibilities(Configuration conf, String booleans) {
- conf.set(MRJobConfig.CACHE_ARCHIVES_VISIBILITIES, booleans);
- }
-
- /**
- * This is to check the public/private visibility of the files to be
localized
- *
- * @param conf Configuration which stores the timestamp's
- * @param booleans comma separated list of booleans (true - public)
- * The order should be the same as the order in which the files are added.
- */
- static void setFileVisibilities(Configuration conf, String booleans) {
- conf.set(MRJobConfig.CACHE_FILE_VISIBILITIES, booleans);
- }
-
- /**
- * This is to check the timestamp of the archives to be localized.
- *
- * @param conf Configuration which stores the timestamp's
- * @param timestamps comma separated list of timestamps of archives.
- * The order should be the same as the order in which the archives are added.
- */
- static void setArchiveTimestamps(Configuration conf, String timestamps) {
- conf.set(MRJobConfig.CACHE_ARCHIVES_TIMESTAMPS, timestamps);
- }
-
- /**
- * This is to check the timestamp of the files to be localized.
- *
- * @param conf Configuration which stores the timestamp's
- * @param timestamps comma separated list of timestamps of files.
- * The order should be the same as the order in which the files are added.
- */
- static void setFileTimestamps(Configuration conf, String timestamps) {
- conf.set(MRJobConfig.CACHE_FILE_TIMESTAMPS, timestamps);
- }
-
- /**
* Set the conf to contain the location for localized archives.
*
* @param conf The conf to modify to contain the localized caches
@@ -917,7 +651,7 @@ public class TrackerDistributedCacheMana
* @param conf The conf to modify to contain the localized caches
* @param str a comma separated list of local files
*/
- static void setLocalFiles(Configuration conf, String str) {
+ public static void setLocalFiles(Configuration conf, String str) {
conf.set(MRJobConfig.CACHE_LOCALFILES, str);
}
Modified:
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestSetupWorkDir.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestSetupWorkDir.java?rev=1100409&r1=1100408&r2=1100409&view=diff
==============================================================================
---
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestSetupWorkDir.java
(original)
+++
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestSetupWorkDir.java
Fri May 6 22:47:17 2011
@@ -31,6 +31,7 @@ import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.filecache.DistributedCache;
+import org.apache.hadoop.mapreduce.filecache.TrackerDistributedCacheManager;
/**
* Verifies if TaskRunner.SetupWorkDir() is cleaning up files/dirs pointed
@@ -149,7 +150,7 @@ public class TestSetupWorkDir extends Te
Path myTargetDir = new Path(rootDir, "./tmp");
createEmptyDir(fs, myTargetDir);
createFile(fs, myTargetDir, "cacheFile.txt");
- DistributedCache.setLocalFiles(jConf,
+ TrackerDistributedCacheManager.setLocalFiles(jConf,
(myTargetDir.toString()+Path.SEPARATOR+"cacheFile.txt"));
assertTrue("Did not create cache file in " + myTargetDir,
(fs.listStatus(myTargetDir).length == 1));
@@ -203,7 +204,7 @@ public class TestSetupWorkDir extends Te
Path myTargetDir = new Path(rootDir, "./tmp");
createEmptyDir(fs, myTargetDir);
createFile(fs, myTargetDir, "cacheFile.txt");
- DistributedCache.setLocalFiles(jConf, (myTargetDir.toString() +
+ TrackerDistributedCacheManager.setLocalFiles(jConf,
(myTargetDir.toString() +
Path.SEPARATOR+"cacheFile.txt"));
assertTrue("Did not create cache file in " + myTargetDir,
(fs.listStatus(myTargetDir).length == 1));
Modified:
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/filecache/TestTrackerDistributedCacheManager.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/filecache/TestTrackerDistributedCacheManager.java?rev=1100409&r1=1100408&r2=1100409&view=diff
==============================================================================
---
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/filecache/TestTrackerDistributedCacheManager.java
(original)
+++
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/filecache/TestTrackerDistributedCacheManager.java
Fri May 6 22:47:17 2011
@@ -97,7 +97,7 @@ public class TestTrackerDistributedCache
// during setUp() rather than in a subtle way later.
assertTrue("Test root directory " + TEST_ROOT + " and all of its " +
"parent directories must have a+x permissions",
- TrackerDistributedCacheManager.ancestorsHaveExecutePermissions(
+ ClientDistributedCacheManager.ancestorsHaveExecutePermissions(
fs, new Path(TEST_ROOT.toString())));
// Prepare the tests' mapred-local-dir
@@ -160,8 +160,8 @@ public class TestTrackerDistributedCache
subConf.set(MRJobConfig.USER_NAME, userName);
DistributedCache.addCacheFile(firstCacheFile.toUri(), subConf);
DistributedCache.addFileToClassPath(secondCacheFile, subConf);
- TrackerDistributedCacheManager.determineTimestamps(subConf);
- TrackerDistributedCacheManager.determineCacheVisibilities(subConf);
+ ClientDistributedCacheManager.determineTimestamps(subConf);
+ ClientDistributedCacheManager.determineCacheVisibilities(subConf);
// ****** End of imitating JobClient code
Path jobFile = new Path(TEST_ROOT_DIR, "job.xml");
@@ -239,8 +239,8 @@ public class TestTrackerDistributedCache
job1.setUser(userName);
job1.addCacheFile(secondCacheFile.toUri());
Configuration conf1 = job1.getConfiguration();
- TrackerDistributedCacheManager.determineTimestamps(conf1);
- TrackerDistributedCacheManager.determineCacheVisibilities(conf1);
+ ClientDistributedCacheManager.determineTimestamps(conf1);
+ ClientDistributedCacheManager.determineCacheVisibilities(conf1);
// Task localizing for first job
TaskDistributedCacheManager handle = manager
@@ -267,8 +267,8 @@ public class TestTrackerDistributedCache
// add a file that is never localized
job2.addCacheFile(thirdCacheFile.toUri());
Configuration conf2 = job2.getConfiguration();
- TrackerDistributedCacheManager.determineTimestamps(conf2);
- TrackerDistributedCacheManager.determineCacheVisibilities(conf2);
+ ClientDistributedCacheManager.determineTimestamps(conf2);
+ ClientDistributedCacheManager.determineCacheVisibilities(conf2);
// Task localizing for second job
// localization for the "firstCacheFile" will fail.
@@ -376,8 +376,8 @@ public class TestTrackerDistributedCache
job1.setUser(userName);
job1.addCacheFile(cacheFile.toUri());
Configuration conf1 = job1.getConfiguration();
- TrackerDistributedCacheManager.determineTimestamps(conf1);
- TrackerDistributedCacheManager.determineCacheVisibilities(conf1);
+ ClientDistributedCacheManager.determineTimestamps(conf1);
+ ClientDistributedCacheManager.determineCacheVisibilities(conf1);
// Task localizing for job
TaskDistributedCacheManager handle = manager
@@ -678,8 +678,8 @@ public class TestTrackerDistributedCache
Configuration subConf = new Configuration(myConf);
subConf.set(MRJobConfig.USER_NAME, userName);
DistributedCache.addCacheFile(firstCacheFile.toUri(), subConf);
- TrackerDistributedCacheManager.determineTimestamps(subConf);
- TrackerDistributedCacheManager.determineCacheVisibilities(subConf);
+ ClientDistributedCacheManager.determineTimestamps(subConf);
+ ClientDistributedCacheManager.determineCacheVisibilities(subConf);
// ****** End of imitating JobClient code
// ****** Imitate TaskRunner code.
@@ -745,8 +745,8 @@ public class TestTrackerDistributedCache
Configuration subConf2 = new Configuration(myConf);
subConf2.set(MRJobConfig.USER_NAME, userName);
DistributedCache.addCacheFile(firstCacheFile.toUri(), subConf2);
- TrackerDistributedCacheManager.determineTimestamps(subConf2);
- TrackerDistributedCacheManager.determineCacheVisibilities(subConf2);
+ ClientDistributedCacheManager.determineTimestamps(subConf2);
+ ClientDistributedCacheManager.determineCacheVisibilities(subConf2);
handle =
manager.newTaskDistributedCacheManager(subConf2);
Modified:
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/filecache/TestURIFragments.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/filecache/TestURIFragments.java?rev=1100409&r1=1100408&r2=1100409&view=diff
==============================================================================
---
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/filecache/TestURIFragments.java
(original)
+++
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/filecache/TestURIFragments.java
Fri May 6 22:47:17 2011
@@ -26,87 +26,87 @@ import org.junit.Test;
public class TestURIFragments {
/**
- * Tests {@link TrackerDistributedCacheManager#checkURIs(URI[], URI[]).
+ * Tests {@link DistributedCache#checkURIs(URI[], URI[]).
*/
@Test
public void testURIs() throws URISyntaxException {
- assertTrue(TrackerDistributedCacheManager.checkURIs(null, null));
+ assertTrue(DistributedCache.checkURIs(null, null));
// uris with no fragments
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] { new URI(
+ assertFalse(DistributedCache.checkURIs(new URI[] { new URI(
"file://foo/bar/myCacheFile.txt") }, null));
- assertFalse(TrackerDistributedCacheManager.checkURIs(null,
+ assertFalse(DistributedCache.checkURIs(null,
new URI[] { new URI("file://foo/bar/myCacheArchive.txt") }));
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] {
+ assertFalse(DistributedCache.checkURIs(new URI[] {
new URI("file://foo/bar/myCacheFile1.txt#file"),
new URI("file://foo/bar/myCacheFile2.txt") }, null));
- assertFalse(TrackerDistributedCacheManager.checkURIs(null, new URI[] {
+ assertFalse(DistributedCache.checkURIs(null, new URI[] {
new URI("file://foo/bar/myCacheArchive1.txt"),
new URI("file://foo/bar/myCacheArchive2.txt#archive") }));
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] { new URI(
+ assertFalse(DistributedCache.checkURIs(new URI[] { new URI(
"file://foo/bar/myCacheFile.txt") }, new URI[] { new URI(
"file://foo/bar/myCacheArchive.txt") }));
// conflicts in fragment names
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] {
+ assertFalse(DistributedCache.checkURIs(new URI[] {
new URI("file://foo/bar/myCacheFile1.txt#file"),
new URI("file://foo/bar/myCacheFile2.txt#file") }, null));
- assertFalse(TrackerDistributedCacheManager.checkURIs(null, new URI[] {
+ assertFalse(DistributedCache.checkURIs(null, new URI[] {
new URI("file://foo/bar/myCacheArchive1.txt#archive"),
new URI("file://foo/bar/myCacheArchive2.txt#archive") }));
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] { new URI(
+ assertFalse(DistributedCache.checkURIs(new URI[] { new URI(
"file://foo/bar/myCacheFile.txt#cache") }, new URI[] { new URI(
"file://foo/bar/myCacheArchive.txt#cache") }));
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] {
+ assertFalse(DistributedCache.checkURIs(new URI[] {
new URI("file://foo/bar/myCacheFile1.txt#file1"),
new URI("file://foo/bar/myCacheFile2.txt#file2") }, new URI[] {
new URI("file://foo/bar/myCacheArchive1.txt#archive"),
new URI("file://foo/bar/myCacheArchive2.txt#archive") }));
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] {
+ assertFalse(DistributedCache.checkURIs(new URI[] {
new URI("file://foo/bar/myCacheFile1.txt#file"),
new URI("file://foo/bar/myCacheFile2.txt#file") }, new URI[] {
new URI("file://foo/bar/myCacheArchive1.txt#archive1"),
new URI("file://foo/bar/myCacheArchive2.txt#archive2") }));
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] {
+ assertFalse(DistributedCache.checkURIs(new URI[] {
new URI("file://foo/bar/myCacheFile1.txt#file1"),
new URI("file://foo/bar/myCacheFile2.txt#cache") }, new URI[] {
new URI("file://foo/bar/myCacheArchive1.txt#cache"),
new URI("file://foo/bar/myCacheArchive2.txt#archive2") }));
// test ignore case
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] {
+ assertFalse(DistributedCache.checkURIs(new URI[] {
new URI("file://foo/bar/myCacheFile1.txt#file"),
new URI("file://foo/bar/myCacheFile2.txt#FILE") }, null));
- assertFalse(TrackerDistributedCacheManager.checkURIs(null, new URI[] {
+ assertFalse(DistributedCache.checkURIs(null, new URI[] {
new URI("file://foo/bar/myCacheArchive1.txt#archive"),
new URI("file://foo/bar/myCacheArchive2.txt#ARCHIVE") }));
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] { new URI(
+ assertFalse(DistributedCache.checkURIs(new URI[] { new URI(
"file://foo/bar/myCacheFile.txt#cache") }, new URI[] { new URI(
"file://foo/bar/myCacheArchive.txt#CACHE") }));
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] {
+ assertFalse(DistributedCache.checkURIs(new URI[] {
new URI("file://foo/bar/myCacheFile1.txt#file1"),
new URI("file://foo/bar/myCacheFile2.txt#file2") }, new URI[] {
new URI("file://foo/bar/myCacheArchive1.txt#ARCHIVE"),
new URI("file://foo/bar/myCacheArchive2.txt#archive") }));
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] {
+ assertFalse(DistributedCache.checkURIs(new URI[] {
new URI("file://foo/bar/myCacheFile1.txt#FILE"),
new URI("file://foo/bar/myCacheFile2.txt#file") }, new URI[] {
new URI("file://foo/bar/myCacheArchive1.txt#archive1"),
new URI("file://foo/bar/myCacheArchive2.txt#archive2") }));
- assertFalse(TrackerDistributedCacheManager.checkURIs(new URI[] {
+ assertFalse(DistributedCache.checkURIs(new URI[] {
new URI("file://foo/bar/myCacheFile1.txt#file1"),
new URI("file://foo/bar/myCacheFile2.txt#CACHE") }, new URI[] {
new URI("file://foo/bar/myCacheArchive1.txt#cache"),
new URI("file://foo/bar/myCacheArchive2.txt#archive2") }));
// allowed uri combinations
- assertTrue(TrackerDistributedCacheManager.checkURIs(new URI[] {
+ assertTrue(DistributedCache.checkURIs(new URI[] {
new URI("file://foo/bar/myCacheFile1.txt#file1"),
new URI("file://foo/bar/myCacheFile2.txt#file2") }, null));
- assertTrue(TrackerDistributedCacheManager.checkURIs(null, new URI[] {
+ assertTrue(DistributedCache.checkURIs(null, new URI[] {
new URI("file://foo/bar/myCacheArchive1.txt#archive1"),
new URI("file://foo/bar/myCacheArchive2.txt#archive2") }));
- assertTrue(TrackerDistributedCacheManager.checkURIs(new URI[] {
+ assertTrue(DistributedCache.checkURIs(new URI[] {
new URI("file://foo/bar/myCacheFile1.txt#file1"),
new URI("file://foo/bar/myCacheFile2.txt#file2") }, new URI[] {
new URI("file://foo/bar/myCacheArchive1.txt#archive1"),
