hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From omal...@apache.org
Subject svn commit: r784965 [1/2] - in /hadoop/core/branches/HADOOP-4687/core/src: contrib/ec2/ contrib/ec2/bin/ contrib/eclipse-plugin/ java/ java/org/apache/hadoop/filecache/ java/org/apache/hadoop/fs/ java/org/apache/hadoop/fs/ftp/ java/org/apache/hadoop/fs...
Date Mon, 15 Jun 2009 20:28:08 GMT
Author: omalley
Date: Mon Jun 15 20:28:06 2009
New Revision: 784965

URL: http://svn.apache.org/viewvc?rev=784965&view=rev
Log:
Merged src/core, src/test/core, src/contrib/eclipse-plugin, and
src/contrib/ec2 from trunk 776174:784663

Added:
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/CreateFlag.java
      - copied unchanged from r784663, hadoop/core/trunk/src/core/org/apache/hadoop/fs/CreateFlag.java
Removed:
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/io/DeprecatedUTF8.java
Modified:
    hadoop/core/branches/HADOOP-4687/core/src/contrib/ec2/   (props changed)
    hadoop/core/branches/HADOOP-4687/core/src/contrib/ec2/bin/hadoop-ec2-init-remote.sh
    hadoop/core/branches/HADOOP-4687/core/src/contrib/ec2/bin/launch-hadoop-slaves
    hadoop/core/branches/HADOOP-4687/core/src/contrib/eclipse-plugin/   (props changed)
    hadoop/core/branches/HADOOP-4687/core/src/contrib/eclipse-plugin/build.xml
    hadoop/core/branches/HADOOP-4687/core/src/java/   (props changed)
    hadoop/core/branches/HADOOP-4687/core/src/java/core-default.xml
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/filecache/DistributedCache.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FileSystem.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FilterFileSystem.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FsShell.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/RawLocalFileSystem.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/ftp/FTPFileSystem.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/IFSImpl.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSImpl.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSInputStream.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSOutputStream.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KosmosFileSystem.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3/S3FileSystem.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/Jets3tNativeFileSystemStore.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/NativeFileSystemStore.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/NativeS3FileSystem.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/http/HttpServer.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/io/IOUtils.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/metrics/ContextFactory.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/net/NetUtils.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/util/ProcessTree.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/util/ProcfsBasedProcessTree.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/util/Progress.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/util/RunJar.java
    hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/util/StringUtils.java
    hadoop/core/branches/HADOOP-4687/core/src/test/core/   (props changed)
    hadoop/core/branches/HADOOP-4687/core/src/test/core/org/apache/hadoop/filecache/TestDistributedCache.java
    hadoop/core/branches/HADOOP-4687/core/src/test/core/org/apache/hadoop/fs/kfs/KFSEmulationImpl.java
    hadoop/core/branches/HADOOP-4687/core/src/test/core/org/apache/hadoop/fs/kfs/TestKosmosFileSystem.java
    hadoop/core/branches/HADOOP-4687/core/src/test/core/org/apache/hadoop/fs/s3/S3FileSystemContractBaseTest.java
    hadoop/core/branches/HADOOP-4687/core/src/test/core/org/apache/hadoop/fs/s3native/InMemoryNativeFileSystemStore.java
    hadoop/core/branches/HADOOP-4687/core/src/test/core/org/apache/hadoop/fs/s3native/NativeS3FileSystemContractBaseTest.java
    hadoop/core/branches/HADOOP-4687/core/src/test/core/org/apache/hadoop/util/TestProcfsBasedProcessTree.java

Propchange: hadoop/core/branches/HADOOP-4687/core/src/contrib/ec2/
------------------------------------------------------------------------------
--- svn:mergeinfo (added)
+++ svn:mergeinfo Mon Jun 15 20:28:06 2009
@@ -0,0 +1,2 @@
+/hadoop/core/branches/branch-0.19/core/src/contrib/ec2:713112
+/hadoop/core/trunk/src/contrib/ec2:776175-784663

