/**
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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.fs.cosn;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.HashMap;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.util.concurrent.ListeningExecutorService;

import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BufferedFSInputStream;
import org.apache.hadoop.fs.CreateFlag;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileAlreadyExistsException;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathIOException;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.hadoop.io.retry.RetryProxy;
import org.apache.hadoop.util.BlockingThreadPoolExecutorService;
import org.apache.hadoop.util.Progressable;

/**
 * The core CosN Filesystem implementation.
 */
@InterfaceAudience.Private
@InterfaceStability.Stable
public class CosNFileSystem extends FileSystem {
  static final Logger LOG = LoggerFactory.getLogger(CosNFileSystem.class);

  public static final String SCHEME = "cosn";
  public static final String PATH_DELIMITER = Path.SEPARATOR;

  private URI uri;
  private String bucket;
  private NativeFileSystemStore store;
  private Path workingDir;
  private String owner = "Unknown";
  private String group = "Unknown";

  private ListeningExecutorService boundedIOThreadPool;
  private ListeningExecutorService boundedCopyThreadPool;

  public CosNFileSystem() {
  }

  public CosNFileSystem(NativeFileSystemStore store) {
    this.store = store;
  }

  /**
   * Return the protocol scheme for the FileSystem.
   *
   * @return <code>cosn</code>
   */
  @Override
  public String getScheme() {
    return CosNFileSystem.SCHEME;
  }

  @Override
  public void initialize(URI name, Configuration conf) throws IOException {
    super.initialize(name, conf);
    this.bucket = name.getHost();
    if (this.store == null) {
      this.store = createDefaultStore(conf);
    }
    this.store.initialize(name, conf);
    setConf(conf);
    this.uri = URI.create(name.getScheme() + "://" + name.getAuthority());
    this.workingDir = new Path("/user",
        System.getProperty("user.name")).makeQualified(
        this.uri,
        this.getWorkingDirectory());
    this.owner = getOwnerId();
    this.group = getGroupId();
    LOG.debug("owner:" + owner + ", group:" + group);

    BufferPool.getInstance().initialize(this.getConf());

    // initialize the thread pool
    int uploadThreadPoolSize = this.getConf().getInt(
        CosNConfigKeys.UPLOAD_THREAD_POOL_SIZE_KEY,
        CosNConfigKeys.DEFAULT_UPLOAD_THREAD_POOL_SIZE
    );
    int readAheadPoolSize = this.getConf().getInt(
        CosNConfigKeys.READ_AHEAD_QUEUE_SIZE,
        CosNConfigKeys.DEFAULT_READ_AHEAD_QUEUE_SIZE
    );
    int ioThreadPoolSize = uploadThreadPoolSize + readAheadPoolSize / 3;
    long threadKeepAlive = this.getConf().getLong(
        CosNConfigKeys.THREAD_KEEP_ALIVE_TIME_KEY,
        CosNConfigKeys.DEFAULT_THREAD_KEEP_ALIVE_TIME
    );
    this.boundedIOThreadPool = BlockingThreadPoolExecutorService.newInstance(
        ioThreadPoolSize / 2, ioThreadPoolSize,
        threadKeepAlive, TimeUnit.SECONDS,
        "cos-transfer-thread-pool");
    int copyThreadPoolSize = this.getConf().getInt(
        CosNConfigKeys.COPY_THREAD_POOL_SIZE_KEY,
        CosNConfigKeys.DEFAULT_COPY_THREAD_POOL_SIZE
    );
    this.boundedCopyThreadPool = BlockingThreadPoolExecutorService.newInstance(
        CosNConfigKeys.DEFAULT_COPY_THREAD_POOL_SIZE, copyThreadPoolSize,
        60L, TimeUnit.SECONDS,
        "cos-copy-thread-pool");
  }

