/*
 * 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.hdfs;

import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.UfsConstants;
import alluxio.collections.Pair;
import alluxio.conf.PropertyKey;
import alluxio.retry.CountingRetry;
import alluxio.retry.RetryPolicy;
import alluxio.security.authorization.AccessControlList;
import alluxio.security.authorization.AclEntry;
import alluxio.security.authorization.DefaultAccessControlList;
import alluxio.underfs.AtomicFileOutputStream;
import alluxio.underfs.AtomicFileOutputStreamCallback;
import alluxio.underfs.ChecksumType;
import alluxio.underfs.ConsistentUnderFileSystem;
import alluxio.underfs.UfsDirectoryStatus;
import alluxio.underfs.UfsFileStatus;
import alluxio.underfs.UfsStatus;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.UnderFileSystemConfiguration;
import alluxio.underfs.options.CreateOptions;
import alluxio.underfs.options.DeleteOptions;
import alluxio.underfs.options.FileLocationOptions;
import alluxio.underfs.options.GetStatusOptions;
import alluxio.underfs.options.ListOptions;
import alluxio.underfs.options.MkdirsOptions;
import alluxio.underfs.options.OpenOptions;
import alluxio.util.CommonUtils;
import alluxio.util.UnderFileSystemUtils;
import alluxio.util.network.NetworkAddressUtils;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.Trash;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;

/**
 * HDFS {@link UnderFileSystem} implementation.
 */