Modified: hadoop/core/branches/HADOOP-4687/core/src/contrib/ec2/bin/hadoop-ec2-init-remote.sh
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/contrib/ec2/bin/hadoop-ec2-init-remote.sh?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/contrib/ec2/bin/hadoop-ec2-init-remote.sh (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/contrib/ec2/bin/hadoop-ec2-init-remote.sh Mon Jun 15 20:28:06 2009
@@ -17,7 +17,9 @@
 SECURITY_GROUPS=`wget -q -O - http://169.254.169.254/latest/meta-data/security-groups`
 IS_MASTER=`echo $SECURITY_GROUPS | awk '{ a = match ($0, "-master$"); if (a) print "true"; else print "false"; }'`
 if [ "$IS_MASTER" == "true" ]; then
- MASTER_HOST=`wget -q -O - http://169.254.169.254/latest/meta-data/local-hostname`
+ # use public hostnames for master. private hostnames can be used by substituting:
+ # MASTER_HOST=`wget -q -O - http://169.254.169.254/latest/meta-data/local-hostname`
+ MASTER_HOST=`wget -q -O - 'http://169.254.169.254/latest/meta-data/public-hostname'`
 fi
 
 HADOOP_HOME=`ls -d /usr/local/hadoop-*`
@@ -78,6 +80,12 @@
   <value>3</value>
 </property>
 
+<property>
+  <name>hadoop.rpc.socket.factory.class.default</name>
+  <value>org.apache.hadoop.net.StandardSocketFactory</value>
+  <final>true</final>
+</property>
+
 </configuration>
 EOF
 

Modified: hadoop/core/branches/HADOOP-4687/core/src/contrib/ec2/bin/launch-hadoop-slaves
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/contrib/ec2/bin/launch-hadoop-slaves?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/contrib/ec2/bin/launch-hadoop-slaves (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/contrib/ec2/bin/launch-hadoop-slaves Mon Jun 15 20:28:06 2009
@@ -42,7 +42,9 @@
 
 # Finding Hadoop image
 AMI_IMAGE=`ec2-describe-images -a | grep $S3_BUCKET | grep $HADOOP_VERSION | grep $ARCH |grep available | awk '{print $2}'`
-MASTER_HOST=`cat $MASTER_PRIVATE_IP_PATH`
+# to use private master hostname, substitute below with:
+# MASTER_HOST=`cat $MASTER_PRIVATE_IP_PATH`
+MASTER_HOST=`cat $MASTER_IP_PATH`
 MASTER_ZONE=`cat $MASTER_ZONE_PATH`
 
 # Substituting master hostname

Propchange: hadoop/core/branches/HADOOP-4687/core/src/contrib/eclipse-plugin/
------------------------------------------------------------------------------
--- svn:mergeinfo (added)
+++ svn:mergeinfo Mon Jun 15 20:28:06 2009
@@ -0,0 +1,2 @@
+/hadoop/core/branches/branch-0.19/core/src/contrib/eclipse-plugin:713112
+/hadoop/core/trunk/src/contrib/eclipse-plugin:776175-784663

Modified: hadoop/core/branches/HADOOP-4687/core/src/contrib/eclipse-plugin/build.xml
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/contrib/eclipse-plugin/build.xml?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/contrib/eclipse-plugin/build.xml (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/contrib/eclipse-plugin/build.xml Mon Jun 15 20:28:06 2009
@@ -67,7 +67,8 @@
   <target name="jar" depends="compile" unless="skip.contrib">
     <mkdir dir="${build.dir}/lib"/>
     <copy file="${hadoop.root}/build/hadoop-${version}-core.jar" tofile="${build.dir}/lib/hadoop-core.jar" verbose="true"/>
-    <copy file="${hadoop.root}/lib/commons-cli-2.0-SNAPSHOT.jar" todir="${build.dir}/lib" verbose="true"/>
+    <copy file="${hadoop.root}/build/ivy/lib/Hadoop/common/commons-cli-${commons-cli.version}.jar"  todir="${build.dir}/lib" verbose="true"/>
+    <copy file="${hadoop.root}/build/ivy/lib/Hadoop/common/commons-cli-${commons-cli2.version}.jar"  todir="${build.dir}/lib" verbose="true"/>
     <jar
       jarfile="${build.dir}/hadoop-${version}-${name}.jar"
       manifest="${root}/META-INF/MANIFEST.MF">

Propchange: hadoop/core/branches/HADOOP-4687/core/src/java/
------------------------------------------------------------------------------
--- svn:mergeinfo (added)
+++ svn:mergeinfo Mon Jun 15 20:28:06 2009
@@ -0,0 +1,2 @@
+/hadoop/core/branches/branch-0.19/core/src/java:713112
+/hadoop/core/trunk/src/core:776175-784663

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/core-default.xml
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/core-default.xml?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/core-default.xml (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/core-default.xml Mon Jun 15 20:28:06 2009
@@ -248,6 +248,23 @@
 
 
 <property>
+  <name>fs.automatic.close</name>
+  <value>true</value>
+  <description>By default, FileSystem instances are automatically closed at program
+  exit using a JVM shutdown hook. Setting this property to false disables this
+  behavior. This is an advanced option that should only be used by server applications
+  requiring a more carefully orchestrated shutdown sequence.
+  </description>
+</property>
+
+<property>
+  <name>fs.s3n.block.size</name>
+  <value>67108864</value>
+  <description>Block size to use when reading files using the native S3
+  filesystem (s3n: URIs).</description>
+</property>
+
+<property>
   <name>local.cache.size</name>
   <value>10737418240</value>
   <description>The limit on the size of cache you want to keep, set by default

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/filecache/DistributedCache.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/filecache/DistributedCache.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/filecache/DistributedCache.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/filecache/DistributedCache.java Mon Jun 15 20:28:06 2009
@@ -34,11 +34,11 @@
  * framework to cache files (text, archives, jars etc.) needed by applications.
  * </p>
  * 
- * <p>Applications specify the files, via urls (hdfs:// or http://) to be 
- * cached via the org.apache.hadoop.mapred.JobConf.
- * The <code>DistributedCache</code> assumes that the
- * files specified via hdfs:// urls are already present on the 
- * {@link FileSystem} at the path specified by the url.</p>
+ * <p>Applications specify the files, via urls (hdfs:// or http://) to be cached 
+ * via the org.apache.hadoop.mapred.JobConf. The
+ * <code>DistributedCache</code> assumes that the files specified via urls are
+ * already present on the {@link FileSystem} at the path specified by the url
+ * and are accessible by every machine in the cluster.</p>
  * 
  * <p>The framework will copy the necessary files on to the slave node before 
  * any tasks for the job are executed on that node. Its efficiency stems from 
@@ -127,9 +127,7 @@
    * previously cached (and valid) or copy it from the {@link FileSystem} now.
    * 
    * @param cache the cache to be localized, this should be specified as 
-   * new URI(hdfs://hostname:port/absolute_path_to_file#LINKNAME). If no schema 
-   * or hostname:port is provided the file is assumed to be in the filesystem
-   * being used in the Configuration
+   * new URI(scheme://scheme-specific-part/absolute_path_to_file#LINKNAME).
    * @param conf The Confguration file which contains the filesystem
    * @param baseDir The base cache Dir where you wnat to localize the files/archives
    * @param fileStatus The file status on the dfs.
@@ -160,9 +158,7 @@
    * previously cached (and valid) or copy it from the {@link FileSystem} now.
    * 
    * @param cache the cache to be localized, this should be specified as 
-   * new URI(hdfs://hostname:port/absolute_path_to_file#LINKNAME). If no schema 
-   * or hostname:port is provided the file is assumed to be in the filesystem
-   * being used in the Configuration
+   * new URI(scheme://scheme-specific-part/absolute_path_to_file#LINKNAME).
    * @param conf The Confguration file which contains the filesystem
    * @param baseDir The base cache Dir where you wnat to localize the files/archives
    * @param fileStatus The file status on the dfs.
@@ -229,9 +225,7 @@
    * previously cached (and valid) or copy it from the {@link FileSystem} now.
    * 
    * @param cache the cache to be localized, this should be specified as 
-   * new URI(hdfs://hostname:port/absolute_path_to_file#LINKNAME). If no schema 
-   * or hostname:port is provided the file is assumed to be in the filesystem
-   * being used in the Configuration
+   * new URI(scheme://scheme-specific-part/absolute_path_to_file#LINKNAME).
    * @param conf The Confguration file which contains the filesystem
    * @param baseDir The base cache Dir where you wnat to localize the files/archives
    * @param isArchive if the cache is an archive or a file. In case it is an 
@@ -348,7 +342,7 @@
     if(cache.getFragment() == null) {
     	doSymlink = false;
     }
-    FileSystem fs = getFileSystem(cache, conf);
+    FileSystem fs = FileSystem.get(cache, conf);
     String link = currentWorkDir.toString() + Path.SEPARATOR + cache.getFragment();
     File flink = new File(link);
     if (ifExistsAndFresh(conf, fs, cache, confFileStamp,
@@ -531,14 +525,6 @@
     }  
   }
   
-  private static FileSystem getFileSystem(URI cache, Configuration conf)
-    throws IOException {
-    if ("hdfs".equals(cache.getScheme()))
-      return FileSystem.get(cache, conf);
-    else
-      return FileSystem.get(conf);
-  }
-
   /**
    * Set the configuration with the given set of archives
    * @param archives The list of archives that need to be localized
@@ -695,7 +681,7 @@
     throws IOException {
     String classpath = conf.get("mapred.job.classpath.files");
     conf.set("mapred.job.classpath.files", classpath == null ? file.toString()
-             : classpath + System.getProperty("path.separator") + file.toString());
+             : classpath + "," + file.toString());
     FileSystem fs = FileSystem.get(conf);
     URI uri = fs.makeQualified(file).toUri();
 
@@ -708,14 +694,14 @@
    * @param conf Configuration that contains the classpath setting
    */
   public static Path[] getFileClassPaths(Configuration conf) {
-    String classpath = conf.get("mapred.job.classpath.files");
-    if (classpath == null)
-      return null;
-    ArrayList list = Collections.list(new StringTokenizer(classpath, System
-                                                          .getProperty("path.separator")));
+    ArrayList<String> list = (ArrayList<String>)conf.getStringCollection(
+                                "mapred.job.classpath.files");
+    if (list.size() == 0) { 
+      return null; 
+    }
     Path[] paths = new Path[list.size()];
     for (int i = 0; i < list.size(); i++) {
-      paths[i] = new Path((String) list.get(i));
+      paths[i] = new Path(list.get(i));
     }
     return paths;
   }
@@ -731,8 +717,7 @@
     throws IOException {
     String classpath = conf.get("mapred.job.classpath.archives");
     conf.set("mapred.job.classpath.archives", classpath == null ? archive
-             .toString() : classpath + System.getProperty("path.separator")
-             + archive.toString());
+             .toString() : classpath + "," + archive.toString());
     FileSystem fs = FileSystem.get(conf);
     URI uri = fs.makeQualified(archive).toUri();
 
@@ -745,14 +730,14 @@
    * @param conf Configuration that contains the classpath setting
    */
   public static Path[] getArchiveClassPaths(Configuration conf) {
-    String classpath = conf.get("mapred.job.classpath.archives");
-    if (classpath == null)
-      return null;
-    ArrayList list = Collections.list(new StringTokenizer(classpath, System
-                                                          .getProperty("path.separator")));
+    ArrayList<String> list = (ArrayList<String>)conf.getStringCollection(
+                                "mapred.job.classpath.archives");
+    if (list.size() == 0) { 
+      return null; 
+    }
     Path[] paths = new Path[list.size()];
     for (int i = 0; i < list.size(); i++) {
-      paths[i] = new Path((String) list.get(i));
+      paths[i] = new Path(list.get(i));
     }
     return paths;
   }

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FileSystem.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FileSystem.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FileSystem.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FileSystem.java Mon Jun 15 20:28:06 2009
@@ -24,7 +24,9 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.IdentityHashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -65,12 +67,12 @@
  * implementation is DistributedFileSystem.
  *****************************************************************/
 public abstract class FileSystem extends Configured implements Closeable {
-  private static final String FS_DEFAULT_NAME_KEY = "fs.default.name";
+  public static final String FS_DEFAULT_NAME_KEY = "fs.default.name";
 
   public static final Log LOG = LogFactory.getLog(FileSystem.class);
 
   /** FileSystem cache */
-  private static final Cache CACHE = new Cache();
+  static final Cache CACHE = new Cache();
 
   /** The key this instance is stored under in the cache. */
   private Cache.Key key;
@@ -224,17 +226,6 @@
     return (LocalFileSystem)newInstance(LocalFileSystem.NAME, conf);
   }
 
-  private static class ClientFinalizer extends Thread {
-    public synchronized void run() {
-      try {
-        FileSystem.closeAll();
-      } catch (IOException e) {
-        LOG.info("FileSystem.closeAll() threw an exception:\n" + e);
-      }
-    }
-  }
-  private static final ClientFinalizer clientFinalizer = new ClientFinalizer();
-
   /**
    * Close all cached filesystems. Be sure those filesystems are not
    * used anymore.
@@ -516,6 +507,7 @@
   /**
    * Opens an FSDataOutputStream at the indicated Path with write-progress
    * reporting.
+   * @deprecated Consider using {@link #create(Path, FsPermission, EnumSet, int, short, long, Progressable)} instead.
    * @param f the file name to open
    * @param permission
    * @param overwrite if a file with this name already exists, then if true,
@@ -527,13 +519,36 @@
    * @throws IOException
    * @see #setPermission(Path, FsPermission)
    */
-  public abstract FSDataOutputStream create(Path f,
+  public FSDataOutputStream create(Path f,
       FsPermission permission,
       boolean overwrite,
       int bufferSize,
       short replication,
       long blockSize,
-      Progressable progress) throws IOException;
+      Progressable progress) throws IOException{
+    return create(f, permission, overwrite ? EnumSet.of(CreateFlag.OVERWRITE)
+        : EnumSet.of(CreateFlag.CREATE), bufferSize, replication, blockSize,
+        progress);
+  }
+  
+  /**
+   * Opens an FSDataOutputStream at the indicated Path with write-progress
+   * reporting.
+   * @param f the file name to open.
+   * @param permission
+   * @param flag determines the semantic of this create.
+   * @param bufferSize the size of the buffer to be used.
+   * @param replication required block replication for the file.
+   * @param blockSize
+   * @param progress
+   * @throws IOException
+   * @see #setPermission(Path, FsPermission)
+   * @see CreateFlag
+   */
+  public abstract FSDataOutputStream create(Path f, FsPermission permission,
+      EnumSet<CreateFlag> flag, int bufferSize, short replication, long blockSize,
+      Progressable progress) throws IOException ;
+  
 
   /**
    * Creates the given Path as a brand-new zero-length file.  If
@@ -1409,7 +1424,10 @@
 
   /** Caching FileSystem objects */
   static class Cache {
+    private final ClientFinalizer clientFinalizer = new ClientFinalizer();
+
     private final Map<Key, FileSystem> map = new HashMap<Key, FileSystem>();
+    private final Set<Key> toAutoClose = new HashSet<Key>();
 
     /** A variable that makes all objects in the cache unique */
     private static AtomicLong unique = new AtomicLong(1);
@@ -1434,6 +1452,10 @@
         }
         fs.key = key;
         map.put(key, fs);
+
+        if (conf.getBoolean("fs.automatic.close", true)) {
+          toAutoClose.add(key);
+        }
       }
       return fs;
     }
@@ -1441,6 +1463,7 @@
     synchronized void remove(Key key, FileSystem fs) {
       if (map.containsKey(key) && fs == map.get(key)) {
         map.remove(key);
+        toAutoClose.remove(key);
         if (map.isEmpty() && !clientFinalizer.isAlive()) {
           if (!Runtime.getRuntime().removeShutdownHook(clientFinalizer)) {
             LOG.info("Could not cancel cleanup thread, though no " +
@@ -1451,11 +1474,27 @@
     }
 
     synchronized void closeAll() throws IOException {
+      closeAll(false);
+    }
+
+    /**
+     * Close all FileSystem instances in the Cache.
+     * @param onlyAutomatic only close those that are marked for automatic closing
+     */
+    synchronized void closeAll(boolean onlyAutomatic) throws IOException {
       List<IOException> exceptions = new ArrayList<IOException>();
-      for(; !map.isEmpty(); ) {
-        Map.Entry<Key, FileSystem> e = map.entrySet().iterator().next();
-        final Key key = e.getKey();
-        final FileSystem fs = e.getValue();
+
+      // Make a copy of the keys in the map since we'll be modifying
+      // the map while iterating over it, which isn't safe.
+      List<Key> keys = new ArrayList<Key>();
+      keys.addAll(map.keySet());
+
+      for (Key key : keys) {
+        final FileSystem fs = map.get(key);
+
+        if (onlyAutomatic && !toAutoClose.contains(key)) {
+          continue;
+        }
 
         //remove from cache
         remove(key, fs);
@@ -1475,6 +1514,16 @@
       }
     }
 
+    private class ClientFinalizer extends Thread {
+      public synchronized void run() {
+        try {
+          closeAll(true);
+        } catch (IOException e) {
+          LOG.info("FileSystem.Cache.closeAll() threw an exception:\n" + e);
+        }
+      }
+    }
+
     /** FileSystem.Cache.Key */
     static class Key {
       final String scheme;

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FilterFileSystem.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FilterFileSystem.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FilterFileSystem.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FilterFileSystem.java Mon Jun 15 20:28:06 2009
@@ -20,6 +20,7 @@
 
 import java.io.*;
 import java.net.URI;
+import java.util.EnumSet;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.permission.FsPermission;
@@ -106,10 +107,10 @@
   /** {@inheritDoc} */
   @Override
   public FSDataOutputStream create(Path f, FsPermission permission,
-      boolean overwrite, int bufferSize, short replication, long blockSize,
+      EnumSet<CreateFlag> flag, int bufferSize, short replication, long blockSize,
       Progressable progress) throws IOException {
     return fs.create(f, permission,
-        overwrite, bufferSize, replication, blockSize, progress);
+        flag, bufferSize, replication, blockSize, progress);
   }
 
   /**

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FsShell.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FsShell.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FsShell.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/FsShell.java Mon Jun 15 20:28:06 2009
@@ -61,6 +61,7 @@
   static final String COPYTOLOCAL_SHORT_USAGE = GET_SHORT_USAGE.replace(
       "-get", "-copyToLocal");
   static final String TAIL_USAGE="-tail [-f] <file>";
+  static final String DU_USAGE="-du [-s] [-h] <paths...>";
 
   /**
    */
@@ -670,58 +671,98 @@
 
   /**
    * Show the size of all files that match the file pattern <i>src</i>
-   * @param src a file pattern specifying source files
+   * @param cmd
+   * @param pos ignore anything before this pos in cmd
    * @throws IOException  
    * @see org.apache.hadoop.fs.FileSystem#globStatus(Path)
    */
-  void du(String src) throws IOException {
-    Path srcPath = new Path(src);
-    FileSystem srcFs = srcPath.getFileSystem(getConf());
-    Path[] pathItems = FileUtil.stat2Paths(srcFs.globStatus(srcPath), 
-                                           srcPath);
-    FileStatus items[] = srcFs.listStatus(pathItems);
-    if ((items == null) || ((items.length == 0) && 
-        (!srcFs.exists(srcPath)))){
-      throw new FileNotFoundException("Cannot access " + src
-            + ": No such file or directory.");
-    } else {
-      System.out.println("Found " + items.length + " items");
-      int maxLength = 10;
-      
-      long length[] = new long[items.length];
-      for (int i = 0; i < items.length; i++) {
-        length[i] = items[i].isDir() ?
-          srcFs.getContentSummary(items[i].getPath()).getLength() :
-          items[i].getLen();
-        int len = String.valueOf(length[i]).length();
-        if (len > maxLength) maxLength = len;
+  void du(String[] cmd, int pos) throws IOException {
+    CommandFormat c = new CommandFormat(
+      "du", 0, Integer.MAX_VALUE, "h", "s");
+    List<String> params;
+    try {
+      params = c.parse(cmd, pos);
+    } catch (IllegalArgumentException iae) {
+      System.err.println("Usage: java FsShell " + DU_USAGE);
+      throw iae;
+    }
+    boolean humanReadable = c.getOpt("h");
+    boolean summary = c.getOpt("s");
+
+    // Default to cwd
+    if (params.isEmpty()) {
+      params.add(".");
+    }
+
+    List<UsagePair> usages = new ArrayList<UsagePair>();
+
+    for (String src : params) {
+      Path srcPath = new Path(src);
+      FileSystem srcFs = srcPath.getFileSystem(getConf());
+      FileStatus globStatus[] = srcFs.globStatus(srcPath);
+      FileStatus statusToPrint[];
+
+      if (summary) {
+        statusToPrint = globStatus;
+      } else {
+        Path statPaths[] = FileUtil.stat2Paths(globStatus, srcPath);
+        statusToPrint = srcFs.listStatus(statPaths);
       }
-      for(int i = 0; i < items.length; i++) {
-        System.out.printf("%-"+ (maxLength + BORDER) +"d", length[i]);
-        System.out.println(items[i].getPath());
+      if ((statusToPrint == null) || ((statusToPrint.length == 0) &&
+                                      (!srcFs.exists(srcPath)))){
+        throw new FileNotFoundException("Cannot access " + src
+                                        + ": No such file or directory.");
+      }
+
+      if (!summary) {
+        System.out.println("Found " + statusToPrint.length + " items");
+      }
+
+      for (FileStatus stat : statusToPrint) {
+        long length;
+        if (summary || stat.isDir()) {
+          length = srcFs.getContentSummary(stat.getPath()).getLength();
+        } else {
+          length = stat.getLen();
+        }
+
+        usages.add(new UsagePair(String.valueOf(stat.getPath()), length));
       }
     }
+    printUsageSummary(usages, humanReadable);
   }
     
   /**
    * Show the summary disk usage of each dir/file 
    * that matches the file pattern <i>src</i>
-   * @param src a file pattern specifying source files
+   * @param cmd
+   * @param pos ignore anything before this pos in cmd
    * @throws IOException  
    * @see org.apache.hadoop.fs.FileSystem#globStatus(Path)
    */
-  void dus(String src) throws IOException {
-    Path srcPath = new Path(src);
-    FileSystem srcFs = srcPath.getFileSystem(getConf());
-    FileStatus status[] = srcFs.globStatus(new Path(src));
-    if (status==null || status.length==0) {
-      throw new FileNotFoundException("Cannot access " + src + 
-          ": No such file or directory.");
+  void dus(String[] cmd, int pos) throws IOException {
+    String newcmd[] = new String[cmd.length + 1];
+    System.arraycopy(cmd, 0, newcmd, 0, cmd.length);
+    newcmd[cmd.length] = "-s";
+    du(newcmd, pos);
+  }
+
+  private void printUsageSummary(List<UsagePair> usages,
+                                 boolean humanReadable) {
+    int maxColumnWidth = 0;
+    for (UsagePair usage : usages) {
+      String toPrint = humanReadable ?
+        StringUtils.humanReadableInt(usage.bytes) : String.valueOf(usage.bytes);
+      if (toPrint.length() > maxColumnWidth) {
+        maxColumnWidth = toPrint.length();
+      }
     }
-    for(int i=0; i<status.length; i++) {
-      long totalSize = srcFs.getContentSummary(status[i].getPath()).getLength();
-      String pathStr = status[i].getPath().toString();
-      System.out.println(("".equals(pathStr)?".":pathStr) + "\t" + totalSize);
+
+    for (UsagePair usage : usages) {
+      String toPrint = humanReadable ?
+        StringUtils.humanReadableInt(usage.bytes) : String.valueOf(usage.bytes);
+      System.out.printf("%-"+ (maxColumnWidth + BORDER) +"s", toPrint);
+      System.out.println(usage.path);
     }
   }
 
@@ -1129,10 +1170,13 @@
 
     while (true) {
       FSDataInputStream in = srcFs.open(path);
-      in.seek(offset);
-      IOUtils.copyBytes(in, System.out, 1024, false);
-      offset = in.getPos();
-      in.close();
+      try {
+        in.seek(offset);
+        IOUtils.copyBytes(in, System.out, 1024);
+        offset = in.getPos();
+      } finally {
+        in.close();
+      }
       if (!foption) {
         break;
       }
@@ -1558,10 +1602,6 @@
           delete(argv[i], true);
         } else if ("-df".equals(cmd)) {
           df(argv[i]);
-        } else if ("-du".equals(cmd)) {
-          du(argv[i]);
-        } else if ("-dus".equals(cmd)) {
-          dus(argv[i]);
         } else if (Count.matches(cmd)) {
           new Count(argv, i, getConf()).runAll();
         } else if ("-ls".equals(cmd)) {
@@ -1809,17 +1849,9 @@
           df(null);
         }
       } else if ("-du".equals(cmd)) {
-        if (i < argv.length) {
-          exitCode = doall(cmd, argv, i);
-        } else {
-          du(".");
-        }
+        du(argv, i);
       } else if ("-dus".equals(cmd)) {
-        if (i < argv.length) {
-          exitCode = doall(cmd, argv, i);
-        } else {
-          dus(".");
-        }         
+        dus(argv, i);
       } else if (Count.matches(cmd)) {
         exitCode = new Count(argv, i, getConf()).runAll();
       } else if ("-mkdir".equals(cmd)) {
@@ -1922,4 +1954,18 @@
           throw new IOException("Multiple IOExceptions: " + exceptions);
     }
   }
+
+
+  /**
+   * Utility class for a line of du output
+   */
+  private static class UsagePair {
+    public String path;
+    public long bytes;
+
+    public UsagePair(String path, long bytes) {
+      this.path = path;
+      this.bytes = bytes;
+    }
+  }
 }

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/RawLocalFileSystem.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/RawLocalFileSystem.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/RawLocalFileSystem.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/RawLocalFileSystem.java Mon Jun 15 20:28:06 2009
@@ -28,6 +28,7 @@
 import java.io.OutputStream;
 import java.net.URI;
 import java.nio.ByteBuffer;
+import java.util.EnumSet;
 import java.util.StringTokenizer;
 
 import org.apache.hadoop.conf.Configuration;
@@ -244,10 +245,19 @@
   /** {@inheritDoc} */
   @Override
   public FSDataOutputStream create(Path f, FsPermission permission,
-      boolean overwrite, int bufferSize, short replication, long blockSize,
+      EnumSet<CreateFlag> flag, int bufferSize, short replication, long blockSize,
       Progressable progress) throws IOException {
+    
+      if(flag.contains(CreateFlag.APPEND)){
+        if (!exists(f)){
+          if(flag.contains(CreateFlag.CREATE))
+            return create(f, false, bufferSize, replication, blockSize, progress);
+        }
+        return append(f, bufferSize, progress);
+    }
+   
     FSDataOutputStream out = create(f,
-        overwrite, bufferSize, replication, blockSize, progress);
+        flag.contains(CreateFlag.OVERWRITE), bufferSize, replication, blockSize, progress);
     setPermission(f, permission);
     return out;
   }

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/ftp/FTPFileSystem.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/ftp/FTPFileSystem.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/ftp/FTPFileSystem.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/ftp/FTPFileSystem.java Mon Jun 15 20:28:06 2009
@@ -21,6 +21,8 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
+import java.util.EnumSet;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.commons.net.ftp.FTP;
@@ -28,6 +30,7 @@
 import org.apache.commons.net.ftp.FTPFile;
 import org.apache.commons.net.ftp.FTPReply;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CreateFlag;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
@@ -193,19 +196,30 @@
    */
   @Override
   public FSDataOutputStream create(Path file, FsPermission permission,
-      boolean overwrite, int bufferSize, short replication, long blockSize,
+      EnumSet<CreateFlag> flag, int bufferSize, short replication, long blockSize,
       Progressable progress) throws IOException {
     final FTPClient client = connect();
     Path workDir = new Path(client.printWorkingDirectory());
     Path absolute = makeAbsolute(workDir, file);
+    
+    boolean overwrite = flag.contains(CreateFlag.OVERWRITE);
+    boolean create = flag.contains(CreateFlag.CREATE);
+    boolean append= flag.contains(CreateFlag.APPEND);
+    
     if (exists(client, file)) {
       if (overwrite) {
         delete(client, file);
+      } else if(append){
+        return append(file, bufferSize, progress);
       } else {
         disconnect(client);
         throw new IOException("File already exists: " + file);
       }
+    } else {
+      if(append && !create)
+        throw new FileNotFoundException("File does not exist: "+ file);
     }
+    
     Path parent = absolute.getParent();
     if (parent == null || !mkdirs(client, parent, FsPermission.getDefault())) {
       parent = (parent == null) ? new Path("/") : parent;

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/IFSImpl.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/IFSImpl.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/IFSImpl.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/IFSImpl.java Mon Jun 15 20:28:06 2009
@@ -12,7 +12,6 @@
  * implied. See the License for the specific language governing
  * permissions and limitations under the License.
  *
- * @author: Sriram Rao (Kosmix Corp.)
  * 
  * We need to provide the ability to the code in fs/kfs without really
  * having a KFS deployment.  In particular, the glue code that wraps

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSImpl.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSImpl.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSImpl.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSImpl.java Mon Jun 15 20:28:06 2009
@@ -12,7 +12,6 @@
  * implied. See the License for the specific language governing
  * permissions and limitations under the License.
  *
- * @author: Sriram Rao (Kosmix Corp.)
  * 
  * Provide the implementation of KFS which turn into calls to KfsAccess.
  */

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSInputStream.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSInputStream.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSInputStream.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSInputStream.java Mon Jun 15 20:28:06 2009
@@ -12,7 +12,6 @@
  * implied. See the License for the specific language governing
  * permissions and limitations under the License.
  *
- * @author: Sriram Rao (Kosmix Corp.)
  * 
  * Implements the Hadoop FSInputStream interfaces to allow applications to read
  * files in Kosmos File System (KFS).

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSOutputStream.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSOutputStream.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSOutputStream.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KFSOutputStream.java Mon Jun 15 20:28:06 2009
@@ -12,7 +12,6 @@
  * implied. See the License for the specific language governing
  * permissions and limitations under the License.
  *
- * @author: Sriram Rao (Kosmix Corp.)
  * 
  * Implements the Hadoop FSOutputStream interfaces to allow applications to write to
  * files in Kosmos File System (KFS).

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KosmosFileSystem.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KosmosFileSystem.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KosmosFileSystem.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/kfs/KosmosFileSystem.java Mon Jun 15 20:28:06 2009
@@ -12,7 +12,6 @@
  * implied. See the License for the specific language governing
  * permissions and limitations under the License.
  *
- * @author: Sriram Rao (Kosmix Corp.)
  * 
  * Implements the Hadoop FS interfaces to allow applications to store
  *files in Kosmos File System (KFS).
@@ -23,9 +22,11 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
+import java.util.EnumSet;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.CreateFlag;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
@@ -120,7 +121,6 @@
     }
 
     @Override
-    @Deprecated
     public boolean isDirectory(Path path) throws IOException {
 	Path absolute = makeAbsolute(path);
         String srep = absolute.toUri().getPath();
@@ -131,7 +131,6 @@
     }
 
     @Override
-    @Deprecated
     public boolean isFile(Path path) throws IOException {
 	Path absolute = makeAbsolute(path);
         String srep = absolute.toUri().getPath();
@@ -186,16 +185,25 @@
 
     @Override
     public FSDataOutputStream create(Path file, FsPermission permission,
-                                     boolean overwrite, int bufferSize,
+                                     EnumSet<CreateFlag> flag, int bufferSize,
 				     short replication, long blockSize, Progressable progress)
 	throws IOException {
 
+      boolean overwrite = flag.contains(CreateFlag.OVERWRITE);
+      boolean create = flag.contains(CreateFlag.CREATE);
+      boolean append= flag.contains(CreateFlag.APPEND);
+      
         if (exists(file)) {
             if (overwrite) {
                 delete(file, true);
+            } else if (append){
+             return append(file, bufferSize, progress);
             } else {
                 throw new IOException("File already exists: " + file);
             }
+        } else {
+          if(append && !create)
+            throw new FileNotFoundException("File does not exist: "+ file);
         }
 
 	Path parent = file.getParent();

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3/S3FileSystem.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3/S3FileSystem.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3/S3FileSystem.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3/S3FileSystem.java Mon Jun 15 20:28:06 2009
@@ -22,12 +22,14 @@
 import java.io.IOException;
 import java.net.URI;
 import java.util.ArrayList;
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CreateFlag;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
@@ -201,18 +203,24 @@
    */
   @Override
   public FSDataOutputStream create(Path file, FsPermission permission,
-      boolean overwrite, int bufferSize,
+      EnumSet<CreateFlag> flag, int bufferSize,
       short replication, long blockSize, Progressable progress)
     throws IOException {
 
     INode inode = store.retrieveINode(makeAbsolute(file));
     if (inode != null) {
-      if (overwrite) {
+      if (flag.contains(CreateFlag.OVERWRITE)) {
         delete(file, true);
+      } else if (flag.contains(CreateFlag.APPEND)){
+        return append(file, bufferSize, progress);
       } else {
         throw new IOException("File already exists: " + file);
       }
     } else {
+      
+      if(flag.contains(CreateFlag.APPEND) && !flag.contains(CreateFlag.CREATE))
+        throw new FileNotFoundException("File does not exist: "+ file);
+      
       Path parent = file.getParent();
       if (parent != null) {
         if (!mkdirs(parent)) {
@@ -324,6 +332,11 @@
     }
     return new S3FileStatus(f.makeQualified(this), inode);
   }
+  
+  @Override
+  public long getDefaultBlockSize() {
+    return getConf().getLong("fs.s3.block.size", 64 * 1024 * 1024);
+  }
 
   // diagnostic methods
 

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/Jets3tNativeFileSystemStore.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/Jets3tNativeFileSystemStore.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/Jets3tNativeFileSystemStore.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/Jets3tNativeFileSystemStore.java Mon Jun 15 20:28:06 2009
@@ -24,6 +24,7 @@
 import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileInputStream;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
@@ -53,10 +54,7 @@
             s3Credentials.getSecretAccessKey());
       this.s3Service = new RestS3Service(awsCredentials);
     } catch (S3ServiceException e) {
-      if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      }
-      throw new S3Exception(e);
+      handleServiceException(e);
     }
     bucket = new S3Bucket(uri.getHost());
   }
@@ -76,10 +74,7 @@
       }
       s3Service.putObject(bucket, object);
     } catch (S3ServiceException e) {
-      if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      }
-      throw new S3Exception(e);
+      handleServiceException(e);
     } finally {
       if (in != null) {
         try {
@@ -99,10 +94,7 @@
       object.setContentLength(0);
       s3Service.putObject(bucket, object);
     } catch (S3ServiceException e) {
-      if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      }
-      throw new S3Exception(e);
+      handleServiceException(e);
     }
   }
   
@@ -116,10 +108,8 @@
       if (e.getMessage().contains("ResponseCode=404")) {
         return null;
       }
-      if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      }
-      throw new S3Exception(e);
+      handleServiceException(e);
+      return null; //never returned - keep compiler happy
     }
   }
   
@@ -128,13 +118,8 @@
       S3Object object = s3Service.getObject(bucket, key);
       return object.getDataInputStream();
     } catch (S3ServiceException e) {
-      if ("NoSuchKey".equals(e.getS3ErrorCode())) {
-        return null;
-      }
-      if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      }
-      throw new S3Exception(e);
+      handleServiceException(key, e);
+      return null; //never returned - keep compiler happy
     }
   }
   
@@ -145,32 +130,22 @@
                                             null, byteRangeStart, null);
       return object.getDataInputStream();
     } catch (S3ServiceException e) {
-      if ("NoSuchKey".equals(e.getS3ErrorCode())) {
-        return null;
-      }
-      if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      }
-      throw new S3Exception(e);
+      handleServiceException(key, e);
+      return null; //never returned - keep compiler happy
     }
   }
 
   public PartialListing list(String prefix, int maxListingLength)
     throws IOException {
-    return list(prefix, maxListingLength, null);
+    return list(prefix, maxListingLength, null, false);
   }
   
-  public PartialListing list(String prefix, int maxListingLength,
-      String priorLastKey) throws IOException {
+  public PartialListing list(String prefix, int maxListingLength, String priorLastKey,
+      boolean recurse) throws IOException {
 
-    return list(prefix, PATH_DELIMITER, maxListingLength, priorLastKey);
+    return list(prefix, recurse ? null : PATH_DELIMITER, maxListingLength, priorLastKey);
   }
 
-  public PartialListing listAll(String prefix, int maxListingLength,
-      String priorLastKey) throws IOException {
-
-    return list(prefix, null, maxListingLength, priorLastKey);
-  }
 
   private PartialListing list(String prefix, String delimiter,
       int maxListingLength, String priorLastKey) throws IOException {
@@ -191,10 +166,8 @@
       return new PartialListing(chunk.getPriorLastKey(), fileMetadata,
           chunk.getCommonPrefixes());
     } catch (S3ServiceException e) {
-      if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      }
-      throw new S3Exception(e);
+      handleServiceException(e);
+      return null; //never returned - keep compiler happy
     }
   }
 
@@ -202,36 +175,27 @@
     try {
       s3Service.deleteObject(bucket, key);
     } catch (S3ServiceException e) {
-      if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      }
-      throw new S3Exception(e);
+      handleServiceException(key, e);
     }
   }
   
-  public void rename(String srcKey, String dstKey) throws IOException {
+  public void copy(String srcKey, String dstKey) throws IOException {
     try {
-      s3Service.moveObject(bucket.getName(), srcKey, bucket.getName(),
+      s3Service.copyObject(bucket.getName(), srcKey, bucket.getName(),
           new S3Object(dstKey), false);
     } catch (S3ServiceException e) {
-      if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      }
-      throw new S3Exception(e);
+      handleServiceException(srcKey, e);
     }
   }
 
   public void purge(String prefix) throws IOException {
     try {
       S3Object[] objects = s3Service.listObjects(bucket, prefix, null);
-      for (int i = 0; i < objects.length; i++) {
-        s3Service.deleteObject(bucket, objects[i].getKey());
+      for (S3Object object : objects) {
+        s3Service.deleteObject(bucket, object.getKey());
       }
     } catch (S3ServiceException e) {
-      if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      }
-      throw new S3Exception(e);
+      handleServiceException(e);
     }
   }
 
@@ -240,16 +204,29 @@
     sb.append(bucket.getName()).append("\n");
     try {
       S3Object[] objects = s3Service.listObjects(bucket);
-      for (int i = 0; i < objects.length; i++) {
-        sb.append(objects[i].getKey()).append("\n");
+      for (S3Object object : objects) {
+        sb.append(object.getKey()).append("\n");
       }
     } catch (S3ServiceException e) {
-      if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      }
-      throw new S3Exception(e);
+      handleServiceException(e);
     }
     System.out.println(sb);
   }
-  
+
+  private void handleServiceException(String key, S3ServiceException e) throws IOException {
+    if ("NoSuchKey".equals(e.getS3ErrorCode())) {
+      throw new FileNotFoundException("Key '" + key + "' does not exist in S3");
+    } else {
+      handleServiceException(e);
+    }
+  }
+
+  private void handleServiceException(S3ServiceException e) throws IOException {
+    if (e.getCause() instanceof IOException) {
+      throw (IOException) e.getCause();
+    }
+    else {
+      throw new S3Exception(e);
+    }
+  }
 }

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/NativeFileSystemStore.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/NativeFileSystemStore.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/NativeFileSystemStore.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/NativeFileSystemStore.java Mon Jun 15 20:28:06 2009
@@ -42,14 +42,12 @@
   InputStream retrieve(String key, long byteRangeStart) throws IOException;
   
   PartialListing list(String prefix, int maxListingLength) throws IOException;
-  PartialListing list(String prefix, int maxListingLength, String priorLastKey)
+  PartialListing list(String prefix, int maxListingLength, String priorLastKey, boolean recursive)
     throws IOException;
-  PartialListing listAll(String prefix, int maxListingLength,
-      String priorLastKey) throws IOException;
   
   void delete(String key) throws IOException;
 
-  void rename(String srcKey, String dstKey) throws IOException;
+  void copy(String srcKey, String dstKey) throws IOException;
   
   /**
    * Delete all keys with the given prefix. Used for testing.

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/NativeS3FileSystem.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/NativeS3FileSystem.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/NativeS3FileSystem.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/fs/s3native/NativeS3FileSystem.java Mon Jun 15 20:28:06 2009
@@ -30,6 +30,7 @@
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -41,6 +42,7 @@
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BufferedFSInputStream;
+import org.apache.hadoop.fs.CreateFlag;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FSInputStream;
@@ -61,6 +63,17 @@
  * Unlike {@link org.apache.hadoop.fs.s3.S3FileSystem} this implementation
  * stores files on S3 in their
  * native form so they can be read by other S3 tools.
+ *
+ * A note about directories. S3 of course has no "native" support for them.
+ * The idiom we choose then is: for any directory created by this class,
+ * we use an empty object "#{dirpath}_$folder$" as a marker.
+ * Further, to interoperate with other S3 tools, we also accept the following:
+ *  - an object "#{dirpath}/' denoting a directory marker
+ *  - if there exists any objects with the prefix "#{dirpath}/", then the
+ *    directory is said to exist
+ *  - if both a file with the name of a directory and a marker for that
+ *    directory exists, then the *file masks the directory*, and the directory
+ *    is never returned.
  * </p>
  * @see org.apache.hadoop.fs.s3.S3FileSystem
  */
@@ -70,7 +83,6 @@
     LogFactory.getLog(NativeS3FileSystem.class);
   
   private static final String FOLDER_SUFFIX = "_$folder$";
-  private static final long MAX_S3_FILE_SIZE = 5 * 1024 * 1024 * 1024L;
   static final String PATH_DELIMITER = Path.SEPARATOR;
   private static final int S3_MAX_LISTING_LENGTH = 1000;
   
@@ -85,6 +97,7 @@
       this.key = key;
     }
     
+    @Override
     public synchronized int read() throws IOException {
       int result = in.read();
       if (result != -1) {
@@ -95,6 +108,7 @@
       }
       return result;
     }
+    @Override
     public synchronized int read(byte[] b, int off, int len)
       throws IOException {
       
@@ -108,18 +122,23 @@
       return result;
     }
 
+    @Override
     public void close() throws IOException {
       in.close();
     }
 
+    @Override
     public synchronized void seek(long pos) throws IOException {
       in.close();
+      LOG.info("Opening key '" + key + "' for reading at position '" + pos + "'");
       in = store.retrieve(key, pos);
       this.pos = pos;
     }
+    @Override
     public synchronized long getPos() throws IOException {
       return pos;
     }
+    @Override
     public boolean seekToNewSource(long targetPos) throws IOException {
       return false;
     }
@@ -140,6 +159,7 @@
       this.conf = conf;
       this.key = key;
       this.backupFile = newBackupFile();
+      LOG.info("OutputStream for key '" + key + "' writing to tempfile '" + this.backupFile + "'");
       try {
         this.digest = MessageDigest.getInstance("MD5");
         this.backupStream = new BufferedOutputStream(new DigestOutputStream(
@@ -174,6 +194,7 @@
       }
 
       backupStream.close();
+      LOG.info("OutputStream for key '" + key + "' closed. Now beginning upload");
       
       try {
         byte[] md5Hash = digest == null ? null : digest.digest();
@@ -185,7 +206,7 @@
         super.close();
         closed = true;
       } 
-
+      LOG.info("OutputStream for key '" + key + "' upload complete");
     }
 
     @Override
@@ -197,8 +218,6 @@
     public void write(byte[] b, int off, int len) throws IOException {
       backupStream.write(b, off, len);
     }
-    
-    
   }
   
   private URI uri;
@@ -242,6 +261,7 @@
     Map<String, RetryPolicy> methodNameToPolicyMap =
       new HashMap<String, RetryPolicy>();
     methodNameToPolicyMap.put("storeFile", methodPolicy);
+    methodNameToPolicyMap.put("rename", methodPolicy);
     
     return (NativeFileSystemStore)
       RetryProxy.create(NativeFileSystemStore.class, store,
@@ -249,10 +269,19 @@
   }
   
   private static String pathToKey(Path path) {
+    if (path.toUri().getScheme() != null && "".equals(path.toUri().getPath())) {
+      // allow uris without trailing slash after bucket to refer to root,
+      // like s3n://mybucket
+      return "";
+    }
     if (!path.isAbsolute()) {
       throw new IllegalArgumentException("Path must be absolute: " + path);
     }
-    return path.toUri().getPath().substring(1); // remove initial slash
+    String ret = path.toUri().getPath().substring(1); // remove initial slash
+    if (ret.endsWith("/") && (ret.indexOf("/") != ret.length() - 1)) {
+      ret = ret.substring(0, ret.length() -1);
+  }
+    return ret;
   }
   
   private static Path keyToPath(String key) {
@@ -267,6 +296,7 @@
   }
 
   /** This optional operation is not yet supported. */
+  @Override
   public FSDataOutputStream append(Path f, int bufferSize,
       Progressable progress) throws IOException {
     throw new IOException("Not supported");
@@ -274,12 +304,21 @@
   
   @Override
   public FSDataOutputStream create(Path f, FsPermission permission,
-      boolean overwrite, int bufferSize, short replication, long blockSize,
+      EnumSet<CreateFlag> flag, int bufferSize, short replication, long blockSize,
       Progressable progress) throws IOException {
 
-    if (exists(f) && !overwrite) {
-      throw new IOException("File already exists:"+f);
+    if(exists(f)) {
+      if(flag.contains(CreateFlag.APPEND)){
+        return append(f, bufferSize, progress);
+      } else if(!flag.contains(CreateFlag.OVERWRITE)) {
+        throw new IOException("File already exists: "+f);
+      }
+    } else {
+      if (flag.contains(CreateFlag.APPEND) && !flag.contains(CreateFlag.CREATE))
+        throw new IOException("File already exists: " + f.toString());
     }
+    
+    LOG.debug("Creating new file '" + f + "' in S3");
     Path absolutePath = makeAbsolute(f);
     String key = pathToKey(absolutePath);
     return new FSDataOutputStream(new NativeS3FsOutputStream(getConf(), store,
@@ -287,27 +326,41 @@
   }
   
   @Override
-  public boolean delete(Path f, boolean recursive) throws IOException {
+  public boolean delete(Path f, boolean recurse) throws IOException {
     FileStatus status;
     try {
       status = getFileStatus(f);
     } catch (FileNotFoundException e) {
+      LOG.debug("Delete called for '" + f + "' but file does not exist, so returning false");
       return false;
     }
     Path absolutePath = makeAbsolute(f);
     String key = pathToKey(absolutePath);
     if (status.isDir()) {
-      FileStatus[] contents = listStatus(f);
-      if (!recursive && contents.length > 0) {
-        throw new IOException("Directory " + f.toString() + " is not empty.");
+      if (!recurse && listStatus(f).length > 0) {
+        throw new IOException("Can not delete " + f + " at is a not empty directory and recurse option is false");
       }
-      for (FileStatus p : contents) {
-        if (!delete(p.getPath(), recursive)) {
-          return false;
+
+      createParent(f);
+
+      LOG.debug("Deleting directory '" + f  + "'");
+      String priorLastKey = null;
+      do {
+        PartialListing listing = store.list(key, S3_MAX_LISTING_LENGTH, priorLastKey, true);
+        for (FileMetadata file : listing.getFiles()) {
+          store.delete(file.getKey());
         }
+        priorLastKey = listing.getPriorLastKey();
+      } while (priorLastKey != null);
+
+      try {
+        store.delete(key + FOLDER_SUFFIX);
+      } catch (FileNotFoundException e) {
+        //this is fine, we don't require a marker
       }
-      store.delete(key + FOLDER_SUFFIX);
     } else {
+      LOG.debug("Deleting file '" + f + "'");
+      createParent(f);
       store.delete(key);
     }
     return true;
@@ -315,7 +368,6 @@
 
   @Override
   public FileStatus getFileStatus(Path f) throws IOException {
-    
     Path absolutePath = makeAbsolute(f);
     String key = pathToKey(absolutePath);
     
@@ -323,23 +375,28 @@
       return newDirectory(absolutePath);
     }
     
+    LOG.debug("getFileStatus retrieving metadata for key '" + key + "'");
     FileMetadata meta = store.retrieveMetadata(key);
     if (meta != null) {
+      LOG.debug("getFileStatus returning 'file' for key '" + key + "'");
       return newFile(meta, absolutePath);
     }
     if (store.retrieveMetadata(key + FOLDER_SUFFIX) != null) {
+      LOG.debug("getFileStatus returning 'directory' for key '" + key + "' as '"
+          + key + FOLDER_SUFFIX + "' exists");
       return newDirectory(absolutePath);
     }
     
+    LOG.debug("getFileStatus listing key '" + key + "'");
     PartialListing listing = store.list(key, 1);
     if (listing.getFiles().length > 0 ||
         listing.getCommonPrefixes().length > 0) {
+      LOG.debug("getFileStatus returning 'directory' for key '" + key + "' as it has contents");
       return newDirectory(absolutePath);
     }
     
-    throw new FileNotFoundException(absolutePath +
-        ": No such file or directory.");
-    
+    LOG.debug("getFileStatus could not find key '" + key + "'");
+    throw new FileNotFoundException("No such file or directory '" + absolutePath + "'");
   }
 
   @Override
@@ -372,16 +429,20 @@
     Set<FileStatus> status = new TreeSet<FileStatus>();
     String priorLastKey = null;
     do {
-      PartialListing listing = store.list(key, S3_MAX_LISTING_LENGTH, 
-          priorLastKey);
+      PartialListing listing = store.list(key, S3_MAX_LISTING_LENGTH, priorLastKey, false);
       for (FileMetadata fileMetadata : listing.getFiles()) {
         Path subpath = keyToPath(fileMetadata.getKey());
         String relativePath = pathUri.relativize(subpath.toUri()).getPath();
-        if (relativePath.endsWith(FOLDER_SUFFIX)) {
-          status.add(newDirectory(new Path(absolutePath,
-              relativePath.substring(0,
-                  relativePath.indexOf(FOLDER_SUFFIX)))));
-        } else {
+
+        if (fileMetadata.getKey().equals(key + "/")) {
+          // this is just the directory we have been asked to list
+        }
+        else if (relativePath.endsWith(FOLDER_SUFFIX)) {
+          status.add(newDirectory(new Path(
+              absolutePath,
+              relativePath.substring(0, relativePath.indexOf(FOLDER_SUFFIX)))));
+        }
+        else {
           status.add(newFile(fileMetadata, subpath));
         }
       }
@@ -398,17 +459,16 @@
       return null;
     }
     
-    return status.toArray(new FileStatus[0]);
+    return status.toArray(new FileStatus[status.size()]);
   }
   
   private FileStatus newFile(FileMetadata meta, Path path) {
-    return new FileStatus(meta.getLength(), false, 1, MAX_S3_FILE_SIZE,
+    return new FileStatus(meta.getLength(), false, 1, getDefaultBlockSize(),
         meta.getLastModified(), path.makeQualified(this));
   }
   
   private FileStatus newDirectory(Path path) {
-    return new FileStatus(0, true, 1, MAX_S3_FILE_SIZE, 0,
-        path.makeQualified(this));
+    return new FileStatus(0, true, 1, 0, 0, path.makeQualified(this));
   }
 
   @Override
@@ -432,10 +492,11 @@
       FileStatus fileStatus = getFileStatus(f);
       if (!fileStatus.isDir()) {
         throw new IOException(String.format(
-            "Can't make directory for path %s since it is a file.", f));
+            "Can't make directory for path '%s' since it is a file.", f));
 
       }
     } catch (FileNotFoundException e) {
+      LOG.debug("Making dir '" + f + "' in S3");
       String key = pathToKey(f) + FOLDER_SUFFIX;
       store.storeEmptyFile(key);    
     }
@@ -444,9 +505,11 @@
 
   @Override
   public FSDataInputStream open(Path f, int bufferSize) throws IOException {
-    if (!exists(f)) {
-      throw new FileNotFoundException(f.toString());
+    FileStatus fs = getFileStatus(f); // will throw if the file doesn't exist
+    if (fs.isDir()) {
+      throw new IOException("'" + f + "' is a directory");
     }
+    LOG.info("Opening '" + f + "' for reading");
     Path absolutePath = makeAbsolute(f);
     String key = pathToKey(absolutePath);
     return new FSDataInputStream(new BufferedFSInputStream(
@@ -456,47 +519,16 @@
   // rename() and delete() use this method to ensure that the parent directory
   // of the source does not vanish.
   private void createParent(Path path) throws IOException {
-      Path parent = path.getParent();
-      if (parent != null) {
-          String key = pathToKey(makeAbsolute(parent));
-          if (key.length() > 0) {
-              store.storeEmptyFile(key + FOLDER_SUFFIX);
-          }
+    Path parent = path.getParent();
+    if (parent != null) {
+      String key = pathToKey(makeAbsolute(parent));
+      if (key.length() > 0) {
+          store.storeEmptyFile(key + FOLDER_SUFFIX);
       }
+    }
   }
   
-  private boolean existsAndIsFile(Path f) throws IOException {
-    
-    Path absolutePath = makeAbsolute(f);
-    String key = pathToKey(absolutePath);
-    
-    if (key.length() == 0) {
-        return false;
-    }
-    
-    FileMetadata meta = store.retrieveMetadata(key);
-    if (meta != null) {
-        // S3 object with given key exists, so this is a file
-        return true;
-    }
     
-    if (store.retrieveMetadata(key + FOLDER_SUFFIX) != null) {
-        // Signifies empty directory
-        return false;
-    }
-    
-    PartialListing listing = store.list(key, 1, null);
-    if (listing.getFiles().length > 0 ||
-        listing.getCommonPrefixes().length > 0) {
-        // Non-empty directory
-        return false;
-    }
-    
-    throw new FileNotFoundException(absolutePath +
-        ": No such file or directory");
-}
-
-
   @Override
   public boolean rename(Path src, Path dst) throws IOException {
 
@@ -507,60 +539,79 @@
       return false;
     }
 
+    final String debugPreamble = "Renaming '" + src + "' to '" + dst + "' - ";
+
     // Figure out the final destination
     String dstKey;
     try {
-      boolean dstIsFile = existsAndIsFile(dst);
+      boolean dstIsFile = !getFileStatus(dst).isDir();
       if (dstIsFile) {
-        // Attempting to overwrite a file using rename()
+        LOG.debug(debugPreamble + "returning false as dst is an already existing file");
         return false;
       } else {
-        // Move to within the existent directory
+        LOG.debug(debugPreamble + "using dst as output directory");
         dstKey = pathToKey(makeAbsolute(new Path(dst, src.getName())));
       }
     } catch (FileNotFoundException e) {
-      // dst doesn't exist, so we can proceed
+      LOG.debug(debugPreamble + "using dst as output destination");
       dstKey = pathToKey(makeAbsolute(dst));
       try {
         if (!getFileStatus(dst.getParent()).isDir()) {
-          return false; // parent dst is a file
+          LOG.debug(debugPreamble + "returning false as dst parent exists and is a file");
+          return false;
         }
       } catch (FileNotFoundException ex) {
-        return false; // parent dst does not exist
+        LOG.debug(debugPreamble + "returning false as dst parent does not exist");
+        return false;
       }
     }
 
+    boolean srcIsFile;
     try {
-      boolean srcIsFile = existsAndIsFile(src);
-      if (srcIsFile) {
-        store.rename(srcKey, dstKey);
-      } else {
-        // Move the folder object
-        store.delete(srcKey + FOLDER_SUFFIX);
-        store.storeEmptyFile(dstKey + FOLDER_SUFFIX);
+      srcIsFile = !getFileStatus(src).isDir();
+    } catch (FileNotFoundException e) {
+      LOG.debug(debugPreamble + "returning false as src does not exist");
+      return false;
+    }
+    if (srcIsFile) {
+      LOG.debug(debugPreamble + "src is file, so doing copy then delete in S3");
+      store.copy(srcKey, dstKey);
+      store.delete(srcKey);
+    } else {
+      LOG.debug(debugPreamble + "src is directory, so copying contents");
+      store.storeEmptyFile(dstKey + FOLDER_SUFFIX);
 
-        // Move everything inside the folder
-        String priorLastKey = null;
-        do {
-          PartialListing listing = store.listAll(srcKey, S3_MAX_LISTING_LENGTH,
-              priorLastKey);
-          for (FileMetadata file : listing.getFiles()) {
-            store.rename(file.getKey(), dstKey
-                + file.getKey().substring(srcKey.length()));
-          }
-          priorLastKey = listing.getPriorLastKey();
-        } while (priorLastKey != null);
-      }
+      List<String> keysToDelete = new ArrayList<String>();
+      String priorLastKey = null;
+      do {
+        PartialListing listing = store.list(srcKey, S3_MAX_LISTING_LENGTH, priorLastKey, true);
+        for (FileMetadata file : listing.getFiles()) {
+          keysToDelete.add(file.getKey());
+          store.copy(file.getKey(), dstKey + file.getKey().substring(srcKey.length()));
+        }
+        priorLastKey = listing.getPriorLastKey();
+      } while (priorLastKey != null);
 
-      createParent(src);
-      return true;
+      LOG.debug(debugPreamble + "all files in src copied, now removing src files");
+      for (String key: keysToDelete) {
+        store.delete(key);
+      }
 
-    } catch (FileNotFoundException e) {
-      // Source file does not exist;
-      return false;
+      try {
+        store.delete(srcKey + FOLDER_SUFFIX);
+      } catch (FileNotFoundException e) {
+        //this is fine, we don't require a marker
+      }
+      LOG.debug(debugPreamble + "done");
     }
-  }
 
+    return true;
+  }
+  
+  @Override
+  public long getDefaultBlockSize() {
+    return getConf().getLong("fs.s3n.block.size", 64 * 1024 * 1024);
+  }
 
   /**
    * Set the working directory to the given directory.
@@ -574,5 +625,4 @@
   public Path getWorkingDirectory() {
     return workingDir;
   }
-
 }

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/http/HttpServer.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/http/HttpServer.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/http/HttpServer.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/http/HttpServer.java Mon Jun 15 20:28:06 2009
@@ -238,13 +238,15 @@
   }
 
   /**
-   * Add an internal servlet in the server.
+   * Add an internal servlet in the server. 
+   * Note: This method is to be used for adding servlets that facilitate
+   * internal communication and not for user facing functionality. For
+   * servlets added using this method, filters are not enabled. 
+   * 
    * @param name The name of the servlet (can be passed as null)
    * @param pathSpec The path spec for the servlet
    * @param clazz The servlet class
-   * @deprecated this is a temporary method
    */
-  @Deprecated
   public void addInternalServlet(String name, String pathSpec,
       Class<? extends HttpServlet> clazz) {
     ServletHolder holder = new ServletHolder(clazz);

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/io/IOUtils.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/io/IOUtils.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/io/IOUtils.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/io/IOUtils.java Mon Jun 15 20:28:06 2009
@@ -41,17 +41,8 @@
   public static void copyBytes(InputStream in, OutputStream out, int buffSize, boolean close) 
     throws IOException {
 
-    PrintStream ps = out instanceof PrintStream ? (PrintStream)out : null;
-    byte buf[] = new byte[buffSize];
     try {
-      int bytesRead = in.read(buf);
-      while (bytesRead >= 0) {
-        out.write(buf, 0, bytesRead);
-        if ((ps != null) && ps.checkError()) {
-          throw new IOException("Unable to write to output stream.");
-        }
-        bytesRead = in.read(buf);
-      }
+      copyBytes(in, out, buffSize);
     } finally {
       if(close) {
         out.close();
@@ -61,6 +52,27 @@
   }
   
   /**
+   * Copies from one stream to another.
+   * 
+   * @param in InputStrem to read from
+   * @param out OutputStream to write to
+   * @param buffSize the size of the buffer 
+   */
+  public static void copyBytes(InputStream in, OutputStream out, int buffSize) 
+    throws IOException {
+
+    PrintStream ps = out instanceof PrintStream ? (PrintStream)out : null;
+    byte buf[] = new byte[buffSize];
+    int bytesRead = in.read(buf);
+    while (bytesRead >= 0) {
+      out.write(buf, 0, bytesRead);
+      if ((ps != null) && ps.checkError()) {
+        throw new IOException("Unable to write to output stream.");
+      }
+      bytesRead = in.read(buf);
+    }
+  }
+  /**
    * Copies from one stream to another. <strong>closes the input and output streams 
    * at the end</strong>.
    * @param in InputStrem to read from

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/metrics/ContextFactory.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/metrics/ContextFactory.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/metrics/ContextFactory.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/metrics/ContextFactory.java Mon Jun 15 20:28:06 2009
@@ -188,16 +188,19 @@
   private void setAttributes() throws IOException {
     InputStream is = getClass().getResourceAsStream(PROPERTIES_FILE);
     if (is != null) {
-      Properties properties = new Properties();
-      properties.load(is);
-      //for (Object propertyNameObj : properties.keySet()) {
-      Iterator it = properties.keySet().iterator();
-      while (it.hasNext()) {
-        String propertyName = (String) it.next();
-        String propertyValue = properties.getProperty(propertyName);
-        setAttribute(propertyName, propertyValue);
+      try {
+        Properties properties = new Properties();
+        properties.load(is);
+        //for (Object propertyNameObj : properties.keySet()) {
+        Iterator it = properties.keySet().iterator();
+        while (it.hasNext()) {
+          String propertyName = (String) it.next();
+          String propertyValue = properties.getProperty(propertyName);
+          setAttribute(propertyName, propertyValue);
+        }
+      } finally {
+        is.close();
       }
-      is.close();
     }
   }
     

Modified: hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/net/NetUtils.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/net/NetUtils.java?rev=784965&r1=784964&r2=784965&view=diff
==============================================================================
--- hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/net/NetUtils.java (original)
+++ hadoop/core/branches/HADOOP-4687/core/src/java/org/apache/hadoop/net/NetUtils.java Mon Jun 15 20:28:06 2009
@@ -132,6 +132,9 @@
    */
   public static InetSocketAddress createSocketAddr(String target,
                                                    int defaultPort) {
+    if (target == null) {
+      throw new IllegalArgumentException("Target address cannot be null.");
+    }
     int colonIndex = target.indexOf(':');
     if (colonIndex < 0 && defaultPort == -1) {
       throw new RuntimeException("Not a host:port pair: " + target);



Mime
View raw message