  private static NativeFileSystemStore createDefaultStore(Configuration conf) {
    NativeFileSystemStore store = new CosNativeFileSystemStore();
    RetryPolicy basePolicy = RetryPolicies.retryUpToMaximumCountWithFixedSleep(
        conf.getInt(CosNConfigKeys.COSN_MAX_RETRIES_KEY,
            CosNConfigKeys.DEFAULT_MAX_RETRIES),
        conf.getLong(CosNConfigKeys.COSN_RETRY_INTERVAL_KEY,
            CosNConfigKeys.DEFAULT_RETRY_INTERVAL),
        TimeUnit.SECONDS);
    Map<Class<? extends Exception>, RetryPolicy> exceptionToPolicyMap =
        new HashMap<>();

    exceptionToPolicyMap.put(IOException.class, basePolicy);
    RetryPolicy methodPolicy = RetryPolicies.retryByException(
        RetryPolicies.TRY_ONCE_THEN_FAIL,
        exceptionToPolicyMap);
    Map<String, RetryPolicy> methodNameToPolicyMap = new HashMap<>();
    methodNameToPolicyMap.put("storeFile", methodPolicy);
    methodNameToPolicyMap.put("rename", methodPolicy);

    return (NativeFileSystemStore) RetryProxy.create(
        NativeFileSystemStore.class, store, methodNameToPolicyMap);
  }

  private String getOwnerId() {
    return System.getProperty("user.name");
  }

  private String getGroupId() {
    return System.getProperty("user.name");
  }

  private String getOwnerInfo(boolean getOwnerId) {
    String ownerInfoId = "";
    try {
      String userName = System.getProperty("user.name");
      String command = "id -u " + userName;
      if (!getOwnerId) {
        command = "id -g " + userName;
      }
      Process child = Runtime.getRuntime().exec(command);
      child.waitFor();

      // Get the input stream and read from it
      InputStream in = child.getInputStream();
      StringBuilder strBuffer = new StringBuilder();
      int c;
      while ((c = in.read()) != -1) {
        strBuffer.append((char) c);
      }
      in.close();
      ownerInfoId = strBuffer.toString();
    } catch (IOException | InterruptedException e) {
      LOG.error("Getting owner info occurs a exception", e);
    }
    return ownerInfoId;
  }

  private static String pathToKey(Path path) {
    if (path.toUri().getScheme() != null && path.toUri().getPath().isEmpty()) {
      // allow uris without trailing slash after bucket to refer to root,
      // like cosn://mybucket
      return "";
    }
    if (!path.isAbsolute()) {
      throw new IllegalArgumentException("Path must be absolute: " + path);
    }
    String ret = path.toUri().getPath();
    if (ret.endsWith("/") && (ret.indexOf("/") != ret.length() - 1)) {
      ret = ret.substring(0, ret.length() - 1);
    }
    return ret;
  }

  private static Path keyToPath(String key) {
    if (!key.startsWith(PATH_DELIMITER)) {
      return new Path("/" + key);
    } else {
      return new Path(key);
    }
  }

  private Path makeAbsolute(Path path) {
    if (path.isAbsolute()) {
      return path;
    }
    return new Path(workingDir, path);
  }

  /**
   * This optional operation is not yet supported.
   */
  @Override
  public FSDataOutputStream append(Path f, int bufferSize,
      Progressable progress) throws IOException {
    throw new IOException("Not supported");
  }

  @Override
  public FSDataOutputStream create(Path f, FsPermission permission,
      boolean overwrite, int bufferSize, short replication, long blockSize,
      Progressable progress) throws IOException {
    FileStatus fileStatus;

    try {
      fileStatus = getFileStatus(f);
      if (fileStatus.isDirectory()) {
        throw new FileAlreadyExistsException(f + " is a directory");
      }
      if (!overwrite) {
        // path references a file and overwrite is disabled
        throw new FileAlreadyExistsException(f + " already exists");
      }

    } catch (FileNotFoundException e) {
      LOG.debug("Creating a new file: [{}] in COS.", f);
    }

    Path absolutePath = makeAbsolute(f);
    String key = pathToKey(absolutePath);
    return new FSDataOutputStream(
        new CosNOutputStream(getConf(), store, key, blockSize,
            this.boundedIOThreadPool), statistics);
  }

  private boolean rejectRootDirectoryDelete(boolean isEmptyDir,
      boolean recursive) throws PathIOException {
    if (isEmptyDir) {
      return true;
    }
    if (recursive) {
      return false;
    } else {
      throw new PathIOException(this.bucket, "Can not delete root path");
    }
  }

