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