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