  @Override
  public FSDataOutputStream createNonRecursive(Path f, FsPermission permission,
      EnumSet<CreateFlag> flags, int bufferSize, short replication,
      long blockSize, Progressable progress) throws IOException {
    Path parent = f.getParent();
    if (null != parent) {
      if (!getFileStatus(parent).isDirectory()) {
        throw new FileAlreadyExistsException("Not a directory: " + parent);
      }
    }

    return create(f, permission, flags.contains(CreateFlag.OVERWRITE),
        bufferSize, replication, blockSize, progress);
  }

  @Override
  public boolean delete(Path f, boolean recursive) throws IOException {
    LOG.debug("Ready to delete path: [{}]. recursive: [{}].", f, recursive);
    FileStatus status;
    try {
      status = getFileStatus(f);
    } catch (FileNotFoundException e) {
      LOG.debug("Ready to delete the file: [{}], but it does not exist.", f);
      return false;
    }
    Path absolutePath = makeAbsolute(f);
    String key = pathToKey(absolutePath);
    if (key.compareToIgnoreCase("/") == 0) {
      FileStatus[] fileStatuses = listStatus(f);
      return this.rejectRootDirectoryDelete(
          fileStatuses.length == 0, recursive);
    }

    if (status.isDirectory()) {
      if (!key.endsWith(PATH_DELIMITER)) {
        key += PATH_DELIMITER;
      }
      if (!recursive && listStatus(f).length > 0) {
        String errMsg = String.format("Can not delete the directory: [%s], as"
            + " it is not empty and option recursive is false.", f);
        throw new IOException(errMsg);
      }

      createParent(f);

      String priorLastKey = null;
      do {
        PartialListing listing = store.list(
            key,
            Constants.COS_MAX_LISTING_LENGTH,
            priorLastKey,
            true);
        for (FileMetadata file : listing.getFiles()) {
          store.delete(file.getKey());
        }
        for (FileMetadata commonPrefix : listing.getCommonPrefixes()) {
          store.delete(commonPrefix.getKey());
        }
        priorLastKey = listing.getPriorLastKey();
      } while (priorLastKey != null);
      try {
        store.delete(key);
      } catch (Exception e) {
        LOG.error("Deleting the COS key: [{}] occurs an exception.", key, e);
      }

    } else {
      LOG.debug("Delete the file: {}", f);
      createParent(f);
      store.delete(key);
    }
    return true;
  }

