001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.fs;
019
020import java.io.FileNotFoundException;
021import java.io.IOException;
022import java.io.InputStream;
023import java.io.OutputStream;
024import java.net.URI;
025import java.security.PrivilegedExceptionAction;
026import java.util.ArrayList;
027import java.util.EnumSet;
028import java.util.HashSet;
029import java.util.IdentityHashMap;
030import java.util.List;
031import java.util.Map;
032import java.util.Set;
033import java.util.Stack;
034import java.util.TreeSet;
035import java.util.Map.Entry;
036
037import org.apache.commons.logging.Log;
038import org.apache.commons.logging.LogFactory;
039import org.apache.hadoop.HadoopIllegalArgumentException;
040import org.apache.hadoop.classification.InterfaceAudience;
041import org.apache.hadoop.classification.InterfaceStability;
042import org.apache.hadoop.conf.Configuration;
043import org.apache.hadoop.fs.FileSystem.Statistics;
044import org.apache.hadoop.fs.Options.CreateOpts;
045import org.apache.hadoop.fs.permission.AclEntry;
046import org.apache.hadoop.fs.permission.AclStatus;
047import org.apache.hadoop.fs.permission.FsAction;
048import org.apache.hadoop.fs.permission.FsPermission;
049import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY;
050import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_DEFAULT;
051import org.apache.hadoop.io.IOUtils;
052import org.apache.hadoop.ipc.RpcClientException;
053import org.apache.hadoop.ipc.RpcServerException;
054import org.apache.hadoop.ipc.UnexpectedServerException;
055import org.apache.hadoop.fs.InvalidPathException;
056import org.apache.hadoop.security.AccessControlException;
057import org.apache.hadoop.security.UserGroupInformation;
058import org.apache.hadoop.security.token.Token;
059import org.apache.hadoop.util.ShutdownHookManager;
060
061/**
062 * The FileContext class provides an interface to the application writer for
063 * using the Hadoop file system.
064 * It provides a set of methods for the usual operation: create, open, 
065 * list, etc 
066 * 
067 * <p>
068 * <b> *** Path Names *** </b>
069 * <p>
070 * 
071 * The Hadoop file system supports a URI name space and URI names.
072 * It offers a forest of file systems that can be referenced using fully
073 * qualified URIs.
074 * Two common Hadoop file systems implementations are
075 * <ul>
076 * <li> the local file system: file:///path
077 * <li> the hdfs file system hdfs://nnAddress:nnPort/path
078 * </ul>
079 * 
080 * While URI names are very flexible, it requires knowing the name or address
081 * of the server. For convenience one often wants to access the default system
082 * in one's environment without knowing its name/address. This has an
083 * additional benefit that it allows one to change one's default fs
084 *  (e.g. admin moves application from cluster1 to cluster2).
085 * <p>
086 * 
087 * To facilitate this, Hadoop supports a notion of a default file system.
088 * The user can set his default file system, although this is
089 * typically set up for you in your environment via your default config.
090 * A default file system implies a default scheme and authority; slash-relative
091 * names (such as /for/bar) are resolved relative to that default FS.
092 * Similarly a user can also have working-directory-relative names (i.e. names
093 * not starting with a slash). While the working directory is generally in the
094 * same default FS, the wd can be in a different FS.
095 * <p>
096 *  Hence Hadoop path names can be one of:
097 *  <ul>
098 *  <li> fully qualified URI: scheme://authority/path
099 *  <li> slash relative names: /path relative to the default file system
100 *  <li> wd-relative names: path  relative to the working dir
101 *  </ul>   
102 *  Relative paths with scheme (scheme:foo/bar) are illegal.
103 *  
104 *  <p>
105 *  <b>****The Role of the FileContext and configuration defaults****</b>
106 *  <p>
107 *  The FileContext provides file namespace context for resolving file names;
108 *  it also contains the umask for permissions, In that sense it is like the
109 *  per-process file-related state in Unix system.
110 *  These two properties
111 *  <ul> 
112 *  <li> default file system i.e your slash)
113 *  <li> umask
114 *  </ul>
115 *  in general, are obtained from the default configuration file
116 *  in your environment,  (@see {@link Configuration}).
117 *  
118 *  No other configuration parameters are obtained from the default config as 
119 *  far as the file context layer is concerned. All file system instances
120 *  (i.e. deployments of file systems) have default properties; we call these
121 *  server side (SS) defaults. Operation like create allow one to select many 
122 *  properties: either pass them in as explicit parameters or use
123 *  the SS properties.
124 *  <p>
125 *  The file system related SS defaults are
126 *  <ul>
127 *  <li> the home directory (default is "/user/userName")
128 *  <li> the initial wd (only for local fs)
129 *  <li> replication factor
130 *  <li> block size
131 *  <li> buffer size
132 *  <li> encryptDataTransfer 
133 *  <li> checksum option. (checksumType and  bytesPerChecksum)
134 *  </ul>
135 *
136 * <p>
137 * <b> *** Usage Model for the FileContext class *** </b>
138 * <p>
139 * Example 1: use the default config read from the $HADOOP_CONFIG/core.xml.
140 *   Unspecified values come from core-defaults.xml in the release jar.
141 *  <ul>  
142 *  <li> myFContext = FileContext.getFileContext(); // uses the default config
143 *                                                // which has your default FS 
144 *  <li>  myFContext.create(path, ...);
145 *  <li>  myFContext.setWorkingDir(path)
146 *  <li>  myFContext.open (path, ...);  
147 *  </ul>  
148 * Example 2: Get a FileContext with a specific URI as the default FS
149 *  <ul>  
150 *  <li> myFContext = FileContext.getFileContext(URI)
151 *  <li> myFContext.create(path, ...);
152 *   ...
153 * </ul> 
154 * Example 3: FileContext with local file system as the default
155 *  <ul> 
156 *  <li> myFContext = FileContext.getLocalFSFileContext()
157 *  <li> myFContext.create(path, ...);
158 *  <li> ...
159 *  </ul> 
160 * Example 4: Use a specific config, ignoring $HADOOP_CONFIG
161 *  Generally you should not need use a config unless you are doing
162 *   <ul> 
163 *   <li> configX = someConfigSomeOnePassedToYou.
164 *   <li> myFContext = getFileContext(configX); // configX is not changed,
165 *                                              // is passed down 
166 *   <li> myFContext.create(path, ...);
167 *   <li>...
168 *  </ul>                                          
169 *    
170 */
171
172@InterfaceAudience.Public
173@InterfaceStability.Evolving /*Evolving for a release,to be changed to Stable */
174public class FileContext {
175  
176  public static final Log LOG = LogFactory.getLog(FileContext.class);
177  /**
178   * Default permission for directory and symlink
179   * In previous versions, this default permission was also used to
180   * create files, so files created end up with ugo+x permission.
181   * See HADOOP-9155 for detail. 
182   * Two new constants are added to solve this, please use 
183   * {@link FileContext#DIR_DEFAULT_PERM} for directory, and use
184   * {@link FileContext#FILE_DEFAULT_PERM} for file.
185   * This constant is kept for compatibility.
186   */
187  public static final FsPermission DEFAULT_PERM = FsPermission.getDefault();
188  /**
189   * Default permission for directory
190   */
191  public static final FsPermission DIR_DEFAULT_PERM = FsPermission.getDirDefault();
192  /**
193   * Default permission for file
194   */
195  public static final FsPermission FILE_DEFAULT_PERM = FsPermission.getFileDefault();
196
197  /**
198   * Priority of the FileContext shutdown hook.
199   */
200  public static final int SHUTDOWN_HOOK_PRIORITY = 20;
201
202  /**
203   * List of files that should be deleted on JVM shutdown.
204   */
205  static final Map<FileContext, Set<Path>> DELETE_ON_EXIT = 
206    new IdentityHashMap<FileContext, Set<Path>>();
207
208  /** JVM shutdown hook thread. */
209  static final FileContextFinalizer FINALIZER = 
210    new FileContextFinalizer();
211  
212  private static final PathFilter DEFAULT_FILTER = new PathFilter() {
213    @Override
214    public boolean accept(final Path file) {
215      return true;
216    }
217  };
218  
219  /**
220   * The FileContext is defined by.
221   *  1) defaultFS (slash)
222   *  2) wd
223   *  3) umask
224   */   
225  private final AbstractFileSystem defaultFS; //default FS for this FileContext.
226  private Path workingDir;          // Fully qualified
227  private FsPermission umask;
228  private final Configuration conf;
229  private final UserGroupInformation ugi;
230  final boolean resolveSymlinks;
231
232  private FileContext(final AbstractFileSystem defFs,
233    final FsPermission theUmask, final Configuration aConf) {
234    defaultFS = defFs;
235    umask = FsPermission.getUMask(aConf);
236    conf = aConf;
237    try {
238      ugi = UserGroupInformation.getCurrentUser();
239    } catch (IOException e) {
240      LOG.error("Exception in getCurrentUser: ",e);
241      throw new RuntimeException("Failed to get the current user " +
242                "while creating a FileContext", e);
243    }
244    /*
245     * Init the wd.
246     * WorkingDir is implemented at the FileContext layer 
247     * NOT at the AbstractFileSystem layer. 
248     * If the DefaultFS, such as localFilesystem has a notion of
249     *  builtin WD, we use that as the initial WD.
250     *  Otherwise the WD is initialized to the home directory.
251     */
252    workingDir = defaultFS.getInitialWorkingDirectory();
253    if (workingDir == null) {
254      workingDir = defaultFS.getHomeDirectory();
255    }
256    resolveSymlinks = conf.getBoolean(
257        CommonConfigurationKeys.FS_CLIENT_RESOLVE_REMOTE_SYMLINKS_KEY,
258        CommonConfigurationKeys.FS_CLIENT_RESOLVE_REMOTE_SYMLINKS_DEFAULT);
259    util = new Util(); // for the inner class
260  }
261
262  /* 
263   * Remove relative part - return "absolute":
264   * If input is relative path ("foo/bar") add wd: ie "/<workingDir>/foo/bar"
265   * A fully qualified uri ("hdfs://nn:p/foo/bar") or a slash-relative path
266   * ("/foo/bar") are returned unchanged.
267   * 
268   * Applications that use FileContext should use #makeQualified() since
269   * they really want a fully qualified URI.
270   * Hence this method is not called makeAbsolute() and 
271   * has been deliberately declared private.
272   */
273  Path fixRelativePart(Path p) {
274    if (p.isUriPathAbsolute()) {
275      return p;
276    } else {
277      return new Path(workingDir, p);
278    }
279  }
280
281  /**
282   * Delete all the paths that were marked as delete-on-exit.
283   */
284  static void processDeleteOnExit() {
285    synchronized (DELETE_ON_EXIT) {
286      Set<Entry<FileContext, Set<Path>>> set = DELETE_ON_EXIT.entrySet();
287      for (Entry<FileContext, Set<Path>> entry : set) {
288        FileContext fc = entry.getKey();
289        Set<Path> paths = entry.getValue();
290        for (Path path : paths) {
291          try {
292            fc.delete(path, true);
293          } catch (IOException e) {
294            LOG.warn("Ignoring failure to deleteOnExit for path " + path);
295          }
296        }
297      }
298      DELETE_ON_EXIT.clear();
299    }
300  }
301
302  /**
303   * Get the file system of supplied path.
304   * 
305   * @param absOrFqPath - absolute or fully qualified path
306   * @return the file system of the path
307   * 
308   * @throws UnsupportedFileSystemException If the file system for
309   *           <code>absOrFqPath</code> is not supported.
310   * @throws IOExcepton If the file system for <code>absOrFqPath</code> could
311   *         not be instantiated.
312   */
313  protected AbstractFileSystem getFSofPath(final Path absOrFqPath)
314      throws UnsupportedFileSystemException, IOException {
315    absOrFqPath.checkNotSchemeWithRelative();
316    absOrFqPath.checkNotRelative();
317
318    try { 
319      // Is it the default FS for this FileContext?
320      defaultFS.checkPath(absOrFqPath);
321      return defaultFS;
322    } catch (Exception e) { // it is different FileSystem
323      return getAbstractFileSystem(ugi, absOrFqPath.toUri(), conf);
324    }
325  }
326  
327  private static AbstractFileSystem getAbstractFileSystem(
328      UserGroupInformation user, final URI uri, final Configuration conf)
329      throws UnsupportedFileSystemException, IOException {
330    try {
331      return user.doAs(new PrivilegedExceptionAction<AbstractFileSystem>() {
332        @Override
333        public AbstractFileSystem run() throws UnsupportedFileSystemException {
334          return AbstractFileSystem.get(uri, conf);
335        }
336      });
337    } catch (InterruptedException ex) {
338      LOG.error(ex);
339      throw new IOException("Failed to get the AbstractFileSystem for path: "
340          + uri, ex);
341    }
342  }
343  
344  /**
345   * Protected Static Factory methods for getting a FileContexts
346   * that take a AbstractFileSystem as input. To be used for testing.
347   */
348
349  /**
350   * Create a FileContext with specified FS as default using the specified
351   * config.
352   * 
353   * @param defFS
354   * @param aConf
355   * @return new FileContext with specifed FS as default.
356   */
357  public static FileContext getFileContext(final AbstractFileSystem defFS,
358                    final Configuration aConf) {
359    return new FileContext(defFS, FsPermission.getUMask(aConf), aConf);
360  }
361  
362  /**
363   * Create a FileContext for specified file system using the default config.
364   * 
365   * @param defaultFS
366   * @return a FileContext with the specified AbstractFileSystem
367   *                 as the default FS.
368   */
369  protected static FileContext getFileContext(
370    final AbstractFileSystem defaultFS) {
371    return getFileContext(defaultFS, new Configuration());
372  }
373 
374  /**
375   * Static Factory methods for getting a FileContext.
376   * Note new file contexts are created for each call.
377   * The only singleton is the local FS context using the default config.
378   * 
379   * Methods that use the default config: the default config read from the
380   * $HADOOP_CONFIG/core.xml,
381   * Unspecified key-values for config are defaulted from core-defaults.xml
382   * in the release jar.
383   * 
384   * The keys relevant to the FileContext layer are extracted at time of
385   * construction. Changes to the config after the call are ignore
386   * by the FileContext layer. 
387   * The conf is passed to lower layers like AbstractFileSystem and HDFS which
388   * pick up their own config variables.
389   */
390
391  /**
392   * Create a FileContext using the default config read from the
393   * $HADOOP_CONFIG/core.xml, Unspecified key-values for config are defaulted
394   * from core-defaults.xml in the release jar.
395   * 
396   * @throws UnsupportedFileSystemException If the file system from the default
397   *           configuration is not supported
398   */
399  public static FileContext getFileContext()
400      throws UnsupportedFileSystemException {
401    return getFileContext(new Configuration());
402  }
403
404  /**
405   * @return a FileContext for the local file system using the default config.
406   * @throws UnsupportedFileSystemException If the file system for
407   *           {@link FsConstants#LOCAL_FS_URI} is not supported.
408   */
409  public static FileContext getLocalFSFileContext()
410      throws UnsupportedFileSystemException {
411    return getFileContext(FsConstants.LOCAL_FS_URI);
412  }
413
414  /**
415   * Create a FileContext for specified URI using the default config.
416   * 
417   * @param defaultFsUri
418   * @return a FileContext with the specified URI as the default FS.
419   * 
420   * @throws UnsupportedFileSystemException If the file system for
421   *           <code>defaultFsUri</code> is not supported
422   */
423  public static FileContext getFileContext(final URI defaultFsUri)
424      throws UnsupportedFileSystemException {
425    return getFileContext(defaultFsUri, new Configuration());
426  }
427
428  /**
429   * Create a FileContext for specified default URI using the specified config.
430   * 
431   * @param defaultFsUri
432   * @param aConf
433   * @return new FileContext for specified uri
434   * @throws UnsupportedFileSystemException If the file system with specified is
435   *           not supported
436   * @throws RuntimeException If the file system specified is supported but
437   *         could not be instantiated, or if login fails.
438   */
439  public static FileContext getFileContext(final URI defaultFsUri,
440      final Configuration aConf) throws UnsupportedFileSystemException {
441    UserGroupInformation currentUser = null;
442    AbstractFileSystem defaultAfs = null;
443    if (defaultFsUri.getScheme() == null) {
444      return getFileContext(aConf);
445    }
446    try {
447      currentUser = UserGroupInformation.getCurrentUser();
448      defaultAfs = getAbstractFileSystem(currentUser, defaultFsUri, aConf);
449    } catch (UnsupportedFileSystemException ex) {
450      throw ex;
451    } catch (IOException ex) {
452      LOG.error(ex);
453      throw new RuntimeException(ex);
454    }
455    return getFileContext(defaultAfs, aConf);
456  }
457
458  /**
459   * Create a FileContext using the passed config. Generally it is better to use
460   * {@link #getFileContext(URI, Configuration)} instead of this one.
461   * 
462   * 
463   * @param aConf
464   * @return new FileContext
465   * @throws UnsupportedFileSystemException If file system in the config
466   *           is not supported
467   */
468  public static FileContext getFileContext(final Configuration aConf)
469      throws UnsupportedFileSystemException {
470    final URI defaultFsUri = URI.create(aConf.get(FS_DEFAULT_NAME_KEY,
471        FS_DEFAULT_NAME_DEFAULT));
472    if (   defaultFsUri.getScheme() != null
473        && !defaultFsUri.getScheme().trim().isEmpty()) {
474      return getFileContext(defaultFsUri, aConf);
475    }
476    throw new UnsupportedFileSystemException(String.format(
477        "%s: URI configured via %s carries no scheme",
478        defaultFsUri, FS_DEFAULT_NAME_KEY));
479  }
480
481  /**
482   * @param aConf - from which the FileContext is configured
483   * @return a FileContext for the local file system using the specified config.
484   * 
485   * @throws UnsupportedFileSystemException If default file system in the config
486   *           is not supported
487   * 
488   */
489  public static FileContext getLocalFSFileContext(final Configuration aConf)
490      throws UnsupportedFileSystemException {
491    return getFileContext(FsConstants.LOCAL_FS_URI, aConf);
492  }
493
494  /* This method is needed for tests. */
495  @InterfaceAudience.Private
496  @InterfaceStability.Unstable /* return type will change to AFS once
497                                  HADOOP-6223 is completed */
498  public AbstractFileSystem getDefaultFileSystem() {
499    return defaultFS;
500  }
501  
502  /**
503   * Set the working directory for wd-relative names (such a "foo/bar"). Working
504   * directory feature is provided by simply prefixing relative names with the
505   * working dir. Note this is different from Unix where the wd is actually set
506   * to the inode. Hence setWorkingDir does not follow symlinks etc. This works
507   * better in a distributed environment that has multiple independent roots.
508   * {@link #getWorkingDirectory()} should return what setWorkingDir() set.
509   * 
510   * @param newWDir new working directory
511   * @throws IOException 
512   * <br>
513   *           NewWdir can be one of:
514   *           <ul>
515   *           <li>relative path: "foo/bar";</li>
516   *           <li>absolute without scheme: "/foo/bar"</li>
517   *           <li>fully qualified with scheme: "xx://auth/foo/bar"</li>
518   *           </ul>
519   * <br>
520   *           Illegal WDs:
521   *           <ul>
522   *           <li>relative with scheme: "xx:foo/bar"</li>
523   *           <li>non existent directory</li>
524   *           </ul>
525   */
526  public void setWorkingDirectory(final Path newWDir) throws IOException {
527    newWDir.checkNotSchemeWithRelative();
528    /* wd is stored as a fully qualified path. We check if the given 
529     * path is not relative first since resolve requires and returns 
530     * an absolute path.
531     */  
532    final Path newWorkingDir = new Path(workingDir, newWDir);
533    FileStatus status = getFileStatus(newWorkingDir);
534    if (status.isFile()) {
535      throw new FileNotFoundException("Cannot setWD to a file");
536    }
537    workingDir = newWorkingDir;
538  }
539  
540  /**
541   * Gets the working directory for wd-relative names (such a "foo/bar").
542   */
543  public Path getWorkingDirectory() {
544    return workingDir;
545  }
546  
547  /**
548   * Gets the ugi in the file-context
549   * @return UserGroupInformation
550   */
551  public UserGroupInformation getUgi() {
552    return ugi;
553  }
554  
555  /**
556   * Return the current user's home directory in this file system.
557   * The default implementation returns "/user/$USER/".
558   * @return the home directory
559   */
560  public Path getHomeDirectory() {
561    return defaultFS.getHomeDirectory();
562  }
563  
564  /**
565   * 
566   * @return the umask of this FileContext
567   */
568  public FsPermission getUMask() {
569    return umask;
570  }
571  
572  /**
573   * Set umask to the supplied parameter.
574   * @param newUmask  the new umask
575   */
576  public void setUMask(final FsPermission newUmask) {
577    umask = newUmask;
578  }
579  
580  
581  /**
582   * Resolve the path following any symlinks or mount points
583   * @param f to be resolved
584   * @return fully qualified resolved path
585   * 
586   * @throws FileNotFoundException  If <code>f</code> does not exist
587   * @throws AccessControlException if access denied
588   * @throws IOException If an IO Error occurred
589   * 
590   * Exceptions applicable to file systems accessed over RPC:
591   * @throws RpcClientException If an exception occurred in the RPC client
592   * @throws RpcServerException If an exception occurred in the RPC server
593   * @throws UnexpectedServerException If server implementation throws
594   *           undeclared exception to RPC server
595   * 
596   * RuntimeExceptions:
597   * @throws InvalidPathException If path <code>f</code> is not valid
598   */
599  public Path resolvePath(final Path f) throws FileNotFoundException,
600      UnresolvedLinkException, AccessControlException, IOException {
601    return resolve(f);
602  }
603  
604  /**
605   * Make the path fully qualified if it is isn't. 
606   * A Fully-qualified path has scheme and authority specified and an absolute
607   * path.
608   * Use the default file system and working dir in this FileContext to qualify.
609   * @param path
610   * @return qualified path
611   */
612  public Path makeQualified(final Path path) {
613    return path.makeQualified(defaultFS.getUri(), getWorkingDirectory());
614  }
615
616  /**
617   * Create or overwrite file on indicated path and returns an output stream for
618   * writing into the file.
619   * 
620   * @param f the file name to open
621   * @param createFlag gives the semantics of create; see {@link CreateFlag}
622   * @param opts file creation options; see {@link Options.CreateOpts}.
623   *          <ul>
624   *          <li>Progress - to report progress on the operation - default null
625   *          <li>Permission - umask is applied against permisssion: default is
626   *          FsPermissions:getDefault()
627   * 
628   *          <li>CreateParent - create missing parent path; default is to not
629   *          to create parents
630   *          <li>The defaults for the following are SS defaults of the file
631   *          server implementing the target path. Not all parameters make sense
632   *          for all kinds of file system - eg. localFS ignores Blocksize,
633   *          replication, checksum
634   *          <ul>
635   *          <li>BufferSize - buffersize used in FSDataOutputStream
636   *          <li>Blocksize - block size for file blocks
637   *          <li>ReplicationFactor - replication for blocks
638   *          <li>ChecksumParam - Checksum parameters. server default is used
639   *          if not specified.
640   *          </ul>
641   *          </ul>
642   * 
643   * @return {@link FSDataOutputStream} for created file
644   * 
645   * @throws AccessControlException If access is denied
646   * @throws FileAlreadyExistsException If file <code>f</code> already exists
647   * @throws FileNotFoundException If parent of <code>f</code> does not exist
648   *           and <code>createParent</code> is false
649   * @throws ParentNotDirectoryException If parent of <code>f</code> is not a
650   *           directory.
651   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
652   *           not supported
653   * @throws IOException If an I/O error occurred
654   * 
655   * Exceptions applicable to file systems accessed over RPC:
656   * @throws RpcClientException If an exception occurred in the RPC client
657   * @throws RpcServerException If an exception occurred in the RPC server
658   * @throws UnexpectedServerException If server implementation throws
659   *           undeclared exception to RPC server
660   * 
661   * RuntimeExceptions:
662   * @throws InvalidPathException If path <code>f</code> is not valid
663   */
664  public FSDataOutputStream create(final Path f,
665      final EnumSet<CreateFlag> createFlag, Options.CreateOpts... opts)
666      throws AccessControlException, FileAlreadyExistsException,
667      FileNotFoundException, ParentNotDirectoryException,
668      UnsupportedFileSystemException, IOException {
669    Path absF = fixRelativePart(f);
670
671    // If one of the options is a permission, extract it & apply umask
672    // If not, add a default Perms and apply umask;
673    // AbstractFileSystem#create
674
675    CreateOpts.Perms permOpt = CreateOpts.getOpt(CreateOpts.Perms.class, opts);
676    FsPermission permission = (permOpt != null) ? permOpt.getValue() :
677                                      FILE_DEFAULT_PERM;
678    permission = permission.applyUMask(umask);
679
680    final CreateOpts[] updatedOpts = 
681                      CreateOpts.setOpt(CreateOpts.perms(permission), opts);
682    return new FSLinkResolver<FSDataOutputStream>() {
683      @Override
684      public FSDataOutputStream next(final AbstractFileSystem fs, final Path p) 
685        throws IOException {
686        return fs.create(p, createFlag, updatedOpts);
687      }
688    }.resolve(this, absF);
689  }
690
691  /**
692   * Make(create) a directory and all the non-existent parents.
693   * 
694   * @param dir - the dir to make
695   * @param permission - permissions is set permission&~umask
696   * @param createParent - if true then missing parent dirs are created if false
697   *          then parent must exist
698   * 
699   * @throws AccessControlException If access is denied
700   * @throws FileAlreadyExistsException If directory <code>dir</code> already
701   *           exists
702   * @throws FileNotFoundException If parent of <code>dir</code> does not exist
703   *           and <code>createParent</code> is false
704   * @throws ParentNotDirectoryException If parent of <code>dir</code> is not a
705   *           directory
706   * @throws UnsupportedFileSystemException If file system for <code>dir</code>
707   *         is not supported
708   * @throws IOException If an I/O error occurred
709   * 
710   * Exceptions applicable to file systems accessed over RPC:
711   * @throws RpcClientException If an exception occurred in the RPC client
712   * @throws UnexpectedServerException If server implementation throws 
713   *           undeclared exception to RPC server
714   * 
715   * RuntimeExceptions:
716   * @throws InvalidPathException If path <code>dir</code> is not valid
717   */
718  public void mkdir(final Path dir, final FsPermission permission,
719      final boolean createParent) throws AccessControlException,
720      FileAlreadyExistsException, FileNotFoundException,
721      ParentNotDirectoryException, UnsupportedFileSystemException, 
722      IOException {
723    final Path absDir = fixRelativePart(dir);
724    final FsPermission absFerms = (permission == null ? 
725          FsPermission.getDirDefault() : permission).applyUMask(umask);
726    new FSLinkResolver<Void>() {
727      @Override
728      public Void next(final AbstractFileSystem fs, final Path p) 
729        throws IOException, UnresolvedLinkException {
730        fs.mkdir(p, absFerms, createParent);
731        return null;
732      }
733    }.resolve(this, absDir);
734  }
735
736  /**
737   * Delete a file.
738   * @param f the path to delete.
739   * @param recursive if path is a directory and set to 
740   * true, the directory is deleted else throws an exception. In
741   * case of a file the recursive can be set to either true or false.
742   *
743   * @throws AccessControlException If access is denied
744   * @throws FileNotFoundException If <code>f</code> does not exist
745   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
746   *           not supported
747   * @throws IOException If an I/O error occurred
748   * 
749   * Exceptions applicable to file systems accessed over RPC:
750   * @throws RpcClientException If an exception occurred in the RPC client
751   * @throws RpcServerException If an exception occurred in the RPC server
752   * @throws UnexpectedServerException If server implementation throws 
753   *           undeclared exception to RPC server
754   * 
755   * RuntimeExceptions:
756   * @throws InvalidPathException If path <code>f</code> is invalid
757   */
758  public boolean delete(final Path f, final boolean recursive)
759      throws AccessControlException, FileNotFoundException,
760      UnsupportedFileSystemException, IOException {
761    Path absF = fixRelativePart(f);
762    return new FSLinkResolver<Boolean>() {
763      @Override
764      public Boolean next(final AbstractFileSystem fs, final Path p) 
765        throws IOException, UnresolvedLinkException {
766        return Boolean.valueOf(fs.delete(p, recursive));
767      }
768    }.resolve(this, absF);
769  }
770 
771  /**
772   * Opens an FSDataInputStream at the indicated Path using
773   * default buffersize.
774   * @param f the file name to open
775   *
776   * @throws AccessControlException If access is denied
777   * @throws FileNotFoundException If file <code>f</code> does not exist
778   * @throws UnsupportedFileSystemException If file system for <code>f</code>
779   *         is not supported
780   * @throws IOException If an I/O error occurred
781   * 
782   * Exceptions applicable to file systems accessed over RPC:
783   * @throws RpcClientException If an exception occurred in the RPC client
784   * @throws RpcServerException If an exception occurred in the RPC server
785   * @throws UnexpectedServerException If server implementation throws 
786   *           undeclared exception to RPC server
787   */
788  public FSDataInputStream open(final Path f) throws AccessControlException,
789      FileNotFoundException, UnsupportedFileSystemException, IOException {
790    final Path absF = fixRelativePart(f);
791    return new FSLinkResolver<FSDataInputStream>() {
792      @Override
793      public FSDataInputStream next(final AbstractFileSystem fs, final Path p) 
794        throws IOException, UnresolvedLinkException {
795        return fs.open(p);
796      }
797    }.resolve(this, absF);
798  }
799
800  /**
801   * Opens an FSDataInputStream at the indicated Path.
802   * 
803   * @param f the file name to open
804   * @param bufferSize the size of the buffer to be used.
805   * 
806   * @throws AccessControlException If access is denied
807   * @throws FileNotFoundException If file <code>f</code> does not exist
808   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
809   *           not supported
810   * @throws IOException If an I/O error occurred
811   * 
812   * Exceptions applicable to file systems accessed over RPC:
813   * @throws RpcClientException If an exception occurred in the RPC client
814   * @throws RpcServerException If an exception occurred in the RPC server
815   * @throws UnexpectedServerException If server implementation throws 
816   *           undeclared exception to RPC server
817   */
818  public FSDataInputStream open(final Path f, final int bufferSize)
819      throws AccessControlException, FileNotFoundException,
820      UnsupportedFileSystemException, IOException {
821    final Path absF = fixRelativePart(f);
822    return new FSLinkResolver<FSDataInputStream>() {
823      @Override
824      public FSDataInputStream next(final AbstractFileSystem fs, final Path p) 
825        throws IOException, UnresolvedLinkException {
826        return fs.open(p, bufferSize);
827      }
828    }.resolve(this, absF);
829  }
830
831  /**
832   * Truncate the file in the indicated path to the indicated size.
833   * <ul>
834   * <li>Fails if path is a directory.
835   * <li>Fails if path does not exist.
836   * <li>Fails if path is not closed.
837   * <li>Fails if new size is greater than current size.
838   * </ul>
839   * @param f The path to the file to be truncated
840   * @param newLength The size the file is to be truncated to
841   *
842   * @return <code>true</code> if the file has been truncated to the desired
843   * <code>newLength</code> and is immediately available to be reused for
844   * write operations such as <code>append</code>, or
845   * <code>false</code> if a background process of adjusting the length of
846   * the last block has been started, and clients should wait for it to
847   * complete before proceeding with further file updates.
848   *
849   * @throws AccessControlException If access is denied
850   * @throws FileNotFoundException If file <code>f</code> does not exist
851   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
852   *           not supported
853   * @throws IOException If an I/O error occurred
854   *
855   * Exceptions applicable to file systems accessed over RPC:
856   * @throws RpcClientException If an exception occurred in the RPC client
857   * @throws RpcServerException If an exception occurred in the RPC server
858   * @throws UnexpectedServerException If server implementation throws
859   *           undeclared exception to RPC server
860   */
861  public boolean truncate(final Path f, final long newLength)
862      throws AccessControlException, FileNotFoundException,
863      UnsupportedFileSystemException, IOException {
864    final Path absF = fixRelativePart(f);
865    return new FSLinkResolver<Boolean>() {
866      @Override
867      public Boolean next(final AbstractFileSystem fs, final Path p)
868          throws IOException, UnresolvedLinkException {
869        return fs.truncate(p, newLength);
870      }
871    }.resolve(this, absF);
872  }
873
874  /**
875   * Set replication for an existing file.
876   * 
877   * @param f file name
878   * @param replication new replication
879   *
880   * @return true if successful
881   *
882   * @throws AccessControlException If access is denied
883   * @throws FileNotFoundException If file <code>f</code> does not exist
884   * @throws IOException If an I/O error occurred
885   * 
886   * Exceptions applicable to file systems accessed over RPC:
887   * @throws RpcClientException If an exception occurred in the RPC client
888   * @throws RpcServerException If an exception occurred in the RPC server
889   * @throws UnexpectedServerException If server implementation throws 
890   *           undeclared exception to RPC server
891   */
892  public boolean setReplication(final Path f, final short replication)
893      throws AccessControlException, FileNotFoundException,
894      IOException {
895    final Path absF = fixRelativePart(f);
896    return new FSLinkResolver<Boolean>() {
897      @Override
898      public Boolean next(final AbstractFileSystem fs, final Path p) 
899        throws IOException, UnresolvedLinkException {
900        return Boolean.valueOf(fs.setReplication(p, replication));
901      }
902    }.resolve(this, absF);
903  }
904
905  /**
906   * Renames Path src to Path dst
907   * <ul>
908   * <li
909   * <li>Fails if src is a file and dst is a directory.
910   * <li>Fails if src is a directory and dst is a file.
911   * <li>Fails if the parent of dst does not exist or is a file.
912   * </ul>
913   * <p>
914   * If OVERWRITE option is not passed as an argument, rename fails if the dst
915   * already exists.
916   * <p>
917   * If OVERWRITE option is passed as an argument, rename overwrites the dst if
918   * it is a file or an empty directory. Rename fails if dst is a non-empty
919   * directory.
920   * <p>
921   * Note that atomicity of rename is dependent on the file system
922   * implementation. Please refer to the file system documentation for details
923   * <p>
924   * 
925   * @param src path to be renamed
926   * @param dst new path after rename
927   * 
928   * @throws AccessControlException If access is denied
929   * @throws FileAlreadyExistsException If <code>dst</code> already exists and
930   *           <code>options</options> has {@link Options.Rename#OVERWRITE} 
931   *           option false.
932   * @throws FileNotFoundException If <code>src</code> does not exist
933   * @throws ParentNotDirectoryException If parent of <code>dst</code> is not a
934   *           directory
935   * @throws UnsupportedFileSystemException If file system for <code>src</code>
936   *           and <code>dst</code> is not supported
937   * @throws IOException If an I/O error occurred
938   * 
939   * Exceptions applicable to file systems accessed over RPC:
940   * @throws RpcClientException If an exception occurred in the RPC client
941   * @throws RpcServerException If an exception occurred in the RPC server
942   * @throws UnexpectedServerException If server implementation throws
943   *           undeclared exception to RPC server
944   */
945  public void rename(final Path src, final Path dst,
946      final Options.Rename... options) throws AccessControlException,
947      FileAlreadyExistsException, FileNotFoundException,
948      ParentNotDirectoryException, UnsupportedFileSystemException,
949      IOException {
950    final Path absSrc = fixRelativePart(src);
951    final Path absDst = fixRelativePart(dst);
952    AbstractFileSystem srcFS = getFSofPath(absSrc);
953    AbstractFileSystem dstFS = getFSofPath(absDst);
954    if(!srcFS.getUri().equals(dstFS.getUri())) {
955      throw new IOException("Renames across AbstractFileSystems not supported");
956    }
957    try {
958      srcFS.rename(absSrc, absDst, options);
959    } catch (UnresolvedLinkException e) {
960      /* We do not know whether the source or the destination path
961       * was unresolved. Resolve the source path up until the final
962       * path component, then fully resolve the destination. 
963       */
964      final Path source = resolveIntermediate(absSrc);    
965      new FSLinkResolver<Void>() {
966        @Override
967        public Void next(final AbstractFileSystem fs, final Path p) 
968          throws IOException, UnresolvedLinkException {
969          fs.rename(source, p, options);
970          return null;
971        }
972      }.resolve(this, absDst);
973    }
974  }
975  
976  /**
977   * Set permission of a path.
978   * @param f
979   * @param permission - the new absolute permission (umask is not applied)
980   *
981   * @throws AccessControlException If access is denied
982   * @throws FileNotFoundException If <code>f</code> does not exist
983   * @throws UnsupportedFileSystemException If file system for <code>f</code>
984   *         is not supported
985   * @throws IOException If an I/O error occurred
986   * 
987   * Exceptions applicable to file systems accessed over RPC:
988   * @throws RpcClientException If an exception occurred in the RPC client
989   * @throws RpcServerException If an exception occurred in the RPC server
990   * @throws UnexpectedServerException If server implementation throws 
991   *           undeclared exception to RPC server
992   */
993  public void setPermission(final Path f, final FsPermission permission)
994      throws AccessControlException, FileNotFoundException,
995      UnsupportedFileSystemException, IOException {
996    final Path absF = fixRelativePart(f);
997    new FSLinkResolver<Void>() {
998      @Override
999      public Void next(final AbstractFileSystem fs, final Path p) 
1000        throws IOException, UnresolvedLinkException {
1001        fs.setPermission(p, permission);
1002        return null;
1003      }
1004    }.resolve(this, absF);
1005  }
1006
1007  /**
1008   * Set owner of a path (i.e. a file or a directory). The parameters username
1009   * and groupname cannot both be null.
1010   * 
1011   * @param f The path
1012   * @param username If it is null, the original username remains unchanged.
1013   * @param groupname If it is null, the original groupname remains unchanged.
1014   * 
1015   * @throws AccessControlException If access is denied
1016   * @throws FileNotFoundException If <code>f</code> does not exist
1017   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1018   *           not supported
1019   * @throws IOException If an I/O error occurred
1020   * 
1021   * Exceptions applicable to file systems accessed over RPC:
1022   * @throws RpcClientException If an exception occurred in the RPC client
1023   * @throws RpcServerException If an exception occurred in the RPC server
1024   * @throws UnexpectedServerException If server implementation throws 
1025   *           undeclared exception to RPC server
1026   * 
1027   * RuntimeExceptions:
1028   * @throws HadoopIllegalArgumentException If <code>username</code> or
1029   *           <code>groupname</code> is invalid.
1030   */
1031  public void setOwner(final Path f, final String username,
1032      final String groupname) throws AccessControlException,
1033      UnsupportedFileSystemException, FileNotFoundException,
1034      IOException {
1035    if ((username == null) && (groupname == null)) {
1036      throw new HadoopIllegalArgumentException(
1037          "username and groupname cannot both be null");
1038    }
1039    final Path absF = fixRelativePart(f);
1040    new FSLinkResolver<Void>() {
1041      @Override
1042      public Void next(final AbstractFileSystem fs, final Path p) 
1043        throws IOException, UnresolvedLinkException {
1044        fs.setOwner(p, username, groupname);
1045        return null;
1046      }
1047    }.resolve(this, absF);
1048  }
1049
1050  /**
1051   * Set access time of a file.
1052   * @param f The path
1053   * @param mtime Set the modification time of this file.
1054   *        The number of milliseconds since epoch (Jan 1, 1970). 
1055   *        A value of -1 means that this call should not set modification time.
1056   * @param atime Set the access time of this file.
1057   *        The number of milliseconds since Jan 1, 1970. 
1058   *        A value of -1 means that this call should not set access time.
1059   *
1060   * @throws AccessControlException If access is denied
1061   * @throws FileNotFoundException If <code>f</code> does not exist
1062   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1063   *           not supported
1064   * @throws IOException If an I/O error occurred
1065   * 
1066   * Exceptions applicable to file systems accessed over RPC:
1067   * @throws RpcClientException If an exception occurred in the RPC client
1068   * @throws RpcServerException If an exception occurred in the RPC server
1069   * @throws UnexpectedServerException If server implementation throws 
1070   *           undeclared exception to RPC server
1071   */
1072  public void setTimes(final Path f, final long mtime, final long atime)
1073      throws AccessControlException, FileNotFoundException,
1074      UnsupportedFileSystemException, IOException {
1075    final Path absF = fixRelativePart(f);
1076    new FSLinkResolver<Void>() {
1077      @Override
1078      public Void next(final AbstractFileSystem fs, final Path p) 
1079        throws IOException, UnresolvedLinkException {
1080        fs.setTimes(p, mtime, atime);
1081        return null;
1082      }
1083    }.resolve(this, absF);
1084  }
1085
1086  /**
1087   * Get the checksum of a file.
1088   *
1089   * @param f file path
1090   *
1091   * @return The file checksum.  The default return value is null,
1092   *  which indicates that no checksum algorithm is implemented
1093   *  in the corresponding FileSystem.
1094   *
1095   * @throws AccessControlException If access is denied
1096   * @throws FileNotFoundException If <code>f</code> does not exist
1097   * @throws IOException If an I/O error occurred
1098   * 
1099   * Exceptions applicable to file systems accessed over RPC:
1100   * @throws RpcClientException If an exception occurred in the RPC client
1101   * @throws RpcServerException If an exception occurred in the RPC server
1102   * @throws UnexpectedServerException If server implementation throws 
1103   *           undeclared exception to RPC server
1104   */
1105  public FileChecksum getFileChecksum(final Path f)
1106      throws AccessControlException, FileNotFoundException,
1107      IOException {
1108    final Path absF = fixRelativePart(f);
1109    return new FSLinkResolver<FileChecksum>() {
1110      @Override
1111      public FileChecksum next(final AbstractFileSystem fs, final Path p) 
1112        throws IOException, UnresolvedLinkException {
1113        return fs.getFileChecksum(p);
1114      }
1115    }.resolve(this, absF);
1116  }
1117
1118  /**
1119   * Set the verify checksum flag for the  file system denoted by the path.
1120   * This is only applicable if the 
1121   * corresponding FileSystem supports checksum. By default doesn't do anything.
1122   * @param verifyChecksum
1123   * @param f set the verifyChecksum for the Filesystem containing this path
1124   *
1125   * @throws AccessControlException If access is denied
1126   * @throws FileNotFoundException If <code>f</code> does not exist
1127   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1128   *           not supported
1129   * @throws IOException If an I/O error occurred
1130   * 
1131   * Exceptions applicable to file systems accessed over RPC:
1132   * @throws RpcClientException If an exception occurred in the RPC client
1133   * @throws RpcServerException If an exception occurred in the RPC server
1134   * @throws UnexpectedServerException If server implementation throws 
1135   *           undeclared exception to RPC server
1136   */
1137  public void setVerifyChecksum(final boolean verifyChecksum, final Path f)
1138      throws AccessControlException, FileNotFoundException,
1139      UnsupportedFileSystemException, IOException {
1140    final Path absF = resolve(fixRelativePart(f));
1141    getFSofPath(absF).setVerifyChecksum(verifyChecksum);
1142  }
1143
1144  /**
1145   * Return a file status object that represents the path.
1146   * @param f The path we want information from
1147   *
1148   * @return a FileStatus object
1149   *
1150   * @throws AccessControlException If access is denied
1151   * @throws FileNotFoundException If <code>f</code> does not exist
1152   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1153   *           not supported
1154   * @throws IOException If an I/O error occurred
1155   * 
1156   * Exceptions applicable to file systems accessed over RPC:
1157   * @throws RpcClientException If an exception occurred in the RPC client
1158   * @throws RpcServerException If an exception occurred in the RPC server
1159   * @throws UnexpectedServerException If server implementation throws 
1160   *           undeclared exception to RPC server
1161   */
1162  public FileStatus getFileStatus(final Path f) throws AccessControlException,
1163      FileNotFoundException, UnsupportedFileSystemException, IOException {
1164    final Path absF = fixRelativePart(f);
1165    return new FSLinkResolver<FileStatus>() {
1166      @Override
1167      public FileStatus next(final AbstractFileSystem fs, final Path p) 
1168        throws IOException, UnresolvedLinkException {
1169        return fs.getFileStatus(p);
1170      }
1171    }.resolve(this, absF);
1172  }
1173
1174  /**
1175   * Checks if the user can access a path.  The mode specifies which access
1176   * checks to perform.  If the requested permissions are granted, then the
1177   * method returns normally.  If access is denied, then the method throws an
1178   * {@link AccessControlException}.
1179   * <p/>
1180   * The default implementation of this method calls {@link #getFileStatus(Path)}
1181   * and checks the returned permissions against the requested permissions.
1182   * Note that the getFileStatus call will be subject to authorization checks.
1183   * Typically, this requires search (execute) permissions on each directory in
1184   * the path's prefix, but this is implementation-defined.  Any file system
1185   * that provides a richer authorization model (such as ACLs) may override the
1186   * default implementation so that it checks against that model instead.
1187   * <p>
1188   * In general, applications should avoid using this method, due to the risk of
1189   * time-of-check/time-of-use race conditions.  The permissions on a file may
1190   * change immediately after the access call returns.  Most applications should
1191   * prefer running specific file system actions as the desired user represented
1192   * by a {@link UserGroupInformation}.
1193   *
1194   * @param path Path to check
1195   * @param mode type of access to check
1196   * @throws AccessControlException if access is denied
1197   * @throws FileNotFoundException if the path does not exist
1198   * @throws UnsupportedFileSystemException if file system for <code>path</code>
1199   *   is not supported
1200   * @throws IOException see specific implementation
1201   * 
1202   * Exceptions applicable to file systems accessed over RPC:
1203   * @throws RpcClientException If an exception occurred in the RPC client
1204   * @throws RpcServerException If an exception occurred in the RPC server
1205   * @throws UnexpectedServerException If server implementation throws 
1206   *           undeclared exception to RPC server
1207   */
1208  @InterfaceAudience.LimitedPrivate({"HDFS", "Hive"})
1209  public void access(final Path path, final FsAction mode)
1210      throws AccessControlException, FileNotFoundException,
1211      UnsupportedFileSystemException, IOException {
1212    final Path absPath = fixRelativePart(path);
1213    new FSLinkResolver<Void>() {
1214      @Override
1215      public Void next(AbstractFileSystem fs, Path p) throws IOException,
1216          UnresolvedLinkException {
1217        fs.access(p, mode);
1218        return null;
1219      }
1220    }.resolve(this, absPath);
1221  }
1222
1223  /**
1224   * Return a file status object that represents the path. If the path 
1225   * refers to a symlink then the FileStatus of the symlink is returned.
1226   * The behavior is equivalent to #getFileStatus() if the underlying
1227   * file system does not support symbolic links.
1228   * @param  f The path we want information from.
1229   * @return A FileStatus object
1230   * 
1231   * @throws AccessControlException If access is denied
1232   * @throws FileNotFoundException If <code>f</code> does not exist
1233   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1234   *           not supported
1235   * @throws IOException If an I/O error occurred
1236   */
1237  public FileStatus getFileLinkStatus(final Path f)
1238      throws AccessControlException, FileNotFoundException,
1239      UnsupportedFileSystemException, IOException {
1240    final Path absF = fixRelativePart(f);
1241    return new FSLinkResolver<FileStatus>() {
1242      @Override
1243      public FileStatus next(final AbstractFileSystem fs, final Path p) 
1244        throws IOException, UnresolvedLinkException {
1245        FileStatus fi = fs.getFileLinkStatus(p);
1246        if (fi.isSymlink()) {
1247          fi.setSymlink(FSLinkResolver.qualifySymlinkTarget(fs.getUri(), p,
1248              fi.getSymlink()));
1249        }
1250        return fi;
1251      }
1252    }.resolve(this, absF);
1253  }
1254  
1255  /**
1256   * Returns the target of the given symbolic link as it was specified
1257   * when the link was created.  Links in the path leading up to the
1258   * final path component are resolved transparently.
1259   *
1260   * @param f the path to return the target of
1261   * @return The un-interpreted target of the symbolic link.
1262   * 
1263   * @throws AccessControlException If access is denied
1264   * @throws FileNotFoundException If path <code>f</code> does not exist
1265   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1266   *           not supported
1267   * @throws IOException If the given path does not refer to a symlink
1268   *           or an I/O error occurred
1269   */
1270  public Path getLinkTarget(final Path f) throws AccessControlException,
1271      FileNotFoundException, UnsupportedFileSystemException, IOException {
1272    final Path absF = fixRelativePart(f);
1273    return new FSLinkResolver<Path>() {
1274      @Override
1275      public Path next(final AbstractFileSystem fs, final Path p) 
1276        throws IOException, UnresolvedLinkException {
1277        FileStatus fi = fs.getFileLinkStatus(p);
1278        return fi.getSymlink();
1279      }
1280    }.resolve(this, absF);
1281  }
1282  
1283  /**
1284   * Return blockLocation of the given file for the given offset and len.
1285   *  For a nonexistent file or regions, null will be returned.
1286   *
1287   * This call is most helpful with DFS, where it returns 
1288   * hostnames of machines that contain the given file.
1289   * 
1290   * @param f - get blocklocations of this file
1291   * @param start position (byte offset)
1292   * @param len (in bytes)
1293   *
1294   * @return block locations for given file at specified offset of len
1295   *
1296   * @throws AccessControlException If access is denied
1297   * @throws FileNotFoundException If <code>f</code> does not exist
1298   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1299   *           not supported
1300   * @throws IOException If an I/O error occurred
1301   * 
1302   * Exceptions applicable to file systems accessed over RPC:
1303   * @throws RpcClientException If an exception occurred in the RPC client
1304   * @throws RpcServerException If an exception occurred in the RPC server
1305   * @throws UnexpectedServerException If server implementation throws 
1306   *           undeclared exception to RPC server
1307   * 
1308   * RuntimeExceptions:
1309   * @throws InvalidPathException If path <code>f</code> is invalid
1310   */
1311  @InterfaceAudience.LimitedPrivate({"HDFS", "MapReduce"})
1312  @InterfaceStability.Evolving
1313  public BlockLocation[] getFileBlockLocations(final Path f, final long start,
1314      final long len) throws AccessControlException, FileNotFoundException,
1315      UnsupportedFileSystemException, IOException {
1316    final Path absF = fixRelativePart(f);
1317    return new FSLinkResolver<BlockLocation[]>() {
1318      @Override
1319      public BlockLocation[] next(final AbstractFileSystem fs, final Path p) 
1320        throws IOException, UnresolvedLinkException {
1321        return fs.getFileBlockLocations(p, start, len);
1322      }
1323    }.resolve(this, absF);
1324  }
1325  
1326  /**
1327   * Returns a status object describing the use and capacity of the
1328   * file system denoted by the Parh argument p.
1329   * If the file system has multiple partitions, the
1330   * use and capacity of the partition pointed to by the specified
1331   * path is reflected.
1332   * 
1333   * @param f Path for which status should be obtained. null means the
1334   * root partition of the default file system. 
1335   *
1336   * @return a FsStatus object
1337   *
1338   * @throws AccessControlException If access is denied
1339   * @throws FileNotFoundException If <code>f</code> does not exist
1340   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1341   *           not supported
1342   * @throws IOException If an I/O error occurred
1343   * 
1344   * Exceptions applicable to file systems accessed over RPC:
1345   * @throws RpcClientException If an exception occurred in the RPC client
1346   * @throws RpcServerException If an exception occurred in the RPC server
1347   * @throws UnexpectedServerException If server implementation throws 
1348   *           undeclared exception to RPC server
1349   */
1350  public FsStatus getFsStatus(final Path f) throws AccessControlException,
1351      FileNotFoundException, UnsupportedFileSystemException, IOException {
1352    if (f == null) {
1353      return defaultFS.getFsStatus();
1354    }
1355    final Path absF = fixRelativePart(f);
1356    return new FSLinkResolver<FsStatus>() {
1357      @Override
1358      public FsStatus next(final AbstractFileSystem fs, final Path p) 
1359        throws IOException, UnresolvedLinkException {
1360        return fs.getFsStatus(p);
1361      }
1362    }.resolve(this, absF);
1363  }
1364
1365  /**
1366   * Creates a symbolic link to an existing file. An exception is thrown if 
1367   * the symlink exits, the user does not have permission to create symlink,
1368   * or the underlying file system does not support symlinks.
1369   * 
1370   * Symlink permissions are ignored, access to a symlink is determined by
1371   * the permissions of the symlink target.
1372   * 
1373   * Symlinks in paths leading up to the final path component are resolved 
1374   * transparently. If the final path component refers to a symlink some 
1375   * functions operate on the symlink itself, these are:
1376   * - delete(f) and deleteOnExit(f) - Deletes the symlink.
1377   * - rename(src, dst) - If src refers to a symlink, the symlink is 
1378   *   renamed. If dst refers to a symlink, the symlink is over-written.
1379   * - getLinkTarget(f) - Returns the target of the symlink. 
1380   * - getFileLinkStatus(f) - Returns a FileStatus object describing
1381   *   the symlink.
1382   * Some functions, create() and mkdir(), expect the final path component
1383   * does not exist. If they are given a path that refers to a symlink that 
1384   * does exist they behave as if the path referred to an existing file or 
1385   * directory. All other functions fully resolve, ie follow, the symlink. 
1386   * These are: open, setReplication, setOwner, setTimes, setWorkingDirectory,
1387   * setPermission, getFileChecksum, setVerifyChecksum, getFileBlockLocations,
1388   * getFsStatus, getFileStatus, exists, and listStatus.
1389   * 
1390   * Symlink targets are stored as given to createSymlink, assuming the 
1391   * underlying file system is capable of storing a fully qualified URI.
1392   * Dangling symlinks are permitted. FileContext supports four types of 
1393   * symlink targets, and resolves them as follows
1394   * <pre>
1395   * Given a path referring to a symlink of form:
1396   * 
1397   *   <---X---> 
1398   *   fs://host/A/B/link 
1399   *   <-----Y----->
1400   * 
1401   * In this path X is the scheme and authority that identify the file system,
1402   * and Y is the path leading up to the final path component "link". If Y is
1403   * a symlink  itself then let Y' be the target of Y and X' be the scheme and
1404   * authority of Y'. Symlink targets may:
1405   * 
1406   * 1. Fully qualified URIs
1407   * 
1408   * fs://hostX/A/B/file  Resolved according to the target file system.
1409   * 
1410   * 2. Partially qualified URIs (eg scheme but no host)
1411   * 
1412   * fs:///A/B/file  Resolved according to the target file system. Eg resolving
1413   *                 a symlink to hdfs:///A results in an exception because
1414   *                 HDFS URIs must be fully qualified, while a symlink to 
1415   *                 file:///A will not since Hadoop's local file systems 
1416   *                 require partially qualified URIs.
1417   * 
1418   * 3. Relative paths
1419   * 
1420   * path  Resolves to [Y'][path]. Eg if Y resolves to hdfs://host/A and path 
1421   *       is "../B/file" then [Y'][path] is hdfs://host/B/file
1422   * 
1423   * 4. Absolute paths
1424   * 
1425   * path  Resolves to [X'][path]. Eg if Y resolves hdfs://host/A/B and path
1426   *       is "/file" then [X][path] is hdfs://host/file
1427   * </pre>
1428   * 
1429   * @param target the target of the symbolic link
1430   * @param link the path to be created that points to target
1431   * @param createParent if true then missing parent dirs are created if 
1432   *                     false then parent must exist
1433   *
1434   *
1435   * @throws AccessControlException If access is denied
1436   * @throws FileAlreadyExistsException If file <code>linkcode> already exists
1437   * @throws FileNotFoundException If <code>target</code> does not exist
1438   * @throws ParentNotDirectoryException If parent of <code>link</code> is not a
1439   *           directory.
1440   * @throws UnsupportedFileSystemException If file system for 
1441   *           <code>target</code> or <code>link</code> is not supported
1442   * @throws IOException If an I/O error occurred
1443   */
1444  @SuppressWarnings("deprecation")
1445  public void createSymlink(final Path target, final Path link,
1446      final boolean createParent) throws AccessControlException,
1447      FileAlreadyExistsException, FileNotFoundException,
1448      ParentNotDirectoryException, UnsupportedFileSystemException, 
1449      IOException { 
1450    if (!FileSystem.areSymlinksEnabled()) {
1451      throw new UnsupportedOperationException("Symlinks not supported");
1452    }
1453    final Path nonRelLink = fixRelativePart(link);
1454    new FSLinkResolver<Void>() {
1455      @Override
1456      public Void next(final AbstractFileSystem fs, final Path p) 
1457        throws IOException, UnresolvedLinkException {
1458        fs.createSymlink(target, p, createParent);
1459        return null;
1460      }
1461    }.resolve(this, nonRelLink);
1462  }
1463  
1464  /**
1465   * List the statuses of the files/directories in the given path if the path is
1466   * a directory.
1467   * 
1468   * @param f is the path
1469   *
1470   * @return an iterator that traverses statuses of the files/directories 
1471   *         in the given path
1472   *
1473   * @throws AccessControlException If access is denied
1474   * @throws FileNotFoundException If <code>f</code> does not exist
1475   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1476   *           not supported
1477   * @throws IOException If an I/O error occurred
1478   * 
1479   * Exceptions applicable to file systems accessed over RPC:
1480   * @throws RpcClientException If an exception occurred in the RPC client
1481   * @throws RpcServerException If an exception occurred in the RPC server
1482   * @throws UnexpectedServerException If server implementation throws 
1483   *           undeclared exception to RPC server
1484   */
1485  public RemoteIterator<FileStatus> listStatus(final Path f) throws
1486      AccessControlException, FileNotFoundException,
1487      UnsupportedFileSystemException, IOException {
1488    final Path absF = fixRelativePart(f);
1489    return new FSLinkResolver<RemoteIterator<FileStatus>>() {
1490      @Override
1491      public RemoteIterator<FileStatus> next(
1492          final AbstractFileSystem fs, final Path p) 
1493        throws IOException, UnresolvedLinkException {
1494        return fs.listStatusIterator(p);
1495      }
1496    }.resolve(this, absF);
1497  }
1498
1499  /**
1500   * @return an iterator over the corrupt files under the given path
1501   * (may contain duplicates if a file has more than one corrupt block)
1502   * @throws IOException
1503   */
1504  public RemoteIterator<Path> listCorruptFileBlocks(Path path)
1505    throws IOException {
1506    final Path absF = fixRelativePart(path);
1507    return new FSLinkResolver<RemoteIterator<Path>>() {
1508      @Override
1509      public RemoteIterator<Path> next(final AbstractFileSystem fs,
1510                                       final Path p) 
1511        throws IOException, UnresolvedLinkException {
1512        return fs.listCorruptFileBlocks(p);
1513      }
1514    }.resolve(this, absF);
1515  }
1516  
1517  /**
1518   * List the statuses of the files/directories in the given path if the path is
1519   * a directory. 
1520   * Return the file's status and block locations If the path is a file.
1521   * 
1522   * If a returned status is a file, it contains the file's block locations.
1523   * 
1524   * @param f is the path
1525   *
1526   * @return an iterator that traverses statuses of the files/directories 
1527   *         in the given path
1528   * If any IO exception (for example the input directory gets deleted while
1529   * listing is being executed), next() or hasNext() of the returned iterator
1530   * may throw a RuntimeException with the io exception as the cause.
1531   *
1532   * @throws AccessControlException If access is denied
1533   * @throws FileNotFoundException If <code>f</code> does not exist
1534   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1535   *           not supported
1536   * @throws IOException If an I/O error occurred
1537   * 
1538   * Exceptions applicable to file systems accessed over RPC:
1539   * @throws RpcClientException If an exception occurred in the RPC client
1540   * @throws RpcServerException If an exception occurred in the RPC server
1541   * @throws UnexpectedServerException If server implementation throws 
1542   *           undeclared exception to RPC server
1543   */
1544  public RemoteIterator<LocatedFileStatus> listLocatedStatus(
1545      final Path f) throws
1546      AccessControlException, FileNotFoundException,
1547      UnsupportedFileSystemException, IOException {
1548    final Path absF = fixRelativePart(f);
1549    return new FSLinkResolver<RemoteIterator<LocatedFileStatus>>() {
1550      @Override
1551      public RemoteIterator<LocatedFileStatus> next(
1552          final AbstractFileSystem fs, final Path p) 
1553        throws IOException, UnresolvedLinkException {
1554        return fs.listLocatedStatus(p);
1555      }
1556    }.resolve(this, absF);
1557  }
1558
1559  /**
1560   * Mark a path to be deleted on JVM shutdown.
1561   * 
1562   * @param f the existing path to delete.
1563   *
1564   * @return  true if deleteOnExit is successful, otherwise false.
1565   *
1566   * @throws AccessControlException If access is denied
1567   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1568   *           not supported
1569   * @throws IOException If an I/O error occurred
1570   * 
1571   * Exceptions applicable to file systems accessed over RPC:
1572   * @throws RpcClientException If an exception occurred in the RPC client
1573   * @throws RpcServerException If an exception occurred in the RPC server
1574   * @throws UnexpectedServerException If server implementation throws 
1575   *           undeclared exception to RPC server
1576   */
1577  public boolean deleteOnExit(Path f) throws AccessControlException,
1578      IOException {
1579    if (!this.util().exists(f)) {
1580      return false;
1581    }
1582    synchronized (DELETE_ON_EXIT) {
1583      if (DELETE_ON_EXIT.isEmpty()) {
1584        ShutdownHookManager.get().addShutdownHook(FINALIZER, SHUTDOWN_HOOK_PRIORITY);
1585      }
1586      
1587      Set<Path> set = DELETE_ON_EXIT.get(this);
1588      if (set == null) {
1589        set = new TreeSet<Path>();
1590        DELETE_ON_EXIT.put(this, set);
1591      }
1592      set.add(f);
1593    }
1594    return true;
1595  }
1596  
1597  private final Util util;
1598  public Util util() {
1599    return util;
1600  }
1601  
1602  
1603  /**
1604   * Utility/library methods built over the basic FileContext methods.
1605   * Since this are library functions, the oprtation are not atomic
1606   * and some of them may partially complete if other threads are making
1607   * changes to the same part of the name space.
1608   */
1609  public class Util {
1610    /**
1611     * Does the file exist?
1612     * Note: Avoid using this method if you already have FileStatus in hand.
1613     * Instead reuse the FileStatus 
1614     * @param f the  file or dir to be checked
1615     *
1616     * @throws AccessControlException If access is denied
1617     * @throws IOException If an I/O error occurred
1618     * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1619     *           not supported
1620     * 
1621     * Exceptions applicable to file systems accessed over RPC:
1622     * @throws RpcClientException If an exception occurred in the RPC client
1623     * @throws RpcServerException If an exception occurred in the RPC server
1624     * @throws UnexpectedServerException If server implementation throws 
1625     *           undeclared exception to RPC server
1626     */
1627    public boolean exists(final Path f) throws AccessControlException,
1628      UnsupportedFileSystemException, IOException {
1629      try {
1630        FileStatus fs = FileContext.this.getFileStatus(f);
1631        assert fs != null;
1632        return true;
1633      } catch (FileNotFoundException e) {
1634        return false;
1635      }
1636    }
1637    
1638    /**
1639     * Return the {@link ContentSummary} of path f.
1640     * @param f path
1641     *
1642     * @return the {@link ContentSummary} of path f.
1643     *
1644     * @throws AccessControlException If access is denied
1645     * @throws FileNotFoundException If <code>f</code> does not exist
1646     * @throws UnsupportedFileSystemException If file system for 
1647     *         <code>f</code> is not supported
1648     * @throws IOException If an I/O error occurred
1649     * 
1650     * Exceptions applicable to file systems accessed over RPC:
1651     * @throws RpcClientException If an exception occurred in the RPC client
1652     * @throws RpcServerException If an exception occurred in the RPC server
1653     * @throws UnexpectedServerException If server implementation throws 
1654     *           undeclared exception to RPC server
1655     */
1656    public ContentSummary getContentSummary(Path f)
1657        throws AccessControlException, FileNotFoundException,
1658        UnsupportedFileSystemException, IOException {
1659      FileStatus status = FileContext.this.getFileStatus(f);
1660      if (status.isFile()) {
1661        long length = status.getLen();
1662        return new ContentSummary.Builder().length(length).
1663            fileCount(1).directoryCount(0).spaceConsumed(length).
1664            build();
1665      }
1666      long[] summary = {0, 0, 1};
1667      RemoteIterator<FileStatus> statusIterator =
1668        FileContext.this.listStatus(f);
1669      while(statusIterator.hasNext()) {
1670        FileStatus s = statusIterator.next();
1671        long length = s.getLen();
1672        ContentSummary c = s.isDirectory() ? getContentSummary(s.getPath()) :
1673            new ContentSummary.Builder().length(length).fileCount(1).
1674            directoryCount(0).spaceConsumed(length).build();
1675        summary[0] += c.getLength();
1676        summary[1] += c.getFileCount();
1677        summary[2] += c.getDirectoryCount();
1678      }
1679      return new ContentSummary.Builder().length(summary[0]).
1680          fileCount(summary[1]).directoryCount(summary[2]).
1681          spaceConsumed(summary[0]).build();
1682    }
1683    
1684    /**
1685     * See {@link #listStatus(Path[], PathFilter)}
1686     */
1687    public FileStatus[] listStatus(Path[] files) throws AccessControlException,
1688        FileNotFoundException, IOException {
1689      return listStatus(files, DEFAULT_FILTER);
1690    }
1691     
1692    /**
1693     * Filter files/directories in the given path using the user-supplied path
1694     * filter.
1695     * 
1696     * @param f is the path name
1697     * @param filter is the user-supplied path filter
1698     *
1699     * @return an array of FileStatus objects for the files under the given path
1700     *         after applying the filter
1701     *
1702     * @throws AccessControlException If access is denied
1703     * @throws FileNotFoundException If <code>f</code> does not exist
1704     * @throws UnsupportedFileSystemException If file system for 
1705     *         <code>pathPattern</code> is not supported
1706     * @throws IOException If an I/O error occurred
1707     * 
1708     * Exceptions applicable to file systems accessed over RPC:
1709     * @throws RpcClientException If an exception occurred in the RPC client
1710     * @throws RpcServerException If an exception occurred in the RPC server
1711     * @throws UnexpectedServerException If server implementation throws 
1712     *           undeclared exception to RPC server
1713     */
1714    public FileStatus[] listStatus(Path f, PathFilter filter)
1715        throws AccessControlException, FileNotFoundException,
1716        UnsupportedFileSystemException, IOException {
1717      ArrayList<FileStatus> results = new ArrayList<FileStatus>();
1718      listStatus(results, f, filter);
1719      return results.toArray(new FileStatus[results.size()]);
1720    }
1721    
1722    /**
1723     * Filter files/directories in the given list of paths using user-supplied
1724     * path filter.
1725     * 
1726     * @param files is a list of paths
1727     * @param filter is the filter
1728     *
1729     * @return a list of statuses for the files under the given paths after
1730     *         applying the filter
1731     *
1732     * @throws AccessControlException If access is denied
1733     * @throws FileNotFoundException If a file in <code>files</code> does not 
1734     *           exist
1735     * @throws IOException If an I/O error occurred
1736     * 
1737     * Exceptions applicable to file systems accessed over RPC:
1738     * @throws RpcClientException If an exception occurred in the RPC client
1739     * @throws RpcServerException If an exception occurred in the RPC server
1740     * @throws UnexpectedServerException If server implementation throws 
1741     *           undeclared exception to RPC server
1742     */
1743    public FileStatus[] listStatus(Path[] files, PathFilter filter)
1744        throws AccessControlException, FileNotFoundException, IOException {
1745      ArrayList<FileStatus> results = new ArrayList<FileStatus>();
1746      for (int i = 0; i < files.length; i++) {
1747        listStatus(results, files[i], filter);
1748      }
1749      return results.toArray(new FileStatus[results.size()]);
1750    }
1751  
1752    /*
1753     * Filter files/directories in the given path using the user-supplied path
1754     * filter. Results are added to the given array <code>results</code>.
1755     */
1756    private void listStatus(ArrayList<FileStatus> results, Path f,
1757        PathFilter filter) throws AccessControlException,
1758        FileNotFoundException, IOException {
1759      FileStatus[] listing = listStatus(f);
1760      if (listing != null) {
1761        for (int i = 0; i < listing.length; i++) {
1762          if (filter.accept(listing[i].getPath())) {
1763            results.add(listing[i]);
1764          }
1765        }
1766      }
1767    }
1768
1769    /**
1770     * List the statuses of the files/directories in the given path 
1771     * if the path is a directory.
1772     * 
1773     * @param f is the path
1774     *
1775     * @return an array that contains statuses of the files/directories 
1776     *         in the given path
1777     *
1778     * @throws AccessControlException If access is denied
1779     * @throws FileNotFoundException If <code>f</code> does not exist
1780     * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1781     *           not supported
1782     * @throws IOException If an I/O error occurred
1783     * 
1784     * Exceptions applicable to file systems accessed over RPC:
1785     * @throws RpcClientException If an exception occurred in the RPC client
1786     * @throws RpcServerException If an exception occurred in the RPC server
1787     * @throws UnexpectedServerException If server implementation throws 
1788     *           undeclared exception to RPC server
1789     */
1790    public FileStatus[] listStatus(final Path f) throws AccessControlException,
1791        FileNotFoundException, UnsupportedFileSystemException,
1792        IOException {
1793      final Path absF = fixRelativePart(f);
1794      return new FSLinkResolver<FileStatus[]>() {
1795        @Override
1796        public FileStatus[] next(final AbstractFileSystem fs, final Path p) 
1797          throws IOException, UnresolvedLinkException {
1798          return fs.listStatus(p);
1799        }
1800      }.resolve(FileContext.this, absF);
1801    }
1802
1803    /**
1804     * List the statuses and block locations of the files in the given path.
1805     * 
1806     * If the path is a directory, 
1807     *   if recursive is false, returns files in the directory;
1808     *   if recursive is true, return files in the subtree rooted at the path.
1809     *   The subtree is traversed in the depth-first order.
1810     * If the path is a file, return the file's status and block locations.
1811     * Files across symbolic links are also returned.
1812     * 
1813     * @param f is the path
1814     * @param recursive if the subdirectories need to be traversed recursively
1815     *
1816     * @return an iterator that traverses statuses of the files
1817     * If any IO exception (for example a sub-directory gets deleted while
1818     * listing is being executed), next() or hasNext() of the returned iterator
1819     * may throw a RuntimeException with the IO exception as the cause.
1820     *
1821     * @throws AccessControlException If access is denied
1822     * @throws FileNotFoundException If <code>f</code> does not exist
1823     * @throws UnsupportedFileSystemException If file system for <code>f</code>
1824     *         is not supported
1825     * @throws IOException If an I/O error occurred
1826     * 
1827     * Exceptions applicable to file systems accessed over RPC:
1828     * @throws RpcClientException If an exception occurred in the RPC client
1829     * @throws RpcServerException If an exception occurred in the RPC server
1830     * @throws UnexpectedServerException If server implementation throws 
1831     *           undeclared exception to RPC server
1832     */
1833    public RemoteIterator<LocatedFileStatus> listFiles(
1834        final Path f, final boolean recursive) throws AccessControlException,
1835        FileNotFoundException, UnsupportedFileSystemException, 
1836        IOException {
1837      return new RemoteIterator<LocatedFileStatus>() {
1838        private Stack<RemoteIterator<LocatedFileStatus>> itors = 
1839          new Stack<RemoteIterator<LocatedFileStatus>>();
1840        RemoteIterator<LocatedFileStatus> curItor = listLocatedStatus(f);
1841        LocatedFileStatus curFile;
1842
1843        /**
1844         * Returns <tt>true</tt> if the iterator has more files.
1845         *
1846         * @return <tt>true</tt> if the iterator has more files.
1847         * @throws AccessControlException if not allowed to access next
1848         *                                file's status or locations
1849         * @throws FileNotFoundException if next file does not exist any more
1850         * @throws UnsupportedFileSystemException if next file's 
1851         *                                        fs is unsupported
1852         * @throws IOException for all other IO errors
1853         *                     for example, NameNode is not avaialbe or
1854         *                     NameNode throws IOException due to an error
1855         *                     while getting the status or block locations
1856         */
1857        @Override
1858        public boolean hasNext() throws IOException {
1859          while (curFile == null) {
1860            if (curItor.hasNext()) {
1861              handleFileStat(curItor.next());
1862            } else if (!itors.empty()) {
1863              curItor = itors.pop();
1864            } else {
1865              return false;
1866            }
1867          }
1868          return true;
1869        }
1870
1871        /**
1872         * Process the input stat.
1873         * If it is a file, return the file stat.
1874         * If it is a directory, traverse the directory if recursive is true;
1875         * ignore it if recursive is false.
1876         * If it is a symlink, resolve the symlink first and then process it
1877         * depending on if it is a file or directory.
1878         * @param stat input status
1879         * @throws AccessControlException if access is denied
1880         * @throws FileNotFoundException if file is not found
1881         * @throws UnsupportedFileSystemException if fs is not supported
1882         * @throws IOException for all other IO errors
1883         */
1884        private void handleFileStat(LocatedFileStatus stat)
1885        throws IOException {
1886          if (stat.isFile()) { // file
1887            curFile = stat;
1888          } else if (stat.isSymlink()) { // symbolic link
1889            // resolve symbolic link
1890            FileStatus symstat = FileContext.this.getFileStatus(
1891                stat.getSymlink());
1892            if (symstat.isFile() || (recursive && symstat.isDirectory())) {
1893              itors.push(curItor);
1894              curItor = listLocatedStatus(stat.getPath());
1895            }
1896          } else if (recursive) { // directory
1897            itors.push(curItor);
1898            curItor = listLocatedStatus(stat.getPath());
1899          }
1900        }
1901
1902        /**
1903         * Returns the next file's status with its block locations
1904         *
1905         * @throws AccessControlException if not allowed to access next
1906         *                                file's status or locations
1907         * @throws FileNotFoundException if next file does not exist any more
1908         * @throws UnsupportedFileSystemException if next file's 
1909         *                                        fs is unsupported
1910         * @throws IOException for all other IO errors
1911         *                     for example, NameNode is not avaialbe or
1912         *                     NameNode throws IOException due to an error
1913         *                     while getting the status or block locations
1914         */
1915        @Override
1916        public LocatedFileStatus next() throws IOException {
1917          if (hasNext()) {
1918            LocatedFileStatus result = curFile;
1919            curFile = null;
1920            return result;
1921          } 
1922          throw new java.util.NoSuchElementException("No more entry in " + f);
1923        }
1924      };
1925    }
1926
1927    /**
1928     * <p>Return all the files that match filePattern and are not checksum
1929     * files. Results are sorted by their names.
1930     * 
1931     * <p>
1932     * A filename pattern is composed of <i>regular</i> characters and
1933     * <i>special pattern matching</i> characters, which are:
1934     *
1935     * <dl>
1936     *  <dd>
1937     *   <dl>
1938     *    <p>
1939     *    <dt> <tt> ? </tt>
1940     *    <dd> Matches any single character.
1941     *
1942     *    <p>
1943     *    <dt> <tt> * </tt>
1944     *    <dd> Matches zero or more characters.
1945     *
1946     *    <p>
1947     *    <dt> <tt> [<i>abc</i>] </tt>
1948     *    <dd> Matches a single character from character set
1949     *     <tt>{<i>a,b,c</i>}</tt>.
1950     *
1951     *    <p>
1952     *    <dt> <tt> [<i>a</i>-<i>b</i>] </tt>
1953     *    <dd> Matches a single character from the character range
1954     *     <tt>{<i>a...b</i>}</tt>. Note: character <tt><i>a</i></tt> must be
1955     *     lexicographically less than or equal to character <tt><i>b</i></tt>.
1956     *
1957     *    <p>
1958     *    <dt> <tt> [^<i>a</i>] </tt>
1959     *    <dd> Matches a single char that is not from character set or range
1960     *     <tt>{<i>a</i>}</tt>.  Note that the <tt>^</tt> character must occur
1961     *     immediately to the right of the opening bracket.
1962     *
1963     *    <p>
1964     *    <dt> <tt> \<i>c</i> </tt>
1965     *    <dd> Removes (escapes) any special meaning of character <i>c</i>.
1966     *
1967     *    <p>
1968     *    <dt> <tt> {ab,cd} </tt>
1969     *    <dd> Matches a string from the string set <tt>{<i>ab, cd</i>} </tt>
1970     *    
1971     *    <p>
1972     *    <dt> <tt> {ab,c{de,fh}} </tt>
1973     *    <dd> Matches a string from string set <tt>{<i>ab, cde, cfh</i>}</tt>
1974     *
1975     *   </dl>
1976     *  </dd>
1977     * </dl>
1978     *
1979     * @param pathPattern a regular expression specifying a pth pattern
1980     *
1981     * @return an array of paths that match the path pattern
1982     *
1983     * @throws AccessControlException If access is denied
1984     * @throws UnsupportedFileSystemException If file system for 
1985     *         <code>pathPattern</code> is not supported
1986     * @throws IOException If an I/O error occurred
1987     * 
1988     * Exceptions applicable to file systems accessed over RPC:
1989     * @throws RpcClientException If an exception occurred in the RPC client
1990     * @throws RpcServerException If an exception occurred in the RPC server
1991     * @throws UnexpectedServerException If server implementation throws 
1992     *           undeclared exception to RPC server
1993     */
1994    public FileStatus[] globStatus(Path pathPattern)
1995        throws AccessControlException, UnsupportedFileSystemException,
1996        IOException {
1997      return new Globber(FileContext.this, pathPattern, DEFAULT_FILTER).glob();
1998    }
1999    
2000    /**
2001     * Return an array of FileStatus objects whose path names match pathPattern
2002     * and is accepted by the user-supplied path filter. Results are sorted by
2003     * their path names.
2004     * Return null if pathPattern has no glob and the path does not exist.
2005     * Return an empty array if pathPattern has a glob and no path matches it. 
2006     * 
2007     * @param pathPattern regular expression specifying the path pattern
2008     * @param filter user-supplied path filter
2009     *
2010     * @return an array of FileStatus objects
2011     *
2012     * @throws AccessControlException If access is denied
2013     * @throws UnsupportedFileSystemException If file system for 
2014     *         <code>pathPattern</code> is not supported
2015     * @throws IOException If an I/O error occurred
2016     * 
2017     * Exceptions applicable to file systems accessed over RPC:
2018     * @throws RpcClientException If an exception occurred in the RPC client
2019     * @throws RpcServerException If an exception occurred in the RPC server
2020     * @throws UnexpectedServerException If server implementation throws 
2021     *           undeclared exception to RPC server
2022     */
2023    public FileStatus[] globStatus(final Path pathPattern,
2024        final PathFilter filter) throws AccessControlException,
2025        UnsupportedFileSystemException, IOException {
2026      return new Globber(FileContext.this, pathPattern, filter).glob();
2027    }
2028
2029    /**
2030     * Copy file from src to dest. See
2031     * {@link #copy(Path, Path, boolean, boolean)}
2032     */
2033    public boolean copy(final Path src, final Path dst)
2034        throws AccessControlException, FileAlreadyExistsException,
2035        FileNotFoundException, ParentNotDirectoryException,
2036        UnsupportedFileSystemException, IOException {
2037      return copy(src, dst, false, false);
2038    }
2039    
2040    /**
2041     * Copy from src to dst, optionally deleting src and overwriting dst.
2042     * @param src
2043     * @param dst
2044     * @param deleteSource - delete src if true
2045     * @param overwrite  overwrite dst if true; throw IOException if dst exists
2046     *         and overwrite is false.
2047     *
2048     * @return true if copy is successful
2049     *
2050     * @throws AccessControlException If access is denied
2051     * @throws FileAlreadyExistsException If <code>dst</code> already exists
2052     * @throws FileNotFoundException If <code>src</code> does not exist
2053     * @throws ParentNotDirectoryException If parent of <code>dst</code> is not
2054     *           a directory
2055     * @throws UnsupportedFileSystemException If file system for 
2056     *         <code>src</code> or <code>dst</code> is not supported
2057     * @throws IOException If an I/O error occurred
2058     * 
2059     * Exceptions applicable to file systems accessed over RPC:
2060     * @throws RpcClientException If an exception occurred in the RPC client
2061     * @throws RpcServerException If an exception occurred in the RPC server
2062     * @throws UnexpectedServerException If server implementation throws 
2063     *           undeclared exception to RPC server
2064     * 
2065     * RuntimeExceptions:
2066     * @throws InvalidPathException If path <code>dst</code> is invalid
2067     */
2068    public boolean copy(final Path src, final Path dst, boolean deleteSource,
2069        boolean overwrite) throws AccessControlException,
2070        FileAlreadyExistsException, FileNotFoundException,
2071        ParentNotDirectoryException, UnsupportedFileSystemException, 
2072        IOException {
2073      src.checkNotSchemeWithRelative();
2074      dst.checkNotSchemeWithRelative();
2075      Path qSrc = makeQualified(src);
2076      Path qDst = makeQualified(dst);
2077      checkDest(qSrc.getName(), qDst, overwrite);
2078      FileStatus fs = FileContext.this.getFileStatus(qSrc);
2079      if (fs.isDirectory()) {
2080        checkDependencies(qSrc, qDst);
2081        mkdir(qDst, FsPermission.getDirDefault(), true);
2082        FileStatus[] contents = listStatus(qSrc);
2083        for (FileStatus content : contents) {
2084          copy(makeQualified(content.getPath()), makeQualified(new Path(qDst,
2085              content.getPath().getName())), deleteSource, overwrite);
2086        }
2087      } else {
2088        InputStream in=null;
2089        OutputStream out = null;
2090        try {
2091          in = open(qSrc);
2092          EnumSet<CreateFlag> createFlag = overwrite ? EnumSet.of(
2093              CreateFlag.CREATE, CreateFlag.OVERWRITE) : 
2094                EnumSet.of(CreateFlag.CREATE);
2095          out = create(qDst, createFlag);
2096          IOUtils.copyBytes(in, out, conf, true);
2097        } finally {
2098          IOUtils.closeStream(out);
2099          IOUtils.closeStream(in);
2100        }
2101      }
2102      if (deleteSource) {
2103        return delete(qSrc, true);
2104      } else {
2105        return true;
2106      }
2107    }
2108  }
2109
2110  /**
2111   * Check if copying srcName to dst would overwrite an existing 
2112   * file or directory.
2113   * @param srcName File or directory to be copied.
2114   * @param dst Destination to copy srcName to.
2115   * @param overwrite Whether it's ok to overwrite an existing file. 
2116   * @throws AccessControlException If access is denied.
2117   * @throws IOException If dst is an existing directory, or dst is an 
2118   * existing file and the overwrite option is not passed.
2119   */
2120  private void checkDest(String srcName, Path dst, boolean overwrite)
2121      throws AccessControlException, IOException {
2122    try {
2123      FileStatus dstFs = getFileStatus(dst);
2124      if (dstFs.isDirectory()) {
2125        if (null == srcName) {
2126          throw new IOException("Target " + dst + " is a directory");
2127        }
2128        // Recurse to check if dst/srcName exists.
2129        checkDest(null, new Path(dst, srcName), overwrite);
2130      } else if (!overwrite) {
2131        throw new IOException("Target " + new Path(dst, srcName)
2132            + " already exists");
2133      }
2134    } catch (FileNotFoundException e) {
2135      // dst does not exist - OK to copy.
2136    }
2137  }
2138   
2139  //
2140  // If the destination is a subdirectory of the source, then
2141  // generate exception
2142  //
2143  private static void checkDependencies(Path qualSrc, Path qualDst)
2144    throws IOException {
2145    if (isSameFS(qualSrc, qualDst)) {
2146      String srcq = qualSrc.toString() + Path.SEPARATOR;
2147      String dstq = qualDst.toString() + Path.SEPARATOR;
2148      if (dstq.startsWith(srcq)) {
2149        if (srcq.length() == dstq.length()) {
2150          throw new IOException("Cannot copy " + qualSrc + " to itself.");
2151        } else {
2152          throw new IOException("Cannot copy " + qualSrc +
2153                             " to its subdirectory " + qualDst);
2154        }
2155      }
2156    }
2157  }
2158  
2159  /**
2160   * Are qualSrc and qualDst of the same file system?
2161   * @param qualPath1 - fully qualified path
2162   * @param qualPath2 - fully qualified path
2163   * @return
2164   */
2165  private static boolean isSameFS(Path qualPath1, Path qualPath2) {
2166    URI srcUri = qualPath1.toUri();
2167    URI dstUri = qualPath2.toUri();
2168    return (srcUri.getScheme().equals(dstUri.getScheme()) && 
2169        !(srcUri.getAuthority() != null && dstUri.getAuthority() != null && srcUri
2170        .getAuthority().equals(dstUri.getAuthority())));
2171  }
2172
2173  /**
2174   * Deletes all the paths in deleteOnExit on JVM shutdown.
2175   */
2176  static class FileContextFinalizer implements Runnable {
2177    @Override
2178    public synchronized void run() {
2179      processDeleteOnExit();
2180    }
2181  }
2182
2183  /**
2184   * Resolves all symbolic links in the specified path.
2185   * Returns the new path object.
2186   */
2187  protected Path resolve(final Path f) throws FileNotFoundException,
2188      UnresolvedLinkException, AccessControlException, IOException {
2189    return new FSLinkResolver<Path>() {
2190      @Override
2191      public Path next(final AbstractFileSystem fs, final Path p) 
2192        throws IOException, UnresolvedLinkException {
2193        return fs.resolvePath(p);
2194      }
2195    }.resolve(this, f);
2196  }
2197
2198  /**
2199   * Resolves all symbolic links in the specified path leading up 
2200   * to, but not including the final path component.
2201   * @param f path to resolve
2202   * @return the new path object.
2203   */
2204  protected Path resolveIntermediate(final Path f) throws IOException {
2205    return new FSLinkResolver<FileStatus>() {
2206      @Override
2207      public FileStatus next(final AbstractFileSystem fs, final Path p) 
2208        throws IOException, UnresolvedLinkException {
2209        return fs.getFileLinkStatus(p);
2210      }
2211    }.resolve(this, f).getPath();
2212  }
2213
2214  /**
2215   * Returns the list of AbstractFileSystems accessed in the path. The list may
2216   * contain more than one AbstractFileSystems objects in case of symlinks.
2217   * 
2218   * @param f
2219   *          Path which needs to be resolved
2220   * @return List of AbstractFileSystems accessed in the path
2221   * @throws IOException
2222   */
2223  Set<AbstractFileSystem> resolveAbstractFileSystems(final Path f)
2224      throws IOException {
2225    final Path absF = fixRelativePart(f);
2226    final HashSet<AbstractFileSystem> result 
2227      = new HashSet<AbstractFileSystem>();
2228    new FSLinkResolver<Void>() {
2229      @Override
2230      public Void next(final AbstractFileSystem fs, final Path p)
2231          throws IOException, UnresolvedLinkException {
2232        result.add(fs);
2233        fs.getFileStatus(p);
2234        return null;
2235      }
2236    }.resolve(this, absF);
2237    return result;
2238  }
2239
2240  /**
2241   * Get the statistics for a particular file system
2242   * 
2243   * @param uri
2244   *          the uri to lookup the statistics. Only scheme and authority part
2245   *          of the uri are used as the key to store and lookup.
2246   * @return a statistics object
2247   */
2248  public static Statistics getStatistics(URI uri) {
2249    return AbstractFileSystem.getStatistics(uri);
2250  }
2251
2252  /**
2253   * Clears all the statistics stored in AbstractFileSystem, for all the file
2254   * systems.
2255   */
2256  public static void clearStatistics() {
2257    AbstractFileSystem.clearStatistics();
2258  }
2259
2260  /**
2261   * Prints the statistics to standard output. File System is identified by the
2262   * scheme and authority.
2263   */
2264  public static void printStatistics() {
2265    AbstractFileSystem.printStatistics();
2266  }
2267
2268  /**
2269   * @return Map of uri and statistics for each filesystem instantiated. The uri
2270   *         consists of scheme and authority for the filesystem.
2271   */
2272  public static Map<URI, Statistics> getAllStatistics() {
2273    return AbstractFileSystem.getAllStatistics();
2274  }
2275  
2276  /**
2277   * Get delegation tokens for the file systems accessed for a given
2278   * path.
2279   * @param p Path for which delegations tokens are requested.
2280   * @param renewer the account name that is allowed to renew the token.
2281   * @return List of delegation tokens.
2282   * @throws IOException
2283   */
2284  @InterfaceAudience.LimitedPrivate( { "HDFS", "MapReduce" })
2285  public List<Token<?>> getDelegationTokens(
2286      Path p, String renewer) throws IOException {
2287    Set<AbstractFileSystem> afsSet = resolveAbstractFileSystems(p);
2288    List<Token<?>> tokenList = 
2289        new ArrayList<Token<?>>();
2290    for (AbstractFileSystem afs : afsSet) {
2291      List<Token<?>> afsTokens = afs.getDelegationTokens(renewer);
2292      tokenList.addAll(afsTokens);
2293    }
2294    return tokenList;
2295  }
2296
2297  /**
2298   * Modifies ACL entries of files and directories.  This method can add new ACL
2299   * entries or modify the permissions on existing ACL entries.  All existing
2300   * ACL entries that are not specified in this call are retained without
2301   * changes.  (Modifications are merged into the current ACL.)
2302   *
2303   * @param path Path to modify
2304   * @param aclSpec List<AclEntry> describing modifications
2305   * @throws IOException if an ACL could not be modified
2306   */
2307  public void modifyAclEntries(final Path path, final List<AclEntry> aclSpec)
2308      throws IOException {
2309    Path absF = fixRelativePart(path);
2310    new FSLinkResolver<Void>() {
2311      @Override
2312      public Void next(final AbstractFileSystem fs, final Path p)
2313          throws IOException {
2314        fs.modifyAclEntries(p, aclSpec);
2315        return null;
2316      }
2317    }.resolve(this, absF);
2318  }
2319
2320  /**
2321   * Removes ACL entries from files and directories.  Other ACL entries are
2322   * retained.
2323   *
2324   * @param path Path to modify
2325   * @param aclSpec List<AclEntry> describing entries to remove
2326   * @throws IOException if an ACL could not be modified
2327   */
2328  public void removeAclEntries(final Path path, final List<AclEntry> aclSpec)
2329      throws IOException {
2330    Path absF = fixRelativePart(path);
2331    new FSLinkResolver<Void>() {
2332      @Override
2333      public Void next(final AbstractFileSystem fs, final Path p)
2334          throws IOException {
2335        fs.removeAclEntries(p, aclSpec);
2336        return null;
2337      }
2338    }.resolve(this, absF);
2339  }
2340
2341  /**
2342   * Removes all default ACL entries from files and directories.
2343   *
2344   * @param path Path to modify
2345   * @throws IOException if an ACL could not be modified
2346   */
2347  public void removeDefaultAcl(Path path)
2348      throws IOException {
2349    Path absF = fixRelativePart(path);
2350    new FSLinkResolver<Void>() {
2351      @Override
2352      public Void next(final AbstractFileSystem fs, final Path p)
2353          throws IOException {
2354        fs.removeDefaultAcl(p);
2355        return null;
2356      }
2357    }.resolve(this, absF);
2358  }
2359
2360  /**
2361   * Removes all but the base ACL entries of files and directories.  The entries
2362   * for user, group, and others are retained for compatibility with permission
2363   * bits.
2364   *
2365   * @param path Path to modify
2366   * @throws IOException if an ACL could not be removed
2367   */
2368  public void removeAcl(Path path) throws IOException {
2369    Path absF = fixRelativePart(path);
2370    new FSLinkResolver<Void>() {
2371      @Override
2372      public Void next(final AbstractFileSystem fs, final Path p)
2373          throws IOException {
2374        fs.removeAcl(p);
2375        return null;
2376      }
2377    }.resolve(this, absF);
2378  }
2379
2380  /**
2381   * Fully replaces ACL of files and directories, discarding all existing
2382   * entries.
2383   *
2384   * @param path Path to modify
2385   * @param aclSpec List<AclEntry> describing modifications, must include entries
2386   *   for user, group, and others for compatibility with permission bits.
2387   * @throws IOException if an ACL could not be modified
2388   */
2389  public void setAcl(Path path, final List<AclEntry> aclSpec)
2390      throws IOException {
2391    Path absF = fixRelativePart(path);
2392    new FSLinkResolver<Void>() {
2393      @Override
2394      public Void next(final AbstractFileSystem fs, final Path p)
2395          throws IOException {
2396        fs.setAcl(p, aclSpec);
2397        return null;
2398      }
2399    }.resolve(this, absF);
2400  }
2401
2402  /**
2403   * Gets the ACLs of files and directories.
2404   *
2405   * @param path Path to get
2406   * @return RemoteIterator<AclStatus> which returns each AclStatus
2407   * @throws IOException if an ACL could not be read
2408   */
2409  public AclStatus getAclStatus(Path path) throws IOException {
2410    Path absF = fixRelativePart(path);
2411    return new FSLinkResolver<AclStatus>() {
2412      @Override
2413      public AclStatus next(final AbstractFileSystem fs, final Path p)
2414          throws IOException {
2415        return fs.getAclStatus(p);
2416      }
2417    }.resolve(this, absF);
2418  }
2419
2420  /**
2421   * Set an xattr of a file or directory.
2422   * The name must be prefixed with the namespace followed by ".". For example,
2423   * "user.attr".
2424   * <p/>
2425   * Refer to the HDFS extended attributes user documentation for details.
2426   *
2427   * @param path Path to modify
2428   * @param name xattr name.
2429   * @param value xattr value.
2430   * @throws IOException
2431   */
2432  public void setXAttr(Path path, String name, byte[] value)
2433      throws IOException {
2434    setXAttr(path, name, value, EnumSet.of(XAttrSetFlag.CREATE,
2435        XAttrSetFlag.REPLACE));
2436  }
2437
2438  /**
2439   * Set an xattr of a file or directory.
2440   * The name must be prefixed with the namespace followed by ".". For example,
2441   * "user.attr".
2442   * <p/>
2443   * Refer to the HDFS extended attributes user documentation for details.
2444   *
2445   * @param path Path to modify
2446   * @param name xattr name.
2447   * @param value xattr value.
2448   * @param flag xattr set flag
2449   * @throws IOException
2450   */
2451  public void setXAttr(Path path, final String name, final byte[] value,
2452      final EnumSet<XAttrSetFlag> flag) throws IOException {
2453    final Path absF = fixRelativePart(path);
2454    new FSLinkResolver<Void>() {
2455      @Override
2456      public Void next(final AbstractFileSystem fs, final Path p)
2457          throws IOException {
2458        fs.setXAttr(p, name, value, flag);
2459        return null;
2460      }
2461    }.resolve(this, absF);
2462  }
2463
2464  /**
2465   * Get an xattr for a file or directory.
2466   * The name must be prefixed with the namespace followed by ".". For example,
2467   * "user.attr".
2468   * <p/>
2469   * Refer to the HDFS extended attributes user documentation for details.
2470   *
2471   * @param path Path to get extended attribute
2472   * @param name xattr name.
2473   * @return byte[] xattr value.
2474   * @throws IOException
2475   */
2476  public byte[] getXAttr(Path path, final String name) throws IOException {
2477    final Path absF = fixRelativePart(path);
2478    return new FSLinkResolver<byte[]>() {
2479      @Override
2480      public byte[] next(final AbstractFileSystem fs, final Path p)
2481          throws IOException {
2482        return fs.getXAttr(p, name);
2483      }
2484    }.resolve(this, absF);
2485  }
2486
2487  /**
2488   * Get all of the xattrs for a file or directory.
2489   * Only those xattrs for which the logged-in user has permissions to view
2490   * are returned.
2491   * <p/>
2492   * Refer to the HDFS extended attributes user documentation for details.
2493   *
2494   * @param path Path to get extended attributes
2495   * @return Map<String, byte[]> describing the XAttrs of the file or directory
2496   * @throws IOException
2497   */
2498  public Map<String, byte[]> getXAttrs(Path path) throws IOException {
2499    final Path absF = fixRelativePart(path);
2500    return new FSLinkResolver<Map<String, byte[]>>() {
2501      @Override
2502      public Map<String, byte[]> next(final AbstractFileSystem fs, final Path p)
2503          throws IOException {
2504        return fs.getXAttrs(p);
2505      }
2506    }.resolve(this, absF);
2507  }
2508
2509  /**
2510   * Get all of the xattrs for a file or directory.
2511   * Only those xattrs for which the logged-in user has permissions to view
2512   * are returned.
2513   * <p/>
2514   * Refer to the HDFS extended attributes user documentation for details.
2515   *
2516   * @param path Path to get extended attributes
2517   * @param names XAttr names.
2518   * @return Map<String, byte[]> describing the XAttrs of the file or directory
2519   * @throws IOException
2520   */
2521  public Map<String, byte[]> getXAttrs(Path path, final List<String> names)
2522      throws IOException {
2523    final Path absF = fixRelativePart(path);
2524    return new FSLinkResolver<Map<String, byte[]>>() {
2525      @Override
2526      public Map<String, byte[]> next(final AbstractFileSystem fs, final Path p)
2527          throws IOException {
2528        return fs.getXAttrs(p, names);
2529      }
2530    }.resolve(this, absF);
2531  }
2532
2533  /**
2534   * Remove an xattr of a file or directory.
2535   * The name must be prefixed with the namespace followed by ".". For example,
2536   * "user.attr".
2537   * <p/>
2538   * Refer to the HDFS extended attributes user documentation for details.
2539   *
2540   * @param path Path to remove extended attribute
2541   * @param name xattr name
2542   * @throws IOException
2543   */
2544  public void removeXAttr(Path path, final String name) throws IOException {
2545    final Path absF = fixRelativePart(path);
2546    new FSLinkResolver<Void>() {
2547      @Override
2548      public Void next(final AbstractFileSystem fs, final Path p)
2549          throws IOException {
2550        fs.removeXAttr(p, name);
2551        return null;
2552      }
2553    }.resolve(this, absF);
2554  }
2555
2556  /**
2557   * Get all of the xattr names for a file or directory.
2558   * Only those xattr names which the logged-in user has permissions to view
2559   * are returned.
2560   * <p/>
2561   * Refer to the HDFS extended attributes user documentation for details.
2562   *
2563   * @param path Path to get extended attributes
2564   * @return List<String> of the XAttr names of the file or directory
2565   * @throws IOException
2566   */
2567  public List<String> listXAttrs(Path path) throws IOException {
2568    final Path absF = fixRelativePart(path);
2569    return new FSLinkResolver<List<String>>() {
2570      @Override
2571      public List<String> next(final AbstractFileSystem fs, final Path p)
2572          throws IOException {
2573        return fs.listXAttrs(p);
2574      }
2575    }.resolve(this, absF);
2576  }
2577}