/*
 * Licensed to the University of California, Berkeley 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 tachyon.underfs.hdfs;

import java.util.Map;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;

import tachyon.conf.TachyonConf;
import tachyon.underfs.UnderFileSystem;
import tachyon.underfs.UnderFileSystemFactory;

public class HdfsUnderFileSystemFactory implements UnderFileSystemFactory {
  /**
   * Cache mapping {@link Path}s to existing {@link UnderFileSystem} instances. The Paths should be
   * normalized to root paths because only their schemes and authorities are needed to identify
   * which {@link FileSystem} they belong to.
   */
  private Map<Path, HdfsUnderFileSystem> mHdfsUfsCache = Maps.newHashMap();

  @Override
  public UnderFileSystem create(String path, TachyonConf tachyonConf, Object conf) {
    Preconditions.checkArgument(path != null, "path may not be null");

    // Normalize the path to just its root. This is all that's needed to identify which FileSystem
    // the Path belongs to.
    Path rootPath = getRoot(new Path(path));
    if (!mHdfsUfsCache.containsKey(rootPath)) {
      mHdfsUfsCache.put(rootPath, new HdfsUnderFileSystem(path, tachyonConf, conf));
    }
    return mHdfsUfsCache.get(rootPath);
  }

  private static final Path getRoot(Path path) {
    Path currPath = path;
    while (currPath.getParent() != null) {
      currPath = currPath.getParent();
    }
    return currPath;
  }

  @Override
  public boolean supportsPath(String path, TachyonConf conf) {
    if (path == null) {
      return false;
    }

    return UnderFileSystem.isHadoopUnderFS(path, conf);
  }
}

@@ -1121,22 +1121,237 @@
 ctory;%0A%0A
-public
+/**%0A * Factory for creating HDFS under file systems. As an implementation detail, it caches created%0A * HdfsUnderFileSystems, using paths' schemes and authorities as the keys. This class is%0A * thread-safe.%0A */%0Apublic final
  class H
@@ -2080,24 +2080,61 @@
 ath(path));%0A
+    synchronized (mHdfsUfsCache) %7B%0A  
     if (!mHd
@@ -2165,24 +2165,26 @@
 ootPath)) %7B%0A
+  
       mHdfsU
@@ -2256,30 +2256,34 @@
 conf));%0A    
+  
 %7D%0A    
+ 
+ 
 return mHdfs
@@ -2306,16 +2306,22 @@
 tPath);%0A
+    %7D%0A
   %7D%0A%0A  p
@@ -2334,22 +2334,16 @@
  static 
-final 
 Path get

Make HdfsUnderFileSystemFactory thread safe