  @Override
  public FileStatus getFileStatus(Path f) throws IOException {
    Path absolutePath = makeAbsolute(f);
    String key = pathToKey(absolutePath);

    if (key.length() == 0) {
      // root always exists
      return newDirectory(absolutePath);
    }

    LOG.debug("Call the getFileStatus to obtain the metadata for "
        + "the file: [{}].", f);

    FileMetadata meta = store.retrieveMetadata(key);
    if (meta != null) {
      if (meta.isFile()) {
        LOG.debug("Path: [{}] is a file. COS key: [{}]", f, key);
        return newFile(meta, absolutePath);
      } else {
        LOG.debug("Path: [{}] is a dir. COS key: [{}]", f, key);
        return newDirectory(meta, absolutePath);
      }
    }

    if (!key.endsWith(PATH_DELIMITER)) {
      key += PATH_DELIMITER;
    }

    // Considering that the object store's directory is a common prefix in
    // the object key, it needs to check the existence of the path by listing
    // the COS key.
    LOG.debug("List COS key: [{}] to check the existence of the path.", key);
    PartialListing listing = store.list(key, 1);
    if (listing.getFiles().length > 0
        || listing.getCommonPrefixes().length > 0) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Path: [{}] is a directory. COS key: [{}]", f, key);
      }
      return newDirectory(absolutePath);
    }

    throw new FileNotFoundException(
        "No such file or directory '" + absolutePath + "'");
  }

  @Override
  public URI getUri() {
    return uri;
  }

  /**
   * <p>
   * If <code>f</code> is a file, this method will make a single call to COS.
   * If <code>f</code> is a directory,
   * this method will make a maximum of ( <i>n</i> / 199) + 2 calls to cos,
   * where <i>n</i> is the total number of files
   * and directories contained directly in <code>f</code>.
   * </p>
   */
  @Override
  public FileStatus[] listStatus(Path f) throws IOException {
    Path absolutePath = makeAbsolute(f);
    String key = pathToKey(absolutePath);

    if (key.length() > 0) {
      FileStatus fileStatus = this.getFileStatus(f);
      if (fileStatus.isFile()) {
        return new FileStatus[]{fileStatus};
      }
    }

    if (!key.endsWith(PATH_DELIMITER)) {
      key += PATH_DELIMITER;
    }

    URI pathUri = absolutePath.toUri();
    Set<FileStatus> status = new TreeSet<>();
    String priorLastKey = null;
    do {
      PartialListing listing = store.list(
          key, Constants.COS_MAX_LISTING_LENGTH, priorLastKey, false);
      for (FileMetadata fileMetadata : listing.getFiles()) {
        Path subPath = keyToPath(fileMetadata.getKey());
        if (fileMetadata.getKey().equals(key)) {
          // this is just the directory we have been asked to list.
          LOG.debug("The file list contains the COS key [{}] to be listed.",
              key);
        } else {
          status.add(newFile(fileMetadata, subPath));
        }
      }

      for (FileMetadata commonPrefix : listing.getCommonPrefixes()) {
        Path subPath = keyToPath(commonPrefix.getKey());
        String relativePath = pathUri.relativize(subPath.toUri()).getPath();
        status.add(
            newDirectory(commonPrefix, new Path(absolutePath, relativePath)));
      }
      priorLastKey = listing.getPriorLastKey();
    } while (priorLastKey != null);

    return status.toArray(new FileStatus[status.size()]);
  }

  private FileStatus newFile(FileMetadata meta, Path path) {
    return new FileStatus(meta.getLength(), false, 1, getDefaultBlockSize(),
        meta.getLastModified(), 0, null, this.owner, this.group,
        path.makeQualified(this.getUri(), this.getWorkingDirectory()));
  }

  private FileStatus newDirectory(Path path) {
    return new FileStatus(0, true, 1, 0, 0, 0, null, this.owner, this.group,
        path.makeQualified(this.getUri(), this.getWorkingDirectory()));
  }

  private FileStatus newDirectory(FileMetadata meta, Path path) {
    if (meta == null) {
      return newDirectory(path);
    }
    return new FileStatus(0, true, 1, 0, meta.getLastModified(),
        0, null, this.owner, this.group,
        path.makeQualified(this.getUri(), this.getWorkingDirectory()));
  }

  /**
   * Validate the path from the bottom up.
   *
   * @param path The path to be validated
   * @throws FileAlreadyExistsException The specified path is an existing file
   * @throws IOException                Getting the file status of the
   *                                    specified path occurs
   *                                    an IOException.
   */
  private void validatePath(Path path) throws IOException {
    Path parent = path.getParent();
    do {
      try {
        FileStatus fileStatus = getFileStatus(parent);
        if (fileStatus.isDirectory()) {
          break;
        } else {
          throw new FileAlreadyExistsException(String.format(
              "Can't make directory for path '%s', it is a file.", parent));
        }
      } catch (FileNotFoundException e) {
        LOG.debug("The Path: [{}] does not exist.", path);
      }
      parent = parent.getParent();
    } while (parent != null);
  }

  @Override
  public boolean mkdirs(Path f, FsPermission permission) throws IOException {
    try {
      FileStatus fileStatus = getFileStatus(f);
      if (fileStatus.isDirectory()) {
        return true;
      } else {
        throw new FileAlreadyExistsException("Path is a file: " + f);
      }
    } catch (FileNotFoundException e) {
      validatePath(f);
    }

    return mkDirRecursively(f, permission);
  }

  /**
   * Recursively create a directory.
   *
   * @param f          Absolute path to the directory.
   * @param permission Directory permissions. Permission does not work for
   *                   the CosN filesystem currently.
   * @return Return true if the creation was successful,  throw a IOException.
   * @throws IOException The specified path already exists or an error
   *                     creating the path.
   */
  public boolean mkDirRecursively(Path f, FsPermission permission)
      throws IOException {
    Path absolutePath = makeAbsolute(f);
    List<Path> paths = new ArrayList<>();
    do {
      paths.add(absolutePath);
      absolutePath = absolutePath.getParent();
    } while (absolutePath != null);

    for (Path path : paths) {
      if (path.equals(new Path(CosNFileSystem.PATH_DELIMITER))) {
        break;
      }
      try {
        FileStatus fileStatus = getFileStatus(path);
        if (fileStatus.isFile()) {
          throw new FileAlreadyExistsException(
              String.format("Can't make directory for path: %s, "
                  + "since it is a file.", f));
        }
        if (fileStatus.isDirectory()) {
          break;
        }
      } catch (FileNotFoundException e) {
        LOG.debug("Making dir: [{}] in COS", f);

        String folderPath = pathToKey(makeAbsolute(f));
        if (!folderPath.endsWith(PATH_DELIMITER)) {
          folderPath += PATH_DELIMITER;
        }
        store.storeEmptyFile(folderPath);
      }
    }
    return true;
  }

  private boolean mkdir(Path f) throws IOException {
    try {
      FileStatus fileStatus = getFileStatus(f);
      if (fileStatus.isFile()) {
        throw new FileAlreadyExistsException(
            String.format(
                "Can't make directory for path '%s' since it is a file.", f));
      }
    } catch (FileNotFoundException e) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Make directory: [{}] in COS.", f);
      }

      String folderPath = pathToKey(makeAbsolute(f));
      if (!folderPath.endsWith(PATH_DELIMITER)) {
        folderPath += PATH_DELIMITER;
      }
      store.storeEmptyFile(folderPath);
    }
    return true;
  }

  @Override
  public FSDataInputStream open(Path f, int bufferSize) throws IOException {
    FileStatus fs = getFileStatus(f); // will throw if the file doesn't
    // exist
    if (fs.isDirectory()) {
      throw new FileNotFoundException("'" + f + "' is a directory");
    }
    LOG.info("Open the file: [{}] for reading.", f);
    Path absolutePath = makeAbsolute(f);
    String key = pathToKey(absolutePath);
    long fileSize = store.getFileLength(key);
    return new FSDataInputStream(new BufferedFSInputStream(
        new CosNInputStream(this.getConf(), store, statistics, key, fileSize,
            this.boundedIOThreadPool), bufferSize));
  }

  @Override
  public boolean rename(Path src, Path dst) throws IOException {
    LOG.debug("Rename source path: [{}] to dest path: [{}].", src, dst);

    // Renaming the root directory is not allowed
    if (src.isRoot()) {
      LOG.debug("Cannot rename the root directory of a filesystem.");
      return false;
    }

    // check the source path whether exists or not
    FileStatus srcFileStatus = this.getFileStatus(src);

    // Source path and destination path are not allowed to be the same
    if (src.equals(dst)) {
      LOG.debug("Source path and dest path refer to "
          + "the same file or directory: [{}].", dst);
      throw new IOException("Source path and dest path refer "
          + "the same file or directory");
    }

    // It is not allowed to rename a parent directory to its subdirectory
    Path dstParentPath;
    for (dstParentPath = dst.getParent();
         null != dstParentPath && !src.equals(dstParentPath);
         dstParentPath = dstParentPath.getParent()) {
      // Recursively find the common parent path of the source and
      // destination paths.
      LOG.debug("Recursively find the common parent directory of the source "
              + "and destination paths. The currently found parent path: {}",
          dstParentPath);
    }

    if (null != dstParentPath) {
      LOG.debug("It is not allowed to rename a parent directory:[{}] "
          + "to its subdirectory:[{}].", src, dst);
      throw new IOException(String.format(
          "It is not allowed to rename a parent directory: %s "
              + "to its subdirectory: %s", src, dst));
    }

    FileStatus dstFileStatus;
    try {
      dstFileStatus = this.getFileStatus(dst);

      // The destination path exists and is a file,
      // and the rename operation is not allowed.
      if (dstFileStatus.isFile()) {
        throw new FileAlreadyExistsException(String.format(
            "File: %s already exists", dstFileStatus.getPath()));
      } else {
        // The destination path is an existing directory,
        // and it is checked whether there is a file or directory
        // with the same name as the source path under the destination path
        dst = new Path(dst, src.getName());
        FileStatus[] statuses;
        try {
          statuses = this.listStatus(dst);
        } catch (FileNotFoundException e) {
          statuses = null;
        }
        if (null != statuses && statuses.length > 0) {
          LOG.debug("Cannot rename source file: [{}] to dest file: [{}], "
              + "because the file already exists.", src, dst);
          throw new FileAlreadyExistsException(
              String.format(
                  "File: %s already exists", dst
              )
          );
        }
      }
    } catch (FileNotFoundException e) {
      // destination path not exists
      Path tempDstParentPath = dst.getParent();
      FileStatus dstParentStatus = this.getFileStatus(tempDstParentPath);
      if (!dstParentStatus.isDirectory()) {
        throw new IOException(String.format(
            "Cannot rename %s to %s, %s is a file", src, dst, dst.getParent()
        ));
      }
      // The default root directory is definitely there.
    }

    boolean result;
    if (srcFileStatus.isDirectory()) {
      result = this.copyDirectory(src, dst);
    } else {
      result = this.copyFile(src, dst);
    }

    if (!result) {
      //Since rename is a non-atomic operation, after copy fails,
      // it is not allowed to delete the data of the original path.
      return false;
    } else {
      return this.delete(src, true);
    }
  }

  private boolean copyFile(Path srcPath, Path dstPath) throws IOException {
    String srcKey = pathToKey(srcPath);
    String dstKey = pathToKey(dstPath);
    this.store.copy(srcKey, dstKey);
    return true;
  }

  private boolean copyDirectory(Path srcPath, Path dstPath) throws IOException {
    String srcKey = pathToKey(srcPath);
    if (!srcKey.endsWith(PATH_DELIMITER)) {
      srcKey += PATH_DELIMITER;
    }
    String dstKey = pathToKey(dstPath);
    if (!dstKey.endsWith(PATH_DELIMITER)) {
      dstKey += PATH_DELIMITER;
    }

    if (dstKey.startsWith(srcKey)) {
      throw new IOException(
          "can not copy a directory to a subdirectory of self");
    }

    this.store.storeEmptyFile(dstKey);
    CosNCopyFileContext copyFileContext = new CosNCopyFileContext();

    int copiesToFinishes = 0;
    String priorLastKey = null;
    do {
      PartialListing objectList = this.store.list(
          srcKey, Constants.COS_MAX_LISTING_LENGTH, priorLastKey, true);
      for (FileMetadata file : objectList.getFiles()) {
        this.boundedCopyThreadPool.execute(new CosNCopyFileTask(
            this.store,
            file.getKey(),
            dstKey.concat(file.getKey().substring(srcKey.length())),
            copyFileContext));
        copiesToFinishes++;
        if (!copyFileContext.isCopySuccess()) {
          break;
        }
      }
      priorLastKey = objectList.getPriorLastKey();
    } while (null != priorLastKey);

    copyFileContext.lock();
    try {
      copyFileContext.awaitAllFinish(copiesToFinishes);
    } catch (InterruptedException e) {
      LOG.warn("interrupted when wait copies to finish");
    } finally {
      copyFileContext.lock();
    }

    return copyFileContext.isCopySuccess();
  }

  private void createParent(Path path) throws IOException {
    Path parent = path.getParent();
    if (parent != null) {
      String parentKey = pathToKey(parent);
      LOG.debug("Create parent key: {}", parentKey);
      if (!parentKey.equals(PATH_DELIMITER)) {
        String key = pathToKey(makeAbsolute(parent));
        if (key.length() > 0) {
          try {
            store.storeEmptyFile(key + PATH_DELIMITER);
          } catch (IOException e) {
            LOG.debug("Store a empty file in COS failed.", e);
            throw e;
          }
        }
      }
    }
  }

  @Override
  @SuppressWarnings("deprecation")
  public long getDefaultBlockSize() {
    return getConf().getLong(
        CosNConfigKeys.COSN_BLOCK_SIZE_KEY,
        CosNConfigKeys.DEFAULT_BLOCK_SIZE);
  }

  /**
   * Set the working directory to the given directory.
   */
  @Override
  public void setWorkingDirectory(Path newDir) {
    workingDir = newDir;
  }

  @Override
  public Path getWorkingDirectory() {
    return workingDir;
  }

  @Override
  public String getCanonicalServiceName() {
    // Does not support Token
    return null;
  }

  @Override
  public void close() throws IOException {
    try {
      this.store.close();
      this.boundedIOThreadPool.shutdown();
      this.boundedCopyThreadPool.shutdown();
    } finally {
      super.close();
    }
  }
}
