HADOOP-3003. FileSystem cache key is updated after a

FileSystem object is created. (Tsz Wo (Nicholas), SZE via dhruba)
svn merge -c 639052 



git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.16@639053 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -18,6 +18,9 @@ Release 0.16.2 - Unreleased
     Also makes the _temporary a constant in MRConstants.java.
     (Amareshwari Sriramadasu via ddas)
 
+    HADOOP-3003. FileSystem cache key is updated after a 
+    FileSystem object is created. (Tsz Wo (Nicholas), SZE via dhruba)
+
 Release 0.16.1 - 2008-03-13
 
   INCOMPATIBLE CHANGES

src/java/org/apache/hadoop/dfs/DistributedFileSystem.java

@@ -179,9 +179,13 @@ public class DistributedFileSystem extends FileSystem {
     return dfs.mkdirs(getPathName(f), permission);
   }
 
-  public void close() throws IOException {
-    super.close();
-    dfs.close();
+  /** {@inheritDoc} */
+  public synchronized void close() throws IOException {
+    try {
+      dfs.close();
+    } finally {
+      super.close();
+    }
   }
 
   public String toString() {

src/java/org/apache/hadoop/fs/FileSystem.java

@@ -1082,7 +1082,7 @@ public abstract class FileSystem extends Configured implements Closeable {
    * release any held locks.
    */
   public void close() throws IOException {
-    CACHE.remove(new Cache.Key(getUri(), getConf()));
+    CACHE.remove(new Cache.Key(this), this);
   }
 
   /** Return the total size of all files in the filesystem.*/
@@ -1183,39 +1183,51 @@ public abstract class FileSystem extends Configured implements Closeable {
 
   /** Caching FileSystem objects */
   private static class Cache {
-    final Map<Key, FsRef> map = new HashMap<Key, FsRef>();
+    private final Map<Key, FileSystem> map = new HashMap<Key, FileSystem>();
 
     synchronized FileSystem get(URI uri, Configuration conf) throws IOException{
-      Key key = new Key(uri, conf);
-      FsRef ref = map.get(key);
-      FileSystem fs = ref == null? null: ref.get();
+      FileSystem fs = map.get(new Key(uri, conf));
       if (fs == null) {
         if (map.isEmpty() && !clientFinalizer.isAlive()) {
           Runtime.getRuntime().addShutdownHook(clientFinalizer);
         }
 
         fs = createFileSystem(uri, conf);
-        map.put(key, new FsRef(fs, key));
+        map.put(new Key(fs), fs);
       }
       return fs;
     }
 
-    synchronized FsRef remove(Key key) {
-      FsRef ref = map.remove(key);
-      if (map.isEmpty() && !clientFinalizer.isAlive()) {
-        if (!Runtime.getRuntime().removeShutdownHook(clientFinalizer)) {
-          LOG.info("Could not cancel cleanup thread, though no " +
-                   "FileSystems are open");
+    synchronized void remove(Key key, FileSystem fs) {
+      if (map.containsKey(key) && fs == map.get(key)) {
+        map.remove(key);
+        if (map.isEmpty() && !clientFinalizer.isAlive()) {
+          if (!Runtime.getRuntime().removeShutdownHook(clientFinalizer)) {
+            LOG.info("Could not cancel cleanup thread, though no " +
+                     "FileSystems are open");
+          }
         }
       }
-      return ref;
     }
 
     synchronized void closeAll() throws IOException {
       List<IOException> exceptions = new ArrayList<IOException>();
-      for(FsRef ref : new ArrayList<FsRef>(map.values())) {
-        FileSystem fs = ref.get();
+      for(; !map.isEmpty(); ) {
+        Map.Entry<Key, FileSystem> e = map.entrySet().iterator().next();
+        final Key key = e.getKey();
+        final FileSystem fs = e.getValue();
+
+        //remove from cache
+        remove(key, fs);
+
         if (fs != null) {
+          //check consistency
+          if (!new Key(fs).equals(key)) {
+            exceptions.add(new IOException(fs.getClass().getSimpleName()
+                + "(=" + fs + ") and " + key.getClass().getSimpleName()
+                + "(=" + key + ") do not match."));
+          }
+
           try {
             fs.close();
           }
@@ -1223,9 +1235,6 @@ public abstract class FileSystem extends Configured implements Closeable {
             exceptions.add(ioe);
           }
         }
-        else {
-          remove(ref.key);
-        }        
       }
 
       if (!exceptions.isEmpty()) {
@@ -1233,25 +1242,16 @@ public abstract class FileSystem extends Configured implements Closeable {
       }
     }
 
-    /** Reference of FileSystem which contains the cache key */
-    private static class FsRef {
-      final FileSystem fs;
-      final Key key;
-      
-      FsRef(FileSystem fs, Key key) {
-        this.fs = fs;
-        this.key = key;
-      }
-
-      FileSystem get() {return fs;}
-    }
-
     /** FileSystem.Cache.Key */
     private static class Key {
       final String scheme;
       final String authority;
       final String username;
 
+      Key(FileSystem fs) throws IOException {
+        this(fs.getUri(), fs.getConf());
+      }
+
       Key(URI uri, Configuration conf) throws IOException {
         scheme = uri.getScheme();
         authority = uri.getAuthority();
@@ -1281,6 +1281,11 @@ public abstract class FileSystem extends Configured implements Closeable {
         }
         return false;        
       }
+
+      /** {@inheritDoc} */
+      public String toString() {
+        return username + "@" + scheme + "://" + authority;        
+      }
     }
   }
 }

src/test/org/apache/hadoop/dfs/TestDistributedFileSystem.java

@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.dfs;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+
+public class TestDistributedFileSystem extends junit.framework.TestCase {
+  public void testFileSystemCloseAll() throws Exception {
+    Configuration conf = new Configuration();
+    MiniDFSCluster cluster = new MiniDFSCluster(conf, 0, true, null);
+    String address = conf.get("fs.default.name");
+
+    try {
+      FileSystem.closeAll();
+
+      conf = new Configuration();
+      conf.set("fs.default.name", address);
+      FileSystem.get(conf);
+      FileSystem.get(conf);
+      FileSystem.closeAll();
+    }
+    finally {
+      if (cluster != null) {cluster.shutdown();}
+    }
+  }
+}

src/test/org/apache/hadoop/fs/TestFileSystem.java

@@ -43,7 +43,6 @@ import org.apache.hadoop.mapred.Reporter;
 import org.apache.hadoop.mapred.SequenceFileInputFormat;
 import org.apache.hadoop.mapred.lib.LongSumReducer;
 import org.apache.hadoop.security.UnixUserGroupInformation;
-import org.apache.hadoop.security.UserGroupInformation;
 
 public class TestFileSystem extends TestCase {
   private static final Log LOG = FileSystem.LOG;
@@ -473,8 +472,18 @@ public class TestFileSystem extends TestCase {
   }
 
   public void testFsCache() throws Exception {
-    Configuration c1 = createConf4Testing("foo");
-    Configuration c2 = createConf4Testing("bar");
-    assertFalse(FileSystem.get(c1) == FileSystem.get(c2));
+    long now = System.currentTimeMillis();
+    Configuration[] conf = {new Configuration(),
+        createConf4Testing("foo" + now), createConf4Testing("bar" + now)};
+    FileSystem[] fs = new FileSystem[conf.length];
+
+    for(int i = 0; i < conf.length; i++) {
+      fs[i] = FileSystem.get(conf[i]);
+      assertEquals(fs[i], FileSystem.get(conf[i]));
+      for(int j = 0; j < i; j++) {
+        assertFalse(fs[j] == fs[i]);
+      }
+    }
+    FileSystem.closeAll();
   }
 }