@ThreadSafe
public class HdfsUnderFileSystem extends ConsistentUnderFileSystem
    implements AtomicFileOutputStreamCallback {
  private static final Logger LOG = LoggerFactory.getLogger(HdfsUnderFileSystem.class);
  protected static final int MAX_TRY = 5;
  protected static final String HDFS_USER = "";
  /** Name of the class for the HDFS Acl provider. */
  protected static final String HDFS_ACL_PROVIDER_CLASS =
      "alluxio.underfs.hdfs.acl.SupportedHdfsAclProvider";

  /** The minimum HDFS production version required for EC. **/
  protected static final String HDFS_EC_MIN_VERSION = "3.0.0";

  /** Name of the class for the HDFS EC Codec Registry. **/
  protected static final String HDFS_EC_CODEC_REGISTRY_CLASS =
      "org.apache.hadoop.io.erasurecode.CodecRegistry";

  protected static final String JAVAX_WS_RS_CORE_MEDIA_TYPE =
      "javax.ws.rs.core.MediaType";

  protected static final String HADOOP_AUTH_METHOD =
          "hadoop.security.authentication";

  protected static final String KRB5_CONF_FILE =
          "java.security.krb5.conf";

  protected static final String KRB_KEYTAB_LOGIN_AUTO_RENEW =
          "hadoop.kerberos.keytab.login.autorenewal.enabled";

  protected static final String CHECKSUM_COMBINE_MODE =
          "dfs.checksum.combine.mode";

  protected static final String USER_NAMESPACE_PREFIX = "user.";

  private final LoadingCache<String, FileSystem> mUserFs;
  protected final HdfsAclProvider mHdfsAclProvider;

  private final boolean mTrashEnable;
  private final LoadingCache<FileSystem, Trash> mFsTrash;

  /**
   * Factory method to constructs a new HDFS {@link UnderFileSystem} instance.
   *
   * @param ufsUri the {@link AlluxioURI} for this UFS
   * @param conf the configuration for Hadoop
   * @return a new HDFS {@link UnderFileSystem} instance
   */
  public static HdfsUnderFileSystem createInstance(AlluxioURI ufsUri,
      UnderFileSystemConfiguration conf) {
    Configuration hdfsConf = createConfiguration(conf);
    return new HdfsUnderFileSystem(ufsUri, conf, hdfsConf);
  }

  /**
   * Constructs a new HDFS {@link UnderFileSystem}.
   *
   * @param ufsUri the {@link AlluxioURI} for this UFS
   * @param conf the configuration for this UFS
   * @param hdfsConf the configuration for HDFS
   */
  public HdfsUnderFileSystem(AlluxioURI ufsUri, UnderFileSystemConfiguration conf,
      Configuration hdfsConf) {
    this(ufsUri, conf, hdfsConf, true);
  }

  protected HdfsUnderFileSystem(AlluxioURI ufsUri, UnderFileSystemConfiguration conf,
      Configuration hdfsConf, boolean useLoadingCache) {
    super(ufsUri, conf);

    mTrashEnable = alluxio.conf.Configuration.getBoolean(
        PropertyKey.UNDERFS_HDFS_TRASH_ENABLED);
    LOG.info(PropertyKey.UNDERFS_HDFS_TRASH_ENABLED.getName() + " is set to {}", mTrashEnable);
    mFsTrash = CacheBuilder.newBuilder().build(new CacheLoader<FileSystem, Trash>() {
      @Override
      public Trash load(FileSystem fs) throws Exception {
        return new Trash(fs, fs.getConf());
      }
    });

    // Create the supported HdfsAclProvider if possible.
    HdfsAclProvider hdfsAclProvider = new NoopHdfsAclProvider();
    try {
      // The HDFS acl provider class may not be available, so the class must be created from a
      // string literal.
      Object o = Class.forName(HDFS_ACL_PROVIDER_CLASS).newInstance();
      if (o instanceof HdfsAclProvider) {
        hdfsAclProvider = (HdfsAclProvider) o;
      } else {
        LOG.warn(
            "SupportedHdfsAclProvider is not instance of HdfsAclProvider. HDFS ACLs will not be "
                + "supported.");
      }
    } catch (Exception e) {
      // ignore
      LOG.warn("Cannot create SupportedHdfsAclProvider. HDFS ACLs is not supported, "
          + "Please upgrade to an HDFS version > 2.4 to enable support for ACL");
      LOG.debug("Exception:", e);
    }
    mHdfsAclProvider = hdfsAclProvider;

    Path path = new Path(ufsUri.toString());
    // UserGroupInformation.setConfiguration(hdfsConf) will trigger service loading.
    // Stash the classloader to prevent service loading throwing exception due to
    // classloader mismatch.
    ClassLoader currentClassLoader = Thread.currentThread().getContextClassLoader();
    try {
      Thread.currentThread().setContextClassLoader(hdfsConf.getClassLoader());
      if (mUfsConf.isSet(PropertyKey.HADOOP_SECURITY_AUTHENTICATION)) {
        hdfsConf.set(HADOOP_AUTH_METHOD,
                mUfsConf.getString(PropertyKey.HADOOP_SECURITY_AUTHENTICATION));
      }
      if (mUfsConf.isSet(PropertyKey.HADOOP_KRB5_CONF_FILE)) {
        System.setProperty(KRB5_CONF_FILE,
                mUfsConf.getString(PropertyKey.HADOOP_KRB5_CONF_FILE));
      }
      if (mUfsConf.isSet(PropertyKey.HADOOP_KERBEROS_KEYTAB_LOGIN_AUTORENEWAL)) {
        hdfsConf.setBoolean(KRB_KEYTAB_LOGIN_AUTO_RENEW,
                mUfsConf.getBoolean(PropertyKey.HADOOP_KERBEROS_KEYTAB_LOGIN_AUTORENEWAL));
      }
      // HDFS default composite type is MD5 of a concatenation of chunk CRCs so no need to set
      if (mUfsConf.getEnum(PropertyKey.UNDERFS_CHECKSUM_TYPE, ChecksumType.class)
                  .equals(ChecksumType.CRC32C)) {
        hdfsConf.set(CHECKSUM_COMBINE_MODE, "COMPOSITE_CRC");
      }

      // Set Hadoop UGI configuration to ensure UGI can be initialized by the shaded classes for
      // group service.
      UserGroupInformation.setConfiguration(hdfsConf);
      // When HDFS version is 3.0.0 or later, initialize HDFS EC CodecRegistry here to ensure
      // RawErasureCoderFactory implementations are loaded by the same classloader of hdfsConf.
      if (UfsConstants.UFS_HADOOP_VERSION.compareTo(HDFS_EC_MIN_VERSION) >= 0) {
        try {
          Class.forName(HDFS_EC_CODEC_REGISTRY_CLASS);
        } catch (ClassNotFoundException e) {
          LOG.warn("Cannot initialize HDFS EC CodecRegistry. "
              + "HDFS EC will not be supported:", e);
        }
      }

      if (StringUtils.equals("hdfs", ufsUri.getScheme())) {
        try {
          // If this class is not loaded here, it will be later loaded by the system classloader
          // from alluxio server jar, but will be called by a class loaded with extension
          // classloader.
          Class.forName(JAVAX_WS_RS_CORE_MEDIA_TYPE);
        } catch (ClassNotFoundException e) {
          LOG.warn("Cannot initialize javax.ws.rs.MediaType.", e);
        }
      }
    } finally {
      Thread.currentThread().setContextClassLoader(currentClassLoader);
    }

    if (useLoadingCache) {
      mUserFs = CacheBuilder.newBuilder().build(new CacheLoader<String, FileSystem>() {
        @Override
        public FileSystem load(String userKey) throws Exception {
          // When running {@link UnderFileSystemContractTest} with hdfs path,
          // the org.apache.hadoop.fs.FileSystem is loaded by {@link ExtensionClassLoader},
          // but the org.apache.hadoop.fs.LocalFileSystem is loaded by {@link AppClassLoader}.
          // When an interface and associated implementation are each loaded
          // by two separate class loaders, an instance of the class from one loader cannot
          // be recognized as implementing the interface from the other loader.
          ClassLoader previousClassLoader = Thread.currentThread().getContextClassLoader();
          try {
            // Set the class loader to ensure FileSystem implementations are
            // loaded by the same class loader to avoid ConfigurationError
            Thread.currentThread().setContextClassLoader(currentClassLoader);
            return path.getFileSystem(hdfsConf);
          } finally {
            Thread.currentThread().setContextClassLoader(previousClassLoader);
          }
        }
      });
    } else {
      mUserFs = null;
    }
  }

  @Override
  public String getUnderFSType() {
    return "hdfs";
  }

  /**
   * Prepares the Hadoop configuration necessary to successfully obtain a {@link FileSystem}
   * instance that can access the provided path.
   * <p>
   * Derived implementations that work with specialised Hadoop {@linkplain FileSystem} API
   * compatible implementations can override this method to add implementation specific
   * configuration necessary for obtaining a usable {@linkplain FileSystem} instance.
   * </p>
   *
   * @param conf the configuration for this UFS
   * @return the configuration for HDFS
   */
  public static Configuration createConfiguration(UnderFileSystemConfiguration conf) {
    Preconditions.checkNotNull(conf, "conf");
    Configuration hdfsConf = new Configuration();

    // Load HDFS site properties from the given file and overwrite the default HDFS conf,
    // the path of this file can be passed through --option
    for (String path : conf.getString(PropertyKey.UNDERFS_HDFS_CONFIGURATION).split(":")) {
      if (!path.isEmpty()) {
        hdfsConf.addResource(new Path(path));
      }
    }

    // On Hadoop 2.x this is strictly unnecessary since it uses ServiceLoader to automatically
    // discover available file system implementations. However this configuration setting is
    // required for earlier Hadoop versions plus it is still honoured as an override even in 2.x so
    // if present propagate it to the Hadoop configuration
    String ufsHdfsImpl = conf.getString(PropertyKey.UNDERFS_HDFS_IMPL);
    if (!StringUtils.isEmpty(ufsHdfsImpl)) {
      hdfsConf.set("fs.hdfs.impl", ufsHdfsImpl);
    }

    // Disable HDFS client caching so that input configuration is respected. Configurable from
    // system property
    hdfsConf.set("fs.hdfs.impl.disable.cache",
        System.getProperty("fs.hdfs.impl.disable.cache", "true"));

    // Set all parameters passed through --option
    for (Map.Entry<String, Object> entry : conf.getMountSpecificConf().entrySet()) {
      hdfsConf.set(entry.getKey(),
          entry.getValue() == null ? null : entry.getValue().toString());
    }
    return hdfsConf;
  }

  @Override
  public void cleanup() {
  }

  @Override
  public void close() {
    // Don't close; file systems are singletons and closing it here could break other users
  }

  @Override
  public OutputStream create(String path, CreateOptions options) throws IOException {
    if (!options.isEnsureAtomic()) {
      return createDirect(path, options);
    }
    return new AtomicFileOutputStream(path, this, options);
  }

  @Override
  public OutputStream createDirect(String path, CreateOptions options) throws IOException {
    IOException te = null;
    FileSystem hdfs = getFs();
    RetryPolicy retryPolicy = new CountingRetry(MAX_TRY);
    while (retryPolicy.attempt()) {
      try {
        // TODO(chaomin): support creating HDFS files with specified block size and replication.
        OutputStream outputStream = new HdfsUnderFileOutputStream(hdfs, path,
            FileSystem.create(hdfs, new Path(path),
              new FsPermission(options.getMode().toShort())));
        if (options.getAcl() != null) {
          setAclEntries(path, options.getAcl().getEntries());
        }
        return outputStream;
      } catch (IOException e) {
        LOG.warn("Attempt count {} : {} ", retryPolicy.getAttemptCount(), e.toString());
        te = e;
      }
    }
    throw te;
  }

  @Override
  public boolean deleteDirectory(String path, DeleteOptions options) throws IOException {
    return isDirectory(path) && delete(path, options.isRecursive(), true);
  }

  @Override
  public boolean deleteFile(String path) throws IOException {
    return isFile(path) && delete(path, false, false);
  }

  @Override
  public boolean exists(String path) throws IOException {
    FileSystem hdfs = getFs();
    return hdfs.exists(new Path(path));
  }

  @Override
  public Pair<AccessControlList, DefaultAccessControlList> getAclPair(String path)
      throws IOException {
    return mHdfsAclProvider.getAcl(getFs(), path);
  }

  @Override
  public void setAclEntries(String path, List<AclEntry> aclEntries) throws IOException {
    mHdfsAclProvider.setAclEntries(getFs(), path, aclEntries);
  }

  @Override
  public long getBlockSizeByte(String path) throws IOException {
    Path tPath = new Path(path);
    FileSystem hdfs = getFs();
    if (!hdfs.exists(tPath)) {
      throw new FileNotFoundException(path);
    }
    FileStatus fs = hdfs.getFileStatus(tPath);
    return fs.getBlockSize();
  }

  @Override
  public UfsDirectoryStatus getDirectoryStatus(String path) throws IOException {
    Path tPath = new Path(path);
    FileSystem hdfs = getFs();
    FileStatus fs = hdfs.getFileStatus(tPath);
    return new UfsDirectoryStatus(path, fs.getOwner(), fs.getGroup(),
        fs.getPermission().toShort(), fs.getModificationTime());
  }

  @Override
  public List<String> getFileLocations(String path) throws IOException {
    return getFileLocations(path, FileLocationOptions.defaults());
  }

  @Override
  @Nullable
  public List<String> getFileLocations(String path, FileLocationOptions options)
      throws IOException {
    // If the user has hinted the underlying storage nodes are not co-located with Alluxio workers.
    if (mUfsConf.getBoolean(PropertyKey.UNDERFS_HDFS_REMOTE)) {
      return null;
    }
    FileSystem hdfs = getFs();
    List<String> ret = new ArrayList<>();
    try {
      // The only usage of fileStatus is to get the path in getFileBlockLocations.
      // In HDFS 2, there is an API getFileBlockLocation(Path path, long offset, long len),
      // but in HDFS 1, the only API is
      // getFileBlockLocation(FileStatus stat, long offset, long len).
      // By constructing the file status manually, we can save one RPC call to getFileStatus.
      FileStatus fileStatus = new FileStatus(0L, false, 0, 0L,
          0L, 0L, null, null, null, new Path(path));
      BlockLocation[] bLocations =
          hdfs.getFileBlockLocations(fileStatus, options.getOffset(), 1);
      if (bLocations.length > 0) {
        String[] names = bLocations[0].getHosts();
        Collections.addAll(ret, names);
      }
    } catch (IOException e) {
      LOG.debug("Unable to get file location for {}", path, e);
    }
    return ret;
  }

  @Override
  public UfsFileStatus getFileStatus(String path, GetStatusOptions options) throws IOException {
    Path tPath = new Path(path);
    FileSystem hdfs = getFs();
    FileStatus fs = hdfs.getFileStatus(tPath);
    String contentHash;
    if (options.isIncludeRealContentHash()) {
      contentHash = Base64.encodeBase64String(hdfs.getFileChecksum(tPath).getBytes());
    }
    else {
      contentHash =
          UnderFileSystemUtils.approximateContentHash(fs.getLen(), fs.getModificationTime());
    }

    return new UfsFileStatus(path, contentHash, fs.getLen(), fs.getModificationTime(),
        fs.getOwner(), fs.getGroup(), fs.getPermission().toShort(), fs.getBlockSize());
  }

  @Override
  public long getSpace(String path, SpaceType type) throws IOException {
    // Ignoring the path given, will give information for entire cluster
    // as Alluxio can load/store data out of entire HDFS cluster
    FileSystem hdfs = getFs();
    long space = -1;
    if (hdfs instanceof DistributedFileSystem) {
      // Note that, getDiskStatus() is an API from Hadoop 1, deprecated by getStatus() from
      // Hadoop 2 and removed in Hadoop 3
      switch (type) {
        case SPACE_TOTAL:
          space = hdfs.getStatus().getCapacity();
          break;
        case SPACE_USED:
          space = hdfs.getStatus().getUsed();
          break;
        case SPACE_FREE:
          space = hdfs.getStatus().getRemaining();
          break;
        default:
          throw new IOException("Unknown space type: " + type);
      }
    }
    return space;
  }

  @Override
  public UfsStatus getStatus(String path, GetStatusOptions options) throws IOException {
    Path tPath = new Path(path);
    FileSystem hdfs = getFs();
    FileStatus fs = hdfs.getFileStatus(tPath);
    if (!fs.isDir()) {
      // Return file status.
      String contentHash;
      if (options.isIncludeRealContentHash()) {
        contentHash = Base64.encodeBase64String(hdfs.getFileChecksum(tPath).getBytes());
      }
      else {
        contentHash =
            UnderFileSystemUtils.approximateContentHash(fs.getLen(), fs.getModificationTime());
      }
      return new UfsFileStatus(path, contentHash, fs.getLen(), fs.getModificationTime(),
          fs.getOwner(), fs.getGroup(), fs.getPermission().toShort(), fs.getBlockSize());
    }
    // Return directory status.
    return new UfsDirectoryStatus(path, fs.getOwner(), fs.getGroup(), fs.getPermission().toShort(),
        fs.getModificationTime());
  }

  @Override
  public boolean isDirectory(String path) throws IOException {
    FileSystem hdfs = getFs();
    return hdfs.isDirectory(new Path(path));
  }

  @Override
  public boolean isFile(String path) throws IOException {
    FileSystem hdfs = getFs();
    return hdfs.isFile(new Path(path));
  }

  @Nullable
  @Override
  public Iterator<UfsStatus> listStatusIterable(String path, ListOptions options, String startAfter,
                                                int batchSize) throws IOException {
    FileSystem hdfs = getFs();
    return new HdfsUfsStatusIterator(path, hdfs);
  }

  @Override
  @Nullable
  public UfsStatus[] listStatus(String path) throws IOException {
    FileStatus[] files = listStatusInternal(path);
    if (files == null) {
      return null;
    }
    UfsStatus[] rtn = new UfsStatus[files.length];
    int i = 0;
    for (FileStatus status : files) {
      // only return the relative path, to keep consistent with java.io.File.list()
      UfsStatus retStatus;
      if (!status.isDir()) {
        String contentHash = UnderFileSystemUtils
            .approximateContentHash(status.getLen(), status.getModificationTime());
        retStatus = new UfsFileStatus(status.getPath().getName(), contentHash, status.getLen(),
            status.getModificationTime(), status.getOwner(), status.getGroup(),
            status.getPermission().toShort(), status.getBlockSize());
      } else {
        retStatus = new UfsDirectoryStatus(status.getPath().getName(), status.getOwner(),
            status.getGroup(), status.getPermission().toShort(), status.getModificationTime());
      }
      rtn[i++] = retStatus;
    }
    return rtn;
  }

  @Override
  public void connectFromMaster(String host) throws IOException {
    if (!mUfsConf.isSet(PropertyKey.MASTER_KEYTAB_KEY_FILE)
        || !mUfsConf.isSet(PropertyKey.MASTER_PRINCIPAL)) {
      return;
    }
    String masterKeytab = mUfsConf.getString(PropertyKey.MASTER_KEYTAB_KEY_FILE);
    String masterPrincipal = mUfsConf.getString(PropertyKey.MASTER_PRINCIPAL);

    login(PropertyKey.MASTER_KEYTAB_KEY_FILE, masterKeytab, PropertyKey.MASTER_PRINCIPAL,
        masterPrincipal, host);
  }

  @Override
  public void connectFromWorker(String host) throws IOException {
    if (!mUfsConf.isSet(PropertyKey.WORKER_KEYTAB_FILE)
        || !mUfsConf.isSet(PropertyKey.WORKER_PRINCIPAL)) {
      return;
    }
    String workerKeytab = mUfsConf.getString(PropertyKey.WORKER_KEYTAB_FILE);
    String workerPrincipal = mUfsConf.getString(PropertyKey.WORKER_PRINCIPAL);

    login(PropertyKey.WORKER_KEYTAB_FILE, workerKeytab, PropertyKey.WORKER_PRINCIPAL,
        workerPrincipal, host);
  }

  private void login(PropertyKey keytabFileKey, String keytabFile, PropertyKey principalKey,
      String principal, String hostname) throws IOException {
    org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
    conf.set(keytabFileKey.toString(), keytabFile);
    conf.set(principalKey.toString(), principal);
    SecurityUtil.login(conf, keytabFileKey.toString(), principalKey.toString(), hostname);
  }

  @Override
  public boolean mkdirs(String path, MkdirsOptions options) throws IOException {
    IOException te = null;
    FileSystem hdfs = getFs();
    RetryPolicy retryPolicy = new CountingRetry(MAX_TRY);
    while (retryPolicy.attempt()) {
      try {
        Path hdfsPath = new Path(path);
        if (hdfs.exists(hdfsPath)) {
          LOG.debug("Trying to create existing directory at {}", path);
          return false;
        }
        if (!hdfs.mkdirs(hdfsPath, new FsPermission(options.getMode().toShort()))) {
          return false;
        }
        // Set the owner to the Alluxio client user to achieve permission delegation.
        // Alluxio server-side user is required to be a HDFS superuser. If it fails to set owner,
        // proceeds with mkdirs and print out an warning message.
        try {
          setOwner(hdfsPath.toString(), options.getOwner(), options.getGroup());
        } catch (IOException e) {
          LOG.warn("Failed to update the ufs dir ownership, default values will be used. " + e);
        }
        return true;
      } catch (IOException e) {
        LOG.debug("{} try to make directory for {} : {}", retryPolicy.getAttemptCount(), path,
            e.toString());
        te = e;
      }
    }
    throw te;
  }

  private boolean isReadLocal(FileSystem fs, Path filePath, OpenOptions options) {
    String localHost = NetworkAddressUtils.getLocalHostName((int) mUfsConf
        .getMs(PropertyKey.NETWORK_HOST_RESOLUTION_TIMEOUT_MS));
    BlockLocation[] blockLocations;
    try {
      blockLocations = fs.getFileBlockLocations(filePath,
          options.getOffset(), options.getLength());
      if (blockLocations == null) {
        // no blocks exist
        return true;
      }

      for (BlockLocation loc : blockLocations) {
        if (Arrays.stream(loc.getHosts()).noneMatch(localHost::equals)) {
          return false;
        }
      }
    } catch (IOException e) {
      return true;
    }
    return true;
  }

  @Override
  public InputStream open(String path, OpenOptions options) throws IOException {
    IOException te = null;
    FileSystem hdfs = getFs();
    RetryPolicy retryPolicy = new CountingRetry(MAX_TRY);
    DistributedFileSystem dfs = null;
    if (hdfs instanceof DistributedFileSystem) {
      dfs = (DistributedFileSystem) hdfs;
    }
    Path filePath = new Path(path);
    boolean remote = options.getPositionShort()
        || mUfsConf.getBoolean(PropertyKey.UNDERFS_HDFS_REMOTE)
        || !isReadLocal(hdfs, filePath, options);
    while (retryPolicy.attempt()) {
      try {
        FSDataInputStream inputStream = hdfs.open(filePath);
        if (remote) {
          LOG.debug("Using pread API to HDFS");
          // pread API instead of seek is more efficient for FSDataInputStream.
          // A seek on FSDataInputStream uses a skip op which is implemented as read + discard
          // and hence ends up reading extra data from the datanode.
          return new HdfsPositionedUnderFileInputStream(inputStream, options.getOffset());
        }
        try {
          inputStream.seek(options.getOffset());
        } catch (IOException e) {
          inputStream.close();
          throw e;
        }
        LOG.debug("Using original API to HDFS");
        return new HdfsUnderFileInputStream(inputStream);
      } catch (IOException e) {
        LOG.debug("{} try to open {} : {}", retryPolicy.getAttemptCount(), path, e.toString());
        te = e;
        if (options.getRecoverFailedOpen() && dfs != null && e.getMessage().toLowerCase()
            .startsWith("cannot obtain block length for")) {
          // This error can occur when an Alluxio journal file was not properly closed by Alluxio.
          // In this scenario, the HDFS lease must be recovered in order for the file to be
          // readable again. The 'recoverLease' API usually needs to be invoked multiple times
          // to complete the lease recovery process.
          try {
            if (dfs.recoverLease(new Path(path))) {
              LOG.warn("HDFS recoverLease-1 success for: {}", path);
            } else {
              // try one more time, after waiting
              CommonUtils.sleepMs(5L * Constants.SECOND_MS);
              if (dfs.recoverLease(new Path(path))) {
                LOG.warn("HDFS recoverLease-2 success for: {}", path);
              } else {
                LOG.warn("HDFS recoverLease: path not closed: {}", path);
              }
            }
          } catch (IOException e1) {
            // ignore exception
            LOG.warn("HDFS recoverLease failed for: {} error: {}", path, e1.getMessage());
          }
        }
      }
    }
    if (te != null) {
      LOG.error("{} failed attempts to open \"{}\" with last error:",
          retryPolicy.getAttemptCount(), path, te);
      throw te;
    }
    throw new IllegalStateException("Exceeded the number of retry attempts with no exception");
  }

  @Override
  public HdfsPositionedUnderFileInputStream openPositionRead(String path, long fileLength) {
    try {
      FSDataInputStream inputStream = getFs().open(new Path(path));
      return new HdfsPositionedUnderFileInputStream(inputStream, 0);
    } catch (IOException e) {
      throw AlluxioHdfsException.from(e);
    }
  }

  @Override
  public boolean renameDirectory(String src, String dst) throws IOException {
    if (!isDirectory(src)) {
      LOG.warn("Unable to rename {} to {} because source does not exist or is a file", src, dst);
      return false;
    }
    return rename(src, dst);
  }

  @Override
  public boolean renameFile(String src, String dst) throws IOException {
    if (!isFile(src)) {
      LOG.warn("Unable to rename {} to {} because source does not exist or is a directory", src,
          dst);
      return false;
    }
    return rename(src, dst);
  }

  @Override
  public void setOwner(String path, String user, String group) throws IOException {
    if (user == null && group == null) {
      return;
    }
    FileSystem hdfs = getFs();
    try {
      FileStatus fileStatus = hdfs.getFileStatus(new Path(path));
      hdfs.setOwner(fileStatus.getPath(), user, group);
    } catch (IOException e) {
      LOG.debug("Exception: ", e);
      if (!mUfsConf.getBoolean(PropertyKey.UNDERFS_ALLOW_SET_OWNER_FAILURE)) {
        LOG.warn("Failed to set owner for {} with user: {}, group: {}: {}. "
            + "Running Alluxio as superuser is required to modify ownership of local files",
            path, user, group, e.toString());
        throw e;
      } else {
        LOG.warn("Failed to set owner for {} with user: {}, group: {}: {}. "
            + "This failure is ignored but may cause permission inconsistency between Alluxio "
            + "and local under file system", path, user, group, e.toString());
      }
    }
  }

  @Override
  public void setAttribute(String path, String name, byte[] value) throws IOException {
    if (StringUtils.isEmpty(name)) {
      LOG.warn("Try to set Xattr to an empty file name for path {}", path);
      return;
    }
    FileSystem hdfs = getFs();
    try {
      FileStatus fileStatus = hdfs.getFileStatus(new Path(path));
      hdfs.setXAttr(fileStatus.getPath(), USER_NAMESPACE_PREFIX + name, value);
    } catch (IOException e) {
      LOG.warn("Failed to set XAttr for {} with name: {}, value: {}: {}. "
              + "Running Alluxio as superuser is required to modify attributes of local files",
          path, name, new String(value), e.toString());
      throw e;
    }
  }

  @Override
  public Map<String, String> getAttributes(String path) throws IOException {
    FileSystem hdfs = getFs();
    try {
      Map<String, byte[]> attrMap = hdfs.getXAttrs(new Path(path));
      Map<String, String> resMap = new HashMap<>(attrMap.size());
      attrMap.entrySet().stream().filter(entry -> entry.getKey().startsWith(USER_NAMESPACE_PREFIX))
          .forEach(entry -> resMap.put(entry.getKey().substring(entry.getKey().indexOf(".") + 1),
              new String(entry.getValue())));
      return Collections.unmodifiableMap(resMap);
    } catch (IOException e) {
      LOG.warn("Failed to get XAttr for {}.", path);
      throw e;
    }
  }

  @Override
  public void setMode(String path, short mode) throws IOException {
    FileSystem hdfs = getFs();
    try {
      FileStatus fileStatus = hdfs.getFileStatus(new Path(path));
      hdfs.setPermission(fileStatus.getPath(), new FsPermission(mode));
    } catch (IOException e) {
      LOG.warn("Fail to set permission for {} with perm {} : {}", path, mode, e.toString());
      throw e;
    }
  }

  @Override
  public boolean supportsFlush() {
    return true;
  }

  /**
   * Delete a file or directory at path.
   *
   * @param path file or directory path
   * @param recursive whether to delete path recursively
   * @param isDirectory whether path is a directory
   * @return true, if succeed
   */
  private boolean delete(String path, boolean recursive, boolean isDirectory) throws IOException {
    IOException te = null;
    FileSystem hdfs = getFs();
    RetryPolicy retryPolicy = new CountingRetry(MAX_TRY);
    while (retryPolicy.attempt()) {
      try {
        Path hdfsPath = new Path(path);
        if (!mTrashEnable) {
          return hdfs.delete(hdfsPath, recursive);
        }
        Trash trash = getTrash(hdfs);
        // move to trash
        if (isDirectory && !recursive && hdfs.listStatus(hdfsPath).length != 0) {
          return false;
        }
        if (trash.moveToTrash(hdfsPath)) {
          // moving file to trash succeeded.
          return true;
        } else {
          // if failed to move this file to trash, delete it.
          LOG.debug("Failed to move '{}' to trash. Now delete it.", path);
          return hdfs.delete(hdfsPath, recursive);
        }
      } catch (IOException e) {
        LOG.warn("Attempt count {} : {}", retryPolicy.getAttemptCount(), e.toString());
        te = e;
      }
    }
    throw te;
  }

  /**
   * List status for given path. Returns an array of {@link FileStatus} with an entry for each file
   * and directory in the directory denoted by this path.
   *
   * @param path the pathname to list
   * @return {@code null} if the path is not a directory
   */
  @Nullable
  private FileStatus[] listStatusInternal(String path) throws IOException {
    FileStatus[] files;
    FileSystem hdfs = getFs();
    Path thePath = new Path(path);
    try {
      files = hdfs.listStatus(thePath);
    } catch (FileNotFoundException e) {
      return null;
    }
    // Check if path is a file
    if (files != null && files.length == 1
        && files[0].getPath().toUri().getPath().equals(thePath.toUri().getPath())) {
      return null;
    }
    return files;
  }

  /**
   * Rename a file or folder to a file or folder.
   *
   * @param src path of source file or directory
   * @param dst path of destination file or directory
   * @return true if rename succeeds
   */
  private boolean rename(String src, String dst) throws IOException {
    IOException te = null;
    FileSystem hdfs = getFs();
    RetryPolicy retryPolicy = new CountingRetry(MAX_TRY);
    while (retryPolicy.attempt()) {
      try {
        return hdfs.rename(new Path(src), new Path(dst));
      } catch (IOException e) {
        LOG.warn("{} try to rename {} to {} : {}", retryPolicy.getAttemptCount(), src, dst,
            e.toString());
        te = e;
      }
    }
    throw te;
  }

  @Override
  public boolean isSeekable() {
    return true;
  }

  /**
   * @return the underlying HDFS {@link FileSystem} object
   */
  @VisibleForTesting
  public FileSystem getFs() throws IOException {
    try {
      // TODO(gpang): handle different users
      return mUserFs.get(HDFS_USER);
    } catch (ExecutionException e) {
      throw new IOException("Failed get FileSystem for " + mUri, e.getCause());
    }
  }

  private Trash getTrash(FileSystem fs) throws IOException {
    try {
      return mFsTrash.get(fs);
    } catch (ExecutionException e) {
      throw new IOException("Failed get Trash for " + fs.getUri(), e.getCause());
    }
  }
}
