/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.underfs;

import alluxio.AlluxioURI;
import alluxio.exception.status.NotFoundException;
import alluxio.exception.status.UnavailableException;
import alluxio.metrics.MetricsSystem;
import alluxio.resource.CloseableResource;

import com.codahale.metrics.Counter;
import com.google.common.base.Preconditions;
import com.google.common.base.Supplier;

import java.io.Closeable;

/**
 * A class that manages the UFS used by different services.
 */
public interface UfsManager extends Closeable {
  /** Container for a UFS and the URI for that UFS. */
  class UfsClient {
    private UnderFileSystem mUfs;
    private final AlluxioURI mUfsMountPointUri;
    private final Supplier<UnderFileSystem> mUfsSupplier;
    private final Counter mCounter;

    /**
     * @param ufsSupplier the supplier function to create a new UFS instance
     * @param ufsMountPointUri the URI for the UFS path which is mounted in Alluxio
     */
    public UfsClient(Supplier<UnderFileSystem> ufsSupplier, AlluxioURI ufsMountPointUri) {
      mUfsSupplier = Preconditions.checkNotNull(ufsSupplier, "ufsSupplier is null");
      mUfsMountPointUri = Preconditions.checkNotNull(ufsMountPointUri, "ufsMountPointUri is null");
      mCounter = MetricsSystem.counter(
          String.format("UfsSessionCount-Ufs:%s", MetricsSystem.escape(mUfsMountPointUri)));
    }

    /**
     * @return the UFS instance
     */
    public synchronized CloseableResource<UnderFileSystem> acquireUfsResource() {
      if (mUfs == null) {
        mUfs = mUfsSupplier.get();
      }
      mCounter.inc();
      return new CloseableResource<UnderFileSystem>(mUfs) {
        @Override
        public void close() {
          mCounter.dec();
        }
      };
    }

    /**
     * @return the URI for the UFS path which is mounted in Alluxio
     */
    public AlluxioURI getUfsMountPointUri() {
      return mUfsMountPointUri;
    }
  }

  /**
   * Keeps track of a mount id and maps it to its URI in Alluxio and configuration. This is an
   * Alluxio-only operation and no interaction to UFS will be made.
   *
   * @param mountId the mount id
   * @param ufsUri the UFS path
   * @param ufsConf the UFS configuration
   */
  void addMount(long mountId, AlluxioURI ufsUri, UnderFileSystemConfiguration ufsConf);

  /**
   * Removes the association from a mount id to a UFS instance. If the mount id is not known, this
   * is a noop.
   *
   * @param mountId the mount id
   *
   */
  void removeMount(long mountId);

  /**
   * Gets UFS information from the manager if this mount ID exists, or throws exception otherwise.
   *
   * @param mountId the mount id
   * @return the UFS information
   * @throws NotFoundException if mount id is not found in mount table
   * @throws UnavailableException if master is not available to query for mount table
   */
  UfsClient get(long mountId) throws NotFoundException, UnavailableException;

  /**
   * @return the UFS client associated with root
   */
  UfsClient getRoot();
}

@@ -873,32 +873,169 @@
 er;%0A
-%0Aimport java.io.Closeabl
+import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0Aimport java.io.Closeable;%0Aimport java.io.IOException;%0Aimport java.util.concurrent.atomic.AtomicReferenc
 e;%0A%0A
@@ -1237,16 +1237,119 @@
 private 
+static final Logger LOG = LoggerFactory.getLogger(UfsClient.class);%0A%0A    private final AtomicReference%3C
 UnderFil
@@ -1355,16 +1355,17 @@
 leSystem
+%3E
  mUfs;%0A 
@@ -2092,16 +2092,54 @@
 Uri)));%0A
+      mUfs = new AtomicReference%3C%3E();%0A
     %7D%0A%0A 
@@ -2200,21 +2200,8 @@
 lic 
-synchronized 
 Clos
@@ -2268,16 +2268,22 @@
 if (mUfs
+.get()
  == null
@@ -2290,26 +2290,41 @@
 ) %7B%0A        
-m
 U
+nderFileSystem u
 fs = mUfsSup
@@ -2332,24 +2332,457 @@
 lier.get();%0A
+        if (!mUfs.compareAndSet(null, ufs)) %7B%0A          // Another thread already added this ufs, close this one.%0A          try %7B%0A            ufs.close();%0A          %7D catch (IOException e) %7B%0A            // ignore the close exception, log a warning%0A            LOG.warn(String%0A                .format(%22Failed to close extra UFS. mount point: %25s error: %25s%22, mUfsMountPointUri,%0A                    e.toString()));%0A          %7D%0A        %7D%0A
       %7D%0A    
@@ -2855,16 +2855,22 @@
 em%3E(mUfs
+.get()
 ) %7B%0A    

Remove sync from acquireUfsResource (#7577)