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