HADOOP-19346. ViewFileSystem.InnerCache: Replaced ReentrantReadWriteLock with ConcurrentHashMap/putIfAbsent() (#7187)

* HADOOP-19346 ViewFileSystem.InnerCache: Replaced ReentrantReadWriteLock with ConcurrentHashMap/putIfAbsent()

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java

@@ -28,6 +28,8 @@ import static org.apache.hadoop.fs.viewfs.Constants.PERMISSION_555;
 import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT;
 import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT_DEFAULT;
 
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.function.Function;
 import java.io.FileNotFoundException;
 import java.io.IOException;
@@ -45,7 +47,6 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Set;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import org.apache.hadoop.util.Preconditions;
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -118,38 +119,32 @@ public class ViewFileSystem extends FileSystem {
    * Caching children filesystems. HADOOP-15565.
    */
   static class InnerCache {
-    private Map<Key, FileSystem> map = new HashMap<>();
+    private ConcurrentMap<Key, FileSystem> map = new ConcurrentHashMap<>();
     private FsGetter fsCreator;
-    private ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
 
     InnerCache(FsGetter fsCreator) {
       this.fsCreator = fsCreator;
     }
 
-    FileSystem get(URI uri, Configuration config) throws IOException {
-      Key key = new Key(uri);
-      FileSystem fs = null;
+    // computeIfAbsent() does not support a mapping function which throws IOException.
+    // Wrap fsCreator.getNewInstance() to not throw IOException and return null instead.
+    FileSystem getNewFileSystem(URI uri, Configuration config) {
       try {
-        rwLock.readLock().lock();
-        fs = map.get(key);
-        if (fs != null) {
-          return fs;
-        }
-      } finally {
-        rwLock.readLock().unlock();
+        return fsCreator.getNewInstance(uri, config);
+      } catch (IOException e) {
+        LOG.error("Failed to create new FileSystem instance for " + uri, e);
+        return null;
       }
-      try {
-        rwLock.writeLock().lock();
-        fs = map.get(key);
-        if (fs != null) {
-          return fs;
-        }
-        fs = fsCreator.getNewInstance(uri, config);
-        map.put(key, fs);
-        return fs;
-      } finally {
-        rwLock.writeLock().unlock();
+    }
+
+    FileSystem get(URI uri, Configuration config) throws IOException {
+      Key key = new Key(uri);
+
+      FileSystem fs = map.computeIfAbsent(key, k -> getNewFileSystem(uri, config));
+      if (fs == null) {
+        throw new IOException("Failed to create new FileSystem instance for " + uri);
       }
+      return fs;
     }
 
     void closeAll() {
@@ -163,12 +158,7 @@ public class ViewFileSystem extends FileSystem {
     }
 
     void clear() {
-      try {
-        rwLock.writeLock().lock();
-        map.clear();
-      } finally {
-        rwLock.writeLock().unlock();
-      }
+      map.clear();
     }
 
     /**