content: /*
   * 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.master.file.meta;
  
  import alluxio.AlluxioURI;
  import alluxio.client.WriteType;
  import alluxio.collections.Pair;
  import alluxio.concurrent.LockMode;
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.exception.BlockInfoException;
  import alluxio.exception.ExceptionMessage;
  import alluxio.exception.FileAlreadyExistsException;
  import alluxio.exception.FileDoesNotExistException;
  import alluxio.exception.InvalidPathException;
  import alluxio.exception.status.UnavailableException;
  import alluxio.grpc.CreateDirectoryPOptions;
  import alluxio.grpc.FileSystemMasterCommonPOptions;
  import alluxio.grpc.XAttrPropagationStrategy;
  import alluxio.master.block.ContainerIdGenerable;
  import alluxio.master.file.RpcContext;
  import alluxio.master.file.contexts.CreateDirectoryContext;
  import alluxio.master.file.contexts.CreateFileContext;
  import alluxio.master.file.contexts.CreatePathContext;
  import alluxio.master.journal.DelegatingJournaled;
  import alluxio.master.journal.JournalContext;
  import alluxio.master.journal.Journaled;
  import alluxio.master.journal.NoopJournalContext;
  import alluxio.master.metastore.DelegatingReadOnlyInodeStore;
  import alluxio.master.metastore.InodeStore;
  import alluxio.master.metastore.ReadOnlyInodeStore;
  import alluxio.proto.journal.File.DeleteFileEntry;
  import alluxio.proto.journal.File.NewBlockEntry;
  import alluxio.proto.journal.File.RenameEntry;
  import alluxio.proto.journal.File.SetAclEntry;
  import alluxio.proto.journal.File.UpdateInodeDirectoryEntry;
  import alluxio.proto.journal.File.UpdateInodeEntry;
  import alluxio.proto.journal.File.UpdateInodeFileEntry;
  import alluxio.resource.CloseableIterator;
  import alluxio.resource.CloseableResource;
  import alluxio.resource.LockResource;
  import alluxio.retry.ExponentialBackoffRetry;
  import alluxio.retry.RetryPolicy;
  import alluxio.security.authorization.AccessControlList;
  import alluxio.security.authorization.DefaultAccessControlList;
  import alluxio.security.authorization.Mode;
  import alluxio.underfs.UfsStatus;
  import alluxio.underfs.UnderFileSystem;
  import alluxio.underfs.options.MkdirsOptions;
  import alluxio.util.CommonUtils;
  import alluxio.util.interfaces.Scoped;
  import alluxio.wire.OperationId;
  
  import com.google.common.base.Preconditions;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  import java.text.MessageFormat;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.locks.Lock;
  import java.util.function.Function;
  import java.util.function.Supplier;
  import javax.annotation.Nullable;
  import javax.annotation.concurrent.NotThreadSafe;
  
  /**
   * Represents the tree of Inodes.
   */
  @NotThreadSafe
  // TODO(jiri): Make this class thread-safe.
  public class InodeTree implements DelegatingJournaled {
    private static final Logger LOG = LoggerFactory.getLogger(InodeTree.class);
    /** The base amount (exponential backoff) to sleep before retrying persisting an inode. */
    private static final int PERSIST_WAIT_BASE_SLEEP_MS = 2;
    /** Maximum amount (exponential backoff) to sleep before retrying persisting an inode. */
    private static final int PERSIST_WAIT_MAX_SLEEP_MS = 1000;
    /** The maximum retries for persisting an inode. */
    private static final int PERSIST_WAIT_MAX_RETRIES = 50;
  
    /** Value to be used for an inode with no parent. */
    public static final long NO_PARENT = -1;
  
    /**
     * Patterns of inode path locking.
     */
    public enum LockPattern {
      /**
       * Read lock every existing inode and edge along the path. Useful when we want to read an inode
       * without modifying anything.
       *
       * Examples
       *
       * path to lock: /a/b/c
       * existing inodes: /a/b
       * result: Read locks on [a, a->b, b]
       *
       * path to lock: /a/b/c
       * existing inodes: /a/b/c
       * result: Read locks on [a, a->b, b, b->c, c]
       *
       * path to lock: /a/b/c
       * existing inodes: /a
       * result: Read locks on [a]
       */
      READ,
      /**
       * Read lock every existing inode and edge along the path, but write lock the final inode if it
       * exists. If the inode does not exist, read lock the edge leading out of the final existing
       * ancestor. Useful when we want to modify an inode's metadata without changing the structure
       * of the inode tree (no create/rename/delete).
       *
       * Examples
       *
       * path to lock: /a/b/c
       * existing inodes: /a/b
       * result: Read locks on [a, a->b, b, b->c]
       *
       * path to lock: /a/b/c
       * existing inodes: /a/b/c
       * result: Read locks on [a, a->b, b, b->c], Write locks on [c]
       *
       * path to lock: /a/b/c
       * existing inodes: /a
       * result: Read locks on [a, a->b]
       */
      WRITE_INODE,
      /**
       * Read lock every existing inode and edge along the path, but write lock the edge leading out
       * of the last existing ancestor. Useful when we want to modify the structure of the inode tree,
       * e.g. when creating, deleting, or renaming inodes.
       *
       * Examples
       *
       * path to lock: /a/b/c
       * existing inodes: /a/b
       * result: Read locks on [a, a->b, b], Write locks on [b->c]
       *
       * path to lock: /a/b/c
       * existing inodes: /a/b/c
       * result: Read locks on [a, a->b, b], Write locks on [b->c, c]
       *
       * path to lock: /a/b/c
       * existing inodes: /a
       * result: Read locks on [a], Write locks [a->b]
       */
      WRITE_EDGE;
  
      /**
       * @return whether the lock pattern is one of the write-type patterns
       */
      public boolean isWrite() {
        return this == WRITE_INODE || this == WRITE_EDGE;
      }
    }
  
    /** Only the root inode should have the empty string as its name. */
    public static final String ROOT_INODE_NAME = "";
  
    /** Path of the root inode. */
    public static final String ROOT_PATH = "/";
  
    /** Number of retries when trying to lock a path, from a given id. */
    public static final int PATH_TRAVERSAL_RETRIES = 1000;
  
    /** Mount table manages the file system mount points. */
    private final MountTable mMountTable;
  
    private final TtlBucketList mTtlBuckets;
  
    /** Manager for inode locking. */
    private final InodeLockManager mInodeLockManager;
  
    private final ReadOnlyInodeStore mInodeStore;
  
    /**
     * Class for managing the persistent state of the inode tree. All metadata changes must go
     * through this class by calling {@link InodeTreePersistentState#applyAndJournal(context, entry)}.
     */
    private final InodeTreePersistentState mState;
  
    /**
     * Inode id management. Inode ids are essentially block ids.
     *
     * inode files: Each file id will be composed of a unique block container id, with the maximum
     * sequence number.
     *
     * inode directories: Each directory id will be a unique block id, in order to avoid any collision
     * with file ids.
     */
    private final ContainerIdGenerable mContainerIdGenerator;
    private final InodeDirectoryIdGenerator mDirectoryIdGenerator;
  
    /**
     * @param inodeStore the inode store
     * @param containerIdGenerator the container id generator to use to get new container ids
     * @param directoryIdGenerator the directory id generator to use to get new directory ids
     * @param mountTable the mount table to manage the file system mount points
     * @param lockManager inode lock manager
     */
    public InodeTree(InodeStore inodeStore, ContainerIdGenerable containerIdGenerator,
        InodeDirectoryIdGenerator directoryIdGenerator, MountTable mountTable,
        InodeLockManager lockManager) {
      mInodeStore = new DelegatingReadOnlyInodeStore(inodeStore);
      mTtlBuckets = new TtlBucketList(mInodeStore);
      mInodeLockManager = lockManager;
      mState = new InodeTreePersistentState(inodeStore, mInodeLockManager, mTtlBuckets);
      mContainerIdGenerator = containerIdGenerator;
      mDirectoryIdGenerator = directoryIdGenerator;
      mMountTable = mountTable;
    }
  
    /**
     * Whether given operation is still cached in retry-cache.
     *
     * @param opId the operation id
     * @return {@code true} if given op is marked complete
     */
    public boolean isOperationComplete(@Nullable OperationId opId) {
      return mState.isOperationComplete(opId);
    }
  
    /**
     * Used to mark an operation as complete in retry-cache.
     * @param opId the operation id
     */
    public void cacheOperation(@Nullable OperationId opId) {
      mState.cacheOperation(opId);
    }
  
    /**
     * Initializes the root of the inode tree.
     *
     * @param owner the root owner
     * @param group the root group
     * @param mode the root mode
     * @param context the journal context to journal the initialization to
     */
    public void initializeRoot(String owner, String group, Mode mode, JournalContext context)
        throws UnavailableException {
      if (mState.getRoot() == null) {
        MutableInodeDirectory root = MutableInodeDirectory.create(
            mDirectoryIdGenerator.getNewDirectoryId(context), NO_PARENT, ROOT_INODE_NAME,
            CreateDirectoryContext
                .mergeFrom(CreateDirectoryPOptions.newBuilder().setMode(mode.toProto()))
                .setOwner(owner).setGroup(group));
        root.setPersistenceState(PersistenceState.PERSISTED);
        mState.applyAndJournal(context, root, ROOT_PATH);
      }
    }
  
    /**
     * @return the list of TTL buckets for tracking inode TTLs
     */
    public TtlBucketList getTtlBuckets() {
      return mTtlBuckets;
    }
  
    /**
     * @return the number of inodes in the inode tree
     */
    public long getInodeCount() {
      return mState.getInodeCount();
    }
  
    /**
     * @return the file size histogram of the inode tree
     */
    public Map<Long, Number> getFileSizeHistogram() {
      return mState.getFileSizeHistogram();
    }
  
    /**
     * Marks an inode directory as having its direct children loaded.
     *
     * @param context journal context supplier
     * @param dir the inode directory
     */
    public void setDirectChildrenLoaded(Supplier<JournalContext> context, InodeDirectory dir) {
      setDirectChildrenLoaded(context, dir, true);
    }
  
    /**
     * Marks an inode directory as having its direct children loaded or not.
     *
     * @param context journal context supplier
     * @param dir the inode directory
     * @param directChildrenLoaded whether to load the direct children if they were not loaded before
     */
    public void setDirectChildrenLoaded(Supplier<JournalContext> context, InodeDirectory dir,
        boolean directChildrenLoaded) {
      mState.applyAndJournal(context, UpdateInodeDirectoryEntry.newBuilder()
          .setId(dir.getId())
          .setDirectChildrenLoaded(directChildrenLoaded)
          .build());
    }
  
    /**
     * @param context journal context supplier
     * @param entry an entry representing an add block operation
     * @return the new block id
     */
    public long newBlock(Supplier<JournalContext> context, NewBlockEntry entry) {
      return mState.applyAndJournal(context, entry);
    }
  
    /**
     * @param context journal context supplier
     * @param entry an entry representing an update inode file operation
     */
    public void updateInodeFile(Supplier<JournalContext> context, UpdateInodeFileEntry entry) {
      mState.applyAndJournal(context, entry);
    }
  
    /**
     * @param context journal context supplier
     * @param entry an entry representing an update inode operation
     */
    public void updateInode(Supplier<JournalContext> context, UpdateInodeEntry entry) {
      mState.applyAndJournal(context, entry);
    }
  
    /**
     * @param inodeId id of inode to be updated
     * @param accessTime the new access time
     * @return the applied inode entry
     */
    public UpdateInodeEntry updateInodeAccessTimeNoJournal(long inodeId, long accessTime) {
      return mState.applyInodeAccessTime(inodeId, accessTime);
    }
  
    /**
     * @param context journal context supplier
     * @param entry an entry representing a rename operation
     */
    public void rename(Supplier<JournalContext> context, RenameEntry entry) {
      mState.applyAndJournal(context, entry);
    }
  
    /**
     * @param context journal context supplier
     * @param entry an entry representing a set acl operation
     */
    public void setAcl(Supplier<JournalContext> context, SetAclEntry entry) {
      mState.applyAndJournal(context, entry);
    }
  
    /**
     * @return username of root of inode tree, null if the inode tree is not initialized
     */
    @Nullable
    public String getRootUserName() {
      if (mState.getRoot() == null) {
        return null;
      }
      return mState.getRoot().getOwner();
    }
  
    /**
     * @return the number of pinned inodes
     */
    public int getPinnedSize() {
      return mState.getPinnedInodeFileIds().size();
    }
  
    /**
     * @param id the id to get the inode for
     * @return whether the inode exists
     */
    public boolean inodeIdExists(long id) {
      return mInodeStore.get(id).isPresent();
    }
  
    /**
     * Locks existing inodes on the path and with the pattern defined by
     * {@link LockingScheme#getPath()} and {@link LockingScheme#getPattern()}.
     *
     * This method uses the {@link Lock#tryLock()} method to gain ownership of the locks. The reason
     * one might want to use this is to avoid the fairness heuristics within the
     * {@link java.util.concurrent.locks.ReentrantReadWriteLock}'s NonFairSync which may block reader
     * threads if a writer if the first in the queue.
     *
     * @param scheme the locking scheme to lock the path with
     * @param journalContext the journal context to flush when the LockedInodePath releases
     * @return the {@link LockedInodePath} representing the locked path of inodes
     * @throws InvalidPathException if the path is invalid
     */
    public LockedInodePath tryLockInodePath(LockingScheme scheme, JournalContext journalContext)
        throws InvalidPathException {
      return lockInodePath(scheme.getPath(), scheme.getPattern(), true, journalContext);
    }
  
    /**
     * Locks existing inodes on the path and with the pattern defined by
     * {@link LockingScheme#getPath()} and {@link LockingScheme#getPattern()}.
     *
     * @param scheme the locking scheme to lock the path with
     * @param journalContext the journal context to flush when the LockedInodePath releases
     * @return the {@link LockedInodePath} representing the locked path of inodes
     * @throws InvalidPathException if the path is invalid
     */
    public LockedInodePath lockInodePath(LockingScheme scheme, JournalContext journalContext)
        throws InvalidPathException {
      return lockInodePath(scheme.getPath(), scheme.getPattern(), journalContext);
    }
  
    /**
     * Locks existing inodes on the specified path, in the specified {@link LockPattern}. The target
     * inode is not required to exist.
     *
     * @param uri the uri to lock
     * @param lockPattern the {@link LockPattern} to lock the inodes with
     * @param journalContext the journal context to flush when the LockedInodePath releases
     * @return the {@link LockedInodePath} representing the locked path of inodes
     * @throws InvalidPathException if the path is invalid
     */
    public LockedInodePath lockInodePath(
        AlluxioURI uri, LockPattern lockPattern, JournalContext journalContext
    )
        throws InvalidPathException {
      return lockInodePath(uri, lockPattern, false, journalContext);
    }
  
    /**
     * Locks existing inodes on the specified path, in the specified {@link LockPattern}. The target
     * inode is not required to exist.
     *
     * @param uri the uri to lock
     * @param lockPattern the {@link LockPattern} to lock the inodes with
     * @param tryLock true to use {@link Lock#tryLock()} or false to use {@link Lock#lock()}
     * @param journalContext the journal context to flush when the LockedInodePath releases
     * @return the {@link LockedInodePath} representing the locked path of inodes
     * @throws InvalidPathException if the path is invalid
     */
    public LockedInodePath lockInodePath(
        AlluxioURI uri, LockPattern lockPattern, boolean tryLock, JournalContext journalContext)
        throws InvalidPathException {
      LockedInodePath inodePath =
          new LockedInodePath(
              uri, mInodeStore, mInodeLockManager, getRoot(), lockPattern, tryLock, journalContext
          );
      try {
        inodePath.traverse();
      } catch (Throwable t) {
        inodePath.close();
        throw t;
      }
      return inodePath;
    }
  
    /**
     * @param uri the {@link AlluxioURI} to check for existence
     * @return whether the inode exists
     */
    public boolean inodePathExists(AlluxioURI uri) {
      try (LockedInodePath inodePath
               = lockInodePath(uri, LockPattern.READ, NoopJournalContext.INSTANCE)
      ) {
        return inodePath.fullPathExists();
      } catch (InvalidPathException e) {
        return false;
      }
    }
  
    /**
     * Locks a path and throws an exception if the path does not exist.
     *
     * @param uri a uri to lock
     * @param lockScheme the scheme to lock with
     * @param journalContext the journal context to flush when the LockedInodePath releases
     * @return a locked inode path for the uri
     */
    public LockedInodePath lockFullInodePath(
        AlluxioURI uri, LockingScheme lockScheme, JournalContext journalContext
    )
        throws InvalidPathException, FileDoesNotExistException {
      LockedInodePath inodePath
          = lockInodePath(uri, lockScheme.getPattern(), NoopJournalContext.INSTANCE);
      if (!inodePath.fullPathExists()) {
        inodePath.close();
        throw new FileDoesNotExistException(ExceptionMessage.PATH_DOES_NOT_EXIST.getMessage(uri));
      }
      return inodePath;
    }
  
    /**
     * Locks a path and throws an exception if the path does not exist.
     *
     * @param uri a uri to lock
     * @param lockPattern the pattern to lock with
     * @param journalContext the journal context to flush when the LockedInodePath releases
     * @return a locked inode path for the uri
     */
    public LockedInodePath lockFullInodePath(
        AlluxioURI uri, LockPattern lockPattern, JournalContext journalContext
    )
        throws InvalidPathException, FileDoesNotExistException {
      LockedInodePath inodePath = lockInodePath(uri, lockPattern, journalContext);
      if (!inodePath.fullPathExists()) {
        inodePath.close();
        throw new FileDoesNotExistException(ExceptionMessage.PATH_DOES_NOT_EXIST.getMessage(uri));
      }
      return inodePath;
    }
  
    /**
     * Locks a path and throws an exception if the path does not exist.
     *
     * @param id the inode id to lock
     * @param lockPattern the pattern to lock with
     * @param journalContext the journal context to flush when the LockedInodePath releases
     * @return a locked inode path for the uri
     */
    public LockedInodePath lockFullInodePath(
        long id, LockPattern lockPattern, JournalContext journalContext
    )
        throws FileDoesNotExistException {
      LockedInodePath inodePath = lockInodePathById(id, lockPattern, journalContext);
      if (!inodePath.fullPathExists()) {
        inodePath.close();
        throw new FileDoesNotExistException(ExceptionMessage.INODE_DOES_NOT_EXIST.getMessage(id));
      }
      return inodePath;
    }
  
    /**
     * Locks existing inodes on the path to the inode specified by an id, in the specified
     * {@link LockPattern}. The target inode must exist. This may require multiple traversals of the
     * tree, so may be inefficient.
     *
     * @param id the inode id
     * @param lockPattern the {@link LockPattern} to lock the inodes with
     * @param journalContext the journal context to flush when the LockedInodePath releases
     * @return the {@link LockedInodePath} representing the locked path of inodes
     * @throws FileDoesNotExistException if the target inode does not exist
     */
    private LockedInodePath lockInodePathById(
        long id, LockPattern lockPattern, JournalContext journalContext
    )
        throws FileDoesNotExistException {
      int count = 0;
      while (true) {
        Optional<Inode> inode = mInodeStore.get(id);
        if (!inode.isPresent()) {
          throw new FileDoesNotExistException(ExceptionMessage.INODE_DOES_NOT_EXIST.getMessage(id));
        }
        // Compute the path given the target inode.
        StringBuilder builder = new StringBuilder();
        computePathForInode(inode.get(), builder);
        AlluxioURI uri = new AlluxioURI(builder.toString());
  
        boolean valid = false;
        LockedInodePath inodePath = null;
        try {
          inodePath = lockInodePath(uri, lockPattern, journalContext);
          if (inodePath.getInode().getId() == id) {
            // Set to true, so the path is not unlocked before returning.
            valid = true;
            return inodePath;
          }
          // The path does not end up at the target inode id. Repeat the traversal.
        } catch (InvalidPathException e) {
          // ignore and repeat the loop
          LOG.debug("Inode lookup id {} computed path {} mismatch id. Repeating.", id, uri);
        } finally {
          if (!valid && inodePath != null) {
            inodePath.close();
          }
        }
        count++;
        if (count > PATH_TRAVERSAL_RETRIES) {
          throw new FileDoesNotExistException(
              MessageFormat.format("inodeId {0,number,#} does not exist; too many retries", id));
        }
      }
    }
  
    /**
     * Locks existing inodes on the two specified paths. The two paths will be locked in the correct
     * order. The target inodes are not required to exist.
     *
     * @param path1 the first path to lock
     * @param lockPattern1 the locking pattern for the first path
     * @param path2 the second path to lock
     * @param lockPattern2 the locking pattern for the second path
     * @param journalContext the journal context to flush when the LockedInodePath releases
     * @return a {@link InodePathPair} representing the two locked paths
     * @throws InvalidPathException if a path is invalid
     */
    public InodePathPair lockInodePathPair(
        AlluxioURI path1, LockPattern lockPattern1,
        AlluxioURI path2, LockPattern lockPattern2,
        JournalContext journalContext
    ) throws InvalidPathException {
      LockedInodePath lockedPath1 = null;
      LockedInodePath lockedPath2 = null;
      boolean valid = false;
      try {
        // Lock paths in a deterministic order.
        if (path1.getPath().compareTo(path2.getPath()) > 0) {
          lockedPath2 = lockInodePath(path2, lockPattern2, journalContext);
          lockedPath1 = lockInodePath(path1, lockPattern1, journalContext);
        } else {
          lockedPath1 = lockInodePath(path1, lockPattern1, journalContext);
          lockedPath2 = lockInodePath(path2, lockPattern2, journalContext);
        }
        valid = true;
        return new InodePathPair(lockedPath1, lockedPath2);
      } finally {
        if (!valid) {
          if (lockedPath1 != null) {
            lockedPath1.close();
          }
          if (lockedPath2 != null) {
            lockedPath2.close();
          }
        }
      }
    }
  
    /**
     * Attempts to extend an existing {@link LockedInodePath} to reach the target inode (the last
     * inode for the full path). If the target inode does not exist, an exception will be thrown.
     *
     * @param inodePath the {@link LockedInodePath} to extend to the target inode
     * @throws InvalidPathException if the path is invalid
     * @throws FileDoesNotExistException if the target inode does not exist
     */
    public void ensureFullInodePath(LockedInodePath inodePath)
        throws InvalidPathException, FileDoesNotExistException {
      if (inodePath.fullPathExists()) {
        return;
      }
      inodePath.traverse();
      if (!inodePath.fullPathExists()) {
        throw new FileDoesNotExistException(
            ExceptionMessage.PATH_DOES_NOT_EXIST.getMessage(inodePath.getUri()));
      }
    }
  
    /**
     * Appends components of the path from a given inode.
     *
     * @param inode the inode to compute the path for
     * @param builder a {@link StringBuilder} that is updated with the path components
     * @throws FileDoesNotExistException if an inode in the path does not exist
     */
    private void computePathForInode(InodeView inode, StringBuilder builder)
        throws FileDoesNotExistException {
      long id;
      long parentId;
      String name;
      try (LockResource lr = mInodeLockManager.lockInode(inode, LockMode.READ, false)) {
        id = inode.getId();
        parentId = inode.getParentId();
        name = inode.getName();
      }
  
      if (isRootId(id)) {
        builder.append(AlluxioURI.SEPARATOR);
      } else if (isRootId(parentId)) {
        builder.append(AlluxioURI.SEPARATOR);
        builder.append(name);
      } else {
        Optional<Inode> parentInode = mInodeStore.get(parentId);
        if (!parentInode.isPresent()) {
          throw new FileDoesNotExistException(
              ExceptionMessage.INODE_DOES_NOT_EXIST.getMessage(parentId));
        }
        computePathForInode(parentInode.get(), builder);
        builder.append(AlluxioURI.SEPARATOR);
        builder.append(name);
      }
    }
  
    /**
     * Run the apply function on each inode in the path to inode and collect the results in
     * the items list.
     * The function should not keep a reference to the inodes since they will not be locked when the
     * function returns. This method should not be called while holding locks on any inodes.
     * The operation is not guaranteed to be atomic.
     *
     * @param inode the inode to calculate the path for
     * @param items where the results will be kept
     * @param apply the function to run on each inode
     * @param <T> the type of result to be collected in the items list
     * @throws FileDoesNotExistException if the path to the inode doesn't exist
     */
    private <T> void processPathForInode(InodeView inode, List<T> items, Function<InodeView, T> apply)
        throws FileDoesNotExistException {
      long id;
      long parentId;
      T item;
      try (LockResource lr = mInodeLockManager.lockInode(inode, LockMode.READ, false)) {
        item = apply.apply(inode);
        id = inode.getId();
        parentId = inode.getParentId();
      }
  
      if (!isRootId(id)) {
        Optional<Inode> parentInode = mInodeStore.get(parentId);
        if (!parentInode.isPresent()) {
          throw new FileDoesNotExistException(
              ExceptionMessage.INODE_DOES_NOT_EXIST.getMessage(parentId));
        }
        processPathForInode(parentInode.get(), items, apply);
      }
      items.add(item);
    }
  
    /**
     * Computes the path components for the given inode view.
     * This method should not be called while holding locks on any inodes.
     * The operation is not guaranteed to be atomic.
     *
     * @return the list of path components
     * @param inode the inode view to calculate the path for
     * @throws FileDoesNotExistException if the path to the inode doesn't exist
     */
    public ArrayList<String> getPathInodeNames(InodeView inode) throws FileDoesNotExistException {
      ArrayList<String> items = new ArrayList<>();
      processPathForInode(inode, items, InodeView::getName);
      return items;
    }
  
    /**
     * Computes the path components for the given inode id.
     * This method should not be called while holding locks on any inodes.
     * The operation is not guaranteed to be atomic.
     *
     * @return the list of path components
     * @param id the inode id to calculate the path for
     * @throws FileDoesNotExistException if the path to the inode doesn't exist
     */
    public ArrayList<String> getPathInodeNames(long id) throws FileDoesNotExistException {
      Inode inode = mInodeStore.get(id).orElseThrow(() -> new FileDoesNotExistException(String.format(
          "Inode %d does not exit", id)));
      return getPathInodeNames(inode);
    }
  
    /**
     * Returns the path for a particular inode id. The inode and the path to the inode must already be
     * locked.
     *
     * @param id the inode id to get the path for
     * @return the {@link AlluxioURI} for the path of the inode
     * @throws FileDoesNotExistException if the path does not exist
     */
    public AlluxioURI getPath(long id) throws FileDoesNotExistException {
      Inode inode = mInodeStore.get(id).orElseThrow(() -> new FileDoesNotExistException(String.format(
          "Inode %d does not exit", id)));
      return getPath(inode);
    }
  
    /**
     * Returns the path for a particular inode. The inode and the path to the inode must already be
     * locked.
     *
     * @param inode the inode to get the path for
     * @return the {@link AlluxioURI} for the path of the inode
     * @throws FileDoesNotExistException if the path does not exist
     */
    public AlluxioURI getPath(InodeView inode) throws FileDoesNotExistException {
      StringBuilder builder = new StringBuilder();
      computePathForInode(inode, builder);
      return new AlluxioURI(builder.toString());
    }
  
    /**
     * @return the root inode
     */
    public InodeDirectory getRoot() {
      return mState.getRoot();
    }
  
    /**
     * Creates a file or directory at path.
     *
     * This method expects that the last edge leading to the target inode to be write-locked. If the
     * last existing inode in the path is /a/b/c and we want to create /a/b/c/d/e, the c->d edge must
     * be write locked.
     *
     * On success, createPath attempts to push the write lock forward as far as possible. For the
     * above example, createPath would take a write lock on d->e, and downgrade the c->d lock from a
     * write lock to a read lock. This may not be possible if inodePath is a composite path which
     * doesn't own the write lock. In that case no downgrade will occur.
     *
     * @param rpcContext the rpc context
     * @param inodePath the path
     * @param context method context
     * @return a list of created inodes
     * @throws FileAlreadyExistsException when there is already a file at path if we want to create a
     *         directory there
     * @throws BlockInfoException when blockSizeBytes is invalid
     * @throws InvalidPathException when path is invalid, for example, (1) when there is nonexistent
     *         necessary parent directories and recursive is false, (2) when one of the necessary
     *         parent directories is actually a file
     * @throws FileDoesNotExistException if the parent of the path does not exist and the recursive
     *         option is false
     */
    public List<Inode> createPath(RpcContext rpcContext, LockedInodePath inodePath,
        CreatePathContext<?, ?> context) throws FileAlreadyExistsException, BlockInfoException,
        InvalidPathException, IOException, FileDoesNotExistException {
      Preconditions.checkState(inodePath.getLockPattern() == LockPattern.WRITE_EDGE);
  
      // TODO(gpang): consider splitting this into createFilePath and createDirectoryPath, with a
      // helper method for the shared logic.
      AlluxioURI path = inodePath.getUri();
      if (path.isRoot()) {
        String errorMessage = "Not allowed to create existing root path: " + path;
        LOG.error(errorMessage);
        throw new FileAlreadyExistsException(errorMessage);
      }
      if (inodePath.fullPathExists()) {
        if (context instanceof CreateDirectoryContext
            && ((CreateDirectoryContext) context).getOptions().getAllowExists()) {
          return Collections.emptyList();
        } else {
          String pathType = "file";
          if (context instanceof CreateDirectoryContext) {
            pathType = "directory";
          }
          String errorMessage = String
              .format("Not allowed to create %s because path already exists: %s", pathType, path);
          throw new FileAlreadyExistsException(errorMessage);
        }
      }
  
      if (context instanceof CreateFileContext) {
        CreateFileContext fileContext = (CreateFileContext) context;
        if (fileContext.getOptions().getBlockSizeBytes() < 1) {
          throw new BlockInfoException(
              "Invalid block size " + fileContext.getOptions().getBlockSizeBytes());
        }
      }
      LOG.debug("createPath {}", path);
  
      String[] pathComponents = inodePath.mPathComponents;
      String name = path.getName();
  
      // pathIndex is the index into pathComponents where we start filling in the path from the inode.
      int pathIndex = inodePath.getExistingInodeCount();
      if (pathIndex < pathComponents.length - 1) {
        // The immediate parent was not found. If it's not recursive, we throw an exception here.
        // Otherwise we add the remaining path components to the list of components to create.
        if (!context.isRecursive()) {
          throw new FileDoesNotExistException(String.format(
              "File %s creation failed. Component %d(%s) does not exist",
              path, pathIndex, pathComponents[pathIndex]));
        }
      }
      // The ancestor inode (parent or ancestor) of the target path.
      Inode ancestorInode = inodePath.getAncestorInode();
      if (!ancestorInode.isDirectory()) {
        throw new InvalidPathException("Could not traverse to parent directory of path " + path
            + ". Component " + pathComponents[pathIndex - 1] + " is not a directory.");
      }
      InodeDirectoryView currentInodeDirectory = ancestorInode.asDirectory();
  
      List<Inode> createdInodes = new ArrayList<>();
      if (context.isPersisted() && context.isPersistNonExistingParentDirectories()) {
        // Synchronously persist directories. These inodes are already READ locked.
        for (Inode inode : inodePath.getInodeList()) {
          if (!inode.isPersisted()) {
            // This cast is safe because we've already verified that the file inode doesn't exist.
            syncPersistExistingDirectory(rpcContext, inode.asDirectory(), context.isMetadataLoad());
          }
        }
      }
      if ((pathIndex < (pathComponents.length - 1)
          || !mInodeStore.getChild(currentInodeDirectory, name).isPresent())
          && context.getOperationTimeMs() > currentInodeDirectory.getLastModificationTimeMs()) {
        // (1) There are components in parent paths that need to be created. Or
        // (2) The last component of the path needs to be created.
        // In these two cases, the last traversed Inode will be modified if the new timestamp is after
        // the existing last modified time.
        long currentId = currentInodeDirectory.getId();
        try (LockResource lr = mInodeLockManager.lockUpdate(currentId)) {
          long updatedLastModified = mInodeStore.get(currentId).get().getLastModificationTimeMs();
          if (updatedLastModified < context.getOperationTimeMs()) {
            UpdateInodeEntry.Builder updateInodeEntry = UpdateInodeEntry.newBuilder()
                .setId(currentId)
                .setLastModificationTimeMs(context.getOperationTimeMs())
                .setLastAccessTimeMs(context.getOperationTimeMs());
            if (context.getXAttr() != null
                && context.getXAttrPropStrat() != null
                && context.getXAttrPropStrat() == XAttrPropagationStrategy.NEW_PATHS) {
              // TODO(czhu): determine if xAttr update strategy is required for this
              updateInodeEntry.putAllXAttr(CommonUtils.convertToByteString(context.getXAttr()));
            }
            mState.applyAndJournal(rpcContext, updateInodeEntry.build());
          }
        }
      }
  
      // Fill in the ancestor directories that were missing.
      // NOTE, we set the mode of missing ancestor directories to be the default value, rather
      // than inheriting the option of the final file to create, because it may not have
      // "execute" permission.
      StringBuilder pathBuilder = new StringBuilder().append(
          String.join(AlluxioURI.SEPARATOR, Arrays.asList(pathComponents).subList(0, pathIndex))
      );
      CreateDirectoryContext missingDirContext = null;
      if (pathIndex < pathComponents.length - 1) {
        missingDirContext = CreateDirectoryContext.defaults();
        missingDirContext.getOptions().setCommonOptions(FileSystemMasterCommonPOptions.newBuilder()
            .setTtl(context.getTtl()).setTtlAction(context.getTtlAction()));
        missingDirContext.setWriteType(context.getWriteType());
        missingDirContext.setOperationTimeMs(context.getOperationTimeMs());
        missingDirContext.setMountPoint(false);
        missingDirContext.setOwner(context.getOwner());
        missingDirContext.setGroup(context.getGroup());
        if (context.isMetadataLoad() && !context.isPersistNonExistingParentDirectories()) {
          // If this is a metadata load, and we are not going to persist internal
          // directories (i.e. adding object markers), then we mark the internal
          // directories as persisted
          missingDirContext.setWriteType(WriteType.THROUGH);
          missingDirContext.setMissingDirFingerprint(context::getMissingDirFingerprint);
        }
        if (context.getXAttr() != null
            && context.getXAttrPropStrat() != null
            && context.getXAttrPropStrat() == XAttrPropagationStrategy.NEW_PATHS) {
          missingDirContext.setXAttr(context.getXAttr());
        }
      }
      for (int k = pathIndex; k < (pathComponents.length - 1); k++) {
        assert missingDirContext != null;
        MutableInodeDirectory newDir = MutableInodeDirectory.create(
            mDirectoryIdGenerator.getNewDirectoryId(rpcContext.getJournalContext()),
            currentInodeDirectory.getId(), pathComponents[k], missingDirContext);
        if (currentInodeDirectory.isPinned() && !newDir.isPinned()) {
          newDir.setPinned(true);
          newDir.setMediumTypes(currentInodeDirectory.getMediumTypes());
        }
        inheritOwnerAndGroupIfEmpty(newDir, currentInodeDirectory);
  
        // if the parent has default ACL, copy that default ACL as the new directory's default
        // and access acl, ANDed with the umask
        // if it is part of a metadata load operation, we ignore the umask and simply inherit
        // the default ACL as the directory's new default and access ACL
        short mode = context.isMetadataLoad() ? Mode.createFullAccess().toShort()
            : newDir.getMode();
        DefaultAccessControlList dAcl = currentInodeDirectory.getDefaultACL();
        if (!dAcl.isEmpty()) {
          Pair<AccessControlList, DefaultAccessControlList> pair =
              dAcl.generateChildDirACL(mode);
          newDir.setInternalAcl(pair.getFirst());
          newDir.setDefaultACL(pair.getSecond());
        }
        if (context.isPersisted() && !context.isPersistNonExistingParentDirectories()) {
          newDir.setPersistenceState(PersistenceState.PERSISTED);
          newDir.setUfsFingerprint(context.getMissingDirFingerprint());
        }
        String newDirPath = k == 0 ? ROOT_PATH
            : pathBuilder.append(AlluxioURI.SEPARATOR).append(pathComponents[k]).toString();
        mState.applyAndJournal(rpcContext, newDir,
            newDirPath);
  
        inodePath.addNextInode(Inode.wrap(newDir));
  
        // Persist the directory *after* it exists in the inode tree. This prevents multiple
        // concurrent creates from trying to persist the same directory name.
        if (context.isPersisted() && context.isPersistNonExistingParentDirectories()) {
          syncPersistExistingDirectory(rpcContext, newDir, context.isMetadataLoad());
        }
        createdInodes.add(Inode.wrap(newDir));
        currentInodeDirectory = newDir;
      }
  
      // Create the final path component.
      MutableInode<?> newInode;
  
      // create the new inode, with a write lock
      if (context instanceof CreateDirectoryContext) {
        CreateDirectoryContext directoryContext = (CreateDirectoryContext) context;
        MutableInodeDirectory newDir = MutableInodeDirectory.create(
            mDirectoryIdGenerator.getNewDirectoryId(rpcContext.getJournalContext()),
            currentInodeDirectory.getId(), name, directoryContext);
  
        if (context.getXAttr() != null) {
          newDir.setXAttr(context.getXAttr());
        }
  
        // if the parent has default ACL, take the default ACL ANDed with the umask as the new
        // directory's default and access acl
        // When it is a metadata load operation, do not take the umask into account
        short mode = context.isMetadataLoad() ? Mode.createFullAccess().toShort()
            : newDir.getMode();
        DefaultAccessControlList dAcl = currentInodeDirectory.getDefaultACL();
        if (!dAcl.isEmpty()) {
          Pair<AccessControlList, DefaultAccessControlList> pair =
              dAcl.generateChildDirACL(mode);
          newDir.setInternalAcl(pair.getFirst());
          newDir.setDefaultACL(pair.getSecond());
        }
  
        if (directoryContext.isPersisted()) {
          // Do not journal the persist entry, since a creation entry will be journaled instead.
          if (context.isMetadataLoad()) {
            // if we are creating the file as a result of loading metadata, the newDir is already
            // persisted, and we got the permissions info from the ufs.
            newDir.setOwner(context.getOwner().intern())
                .setGroup(context.getGroup().intern())
                .setMode(context.getMode().toShort());
  
            Long operationTimeMs = context.getOperationTimeMs();
            if (operationTimeMs != null) {
              newDir.setLastModificationTimeMs(operationTimeMs, true);
              newDir.setLastAccessTimeMs(operationTimeMs, true);
            }
            newDir.setPersistenceState(PersistenceState.PERSISTED);
          } else {
            syncPersistNewDirectory(newDir, context);
          }
        }
        // Do NOT call setOwner/Group after inheriting from parent if empty
        inheritOwnerAndGroupIfEmpty(newDir, currentInodeDirectory);
        newInode = newDir;
      } else if (context instanceof CreateFileContext) {
        CreateFileContext fileContext = (CreateFileContext) context;
        final long blockContainerId;
        if (fileContext.getCompleteFileInfo() != null) {
          blockContainerId = fileContext.getCompleteFileInfo().getContainerId();
        } else {
          blockContainerId = mContainerIdGenerator.getNewContainerId();
        }
  
        MutableInodeFile newFile = MutableInodeFile.create(blockContainerId,
            currentInodeDirectory.getId(), name, System.currentTimeMillis(), fileContext);
  
        // if the parent has a default ACL, copy that default ACL ANDed with the umask as the new
        // file's access ACL.
        // If it is a metadata load operation, do not consider the umask.
        DefaultAccessControlList dAcl = currentInodeDirectory.getDefaultACL();
        short mode = context.isMetadataLoad() ? Mode.createFullAccess().toShort() : newFile.getMode();
        if (!dAcl.isEmpty()) {
          AccessControlList acl = dAcl.generateChildFileACL(mode);
          newFile.setInternalAcl(acl);
        }
        if (fileContext.isCacheable()) {
          newFile.setCacheable(true);
        }
        if (fileContext.getWriteType() == WriteType.ASYNC_THROUGH) {
          newFile.setPersistenceState(PersistenceState.TO_BE_PERSISTED);
        }
        if (context.getXAttr() != null) {
          newFile.setXAttr(context.getXAttr());
        }
  
        // Do NOT call setOwner/Group after inheriting from parent if empty
        inheritOwnerAndGroupIfEmpty(newFile, currentInodeDirectory);
        newInode = newFile;
      } else {
        throw new IllegalStateException(String.format("Unrecognized create options: %s", context));
      }
      if (currentInodeDirectory.isPinned() && !newInode.isPinned()) {
        newInode.setPinned(true);
        newInode.setMediumTypes(currentInodeDirectory.getMediumTypes());
      }
  
      mState.applyAndJournal(rpcContext, newInode,
          inodePath.getUri().getPath());
      Inode inode = Inode.wrap(newInode);
      inodePath.addNextInode(inode);
      createdInodes.add(inode);
      LOG.debug("createFile: File Created: {} parent: {}", newInode, currentInodeDirectory);
      return createdInodes;
    }
  
    // Inherit owner and group from ancestor if both are empty
    private static void inheritOwnerAndGroupIfEmpty(MutableInode<?> newInode,
        InodeDirectoryView ancestorInode) {
      if (Configuration.getBoolean(PropertyKey.MASTER_METASTORE_INODE_INHERIT_OWNER_AND_GROUP)
          && newInode.getOwner().isEmpty() && newInode.getGroup().isEmpty()) {
        // Inherit owner / group if empty
        newInode.setOwner(ancestorInode.getOwner().intern());
        newInode.setGroup(ancestorInode.getGroup().intern());
      }
    }
  
    /**
     * Returns {@link LockedInodePath}s for all descendants of inodePath.
     *
     * @param inodePath a locked inode path. It must be write-locked
     * @return all descendants
     */
    public LockedInodePathList getDescendants(LockedInodePath inodePath) {
      Preconditions.checkState(inodePath.getLockPattern() == LockPattern.WRITE_EDGE
          || (inodePath.getLockPattern() == LockPattern.WRITE_INODE && inodePath.fullPathExists()));
  
      List<LockedInodePath> descendants = new ArrayList<>();
      try {
        gatherDescendants(inodePath, descendants);
      } catch (Throwable t) {
        descendants.forEach(LockedInodePath::close);
        throw t;
      }
      return new LockedInodePathList(descendants);
    }
  
    private void gatherDescendants(LockedInodePath inodePath, List<LockedInodePath> descendants) {
      Inode inode = inodePath.getInodeOrNull();
      if (inode == null || inode.isFile()) {
        return;
      }
      try (CloseableIterator<? extends Inode> it = mInodeStore.getChildren(inode.asDirectory())) {
        while (it.hasNext()) {
          Inode child = it.next();
          LockedInodePath childPath;
          try {
            childPath = inodePath.lockChild(child, LockPattern.WRITE_EDGE);
          } catch (InvalidPathException e) {
            // Child does not exist.
            continue;
          }
          try {
            descendants.add(childPath);
          } catch (Error e) {
            // If adding to descendants fails due to OOM, this object
            // will not be tracked so we must close it manually
            childPath.close();
            throw e;
          }
          gatherDescendants(childPath, descendants);
        }
      }
    }
  
    /**
     * Deletes a single inode from the inode tree by removing it from the parent inode.
     *
     * @param rpcContext the rpc context
     * @param inodePath the {@link LockedInodePath} to delete
     * @param opTimeMs the operation time
     * @throws FileDoesNotExistException if the Inode cannot be retrieved
     */
    public void deleteInode(RpcContext rpcContext, LockedInodePath inodePath, long opTimeMs)
        throws FileDoesNotExistException {
      Preconditions.checkState(inodePath.getLockPattern() == LockPattern.WRITE_EDGE);
      Inode inode = inodePath.getInode();
  
      mState.applyAndJournal(rpcContext, DeleteFileEntry.newBuilder()
          .setId(inode.getId())
          .setRecursive(false)
          .setOpTimeMs(opTimeMs)
          .setPath(inodePath.getUri().getPath())
          .build());
  
      if (inode.isFile()) {
        rpcContext.getBlockDeletionContext().registerBlocksForDeletion(inode.asFile().getBlockIds());
      }
    }
  
    private boolean checkPinningValidity(Set<String> pinnedMediumTypes) {
      List<String> mediumTypeList = Configuration.getList(
          PropertyKey.MASTER_TIERED_STORE_GLOBAL_MEDIUMTYPE);
      for (String medium : pinnedMediumTypes) {
        if (!mediumTypeList.contains(medium)) {
          // mediumTypeList does not contains medium
          return false;
        }
      }
      return true;
    }
  
    /**
     * Sets the pinned state of an inode. If the inode is a directory, the pinned state will be set
     * recursively.
     *
     * @param rpcContext the rpc context
     * @param inodePath the {@link LockedInodePath} to set the pinned state for
     * @param pinned the pinned state to set for the inode (and possible descendants)
     * @param mediumTypes the list of pinned media that that the file can reside in
     * @param opTimeMs the operation time
     * @throws FileDoesNotExistException if inode does not exist
     */
    public void setPinned(RpcContext rpcContext, LockedInodePath inodePath, boolean pinned,
        List<String> mediumTypes, long opTimeMs)
        throws FileDoesNotExistException, InvalidPathException {
      Preconditions.checkState(inodePath.getLockPattern().isWrite());
      Set<String> mediumSet = new HashSet<>(mediumTypes);
      Preconditions.checkState(checkPinningValidity(mediumSet));
  
      Inode inode = inodePath.getInode();
      mState.applyAndJournal(rpcContext, UpdateInodeEntry.newBuilder()
          .setId(inode.getId())
          .setPinned(pinned)
          .addAllMediumType(mediumSet)
          .setLastModificationTimeMs(opTimeMs)
          .build());
  
      if (inode.isDirectory()) {
        assert inode instanceof InodeDirectory;
        // inode is a directory. Set the pinned state for all children.
        try (CloseableIterator<? extends Inode> it = mInodeStore.getChildren(inode.asDirectory())) {
          while (it.hasNext()) {
            Inode child = it.next();
            try (LockedInodePath childPath =
                     inodePath.lockChild(child, LockPattern.WRITE_INODE)) {
              // No need for additional locking since the parent is write-locked.
              setPinned(rpcContext, childPath, pinned, mediumTypes, opTimeMs);
            }
          }
        }
      }
    }
  
    /**
     * Sets the min and/or max replication level of an inode. If the inode is a directory, the state
     * will be set recursively. Arguments replicationMax and replicationMin can be null if they are
     * not meant to be set.
     *
     * @param rpcContext the rpc context
     * @param inodePath the {@link LockedInodePath} to set the pinned state for
     * @param replicationMax the max replication level to set for the inode (and possible descendants)
     * @param replicationMin the min replication level to set for the inode (and possible descendants)
     * @param opTimeMs the operation time
     * @throws FileDoesNotExistException if inode does not exist
     */
    public void setReplication(RpcContext rpcContext, LockedInodePath inodePath,
        Integer replicationMax, Integer replicationMin, long opTimeMs)
        throws FileDoesNotExistException, InvalidPathException {
      Preconditions.checkArgument(replicationMin != null || replicationMax != null,
          "Both min and max replication are null");
      Preconditions.checkArgument(replicationMin == null || replicationMin >= 0,
          "Min replication must be a non-negative integer");
      Preconditions.checkState(inodePath.getLockPattern().isWrite());
  
      Inode inode = inodePath.getInode();
  
      if (inode.isFile()) {
        InodeFile inodeFile = inode.asFile();
        int newMax = (replicationMax == null) ? inodeFile.getReplicationMax() : replicationMax;
        int newMin = (replicationMin == null) ? inodeFile.getReplicationMin() : replicationMin;
  
        Preconditions.checkArgument(newMax == alluxio.Constants.REPLICATION_MAX_INFINITY
            || newMax >= newMin,
            "Cannot set min and max replication to be %s and %s: "
                + "min replication must be smaller or equal than max replication",
            newMin, newMax);
  
        mState.applyAndJournal(rpcContext, UpdateInodeFileEntry.newBuilder()
            .setId(inode.getId())
            .setReplicationMax(newMax)
            .setReplicationMin(newMin)
            .build());
        mState.applyAndJournal(rpcContext, UpdateInodeEntry.newBuilder()
            .setId(inode.getId())
            .setPinned(newMin > 0)
            .addAllMediumType(inode.getMediumTypes())
            .setLastModificationTimeMs(opTimeMs)
            .build());
      } else {
        try (CloseableIterator<? extends Inode> it = mInodeStore.getChildren(inode.asDirectory())) {
          while (it.hasNext()) {
            Inode child = it.next();
            try (LockedInodePath tempInodePath =
                     inodePath.lockChild(child, LockPattern.WRITE_INODE)) {
              // No need for additional locking since the parent is write-locked.
              setReplication(rpcContext, tempInodePath, replicationMax, replicationMin, opTimeMs);
            }
          }
        }
      }
    }
  
    /**
     * @return the set of file ids whose replication max is not infinity
     */
    public Set<Long> getReplicationLimitedFileIds() {
      return mState.getReplicationLimitedFileIds();
    }
  
    /**
     * @return an unmodifiable view of the files with persistence state
     *         {@link PersistenceState#TO_BE_PERSISTED}
     */
    public Set<Long> getToBePersistedIds() {
      return mState.getToBePersistedIds();
    }
  
    /**
     * @return the set of file ids which are pinned
     */
    public Set<Long> getPinIdSet() {
      return mState.getPinnedInodeFileIds();
    }
  
    /**
     * @return the inode lock manager for the inode tree
     */
    public InodeLockManager getInodeLockManager() {
      return mInodeLockManager;
    }
  
    /**
     * @param fileId the file id to check
     * @return true if the given file id is the root id
     */
    public boolean isRootId(long fileId) {
      Preconditions.checkNotNull(mState.getRoot(),
          "Cannot call isRootId() before initializeRoot()");
      return fileId == mState.getRoot().getId();
    }
  
    @Override
    public Journaled getDelegate() {
      return mState;
    }
  
    /**
     * Synchronously persists an inode directory to the UFS. If concurrent calls are made, only
     * one thread will persist to UFS, and the others will wait until it is persisted.
     *
     * @param context journal context supplier
     * @param dir the inode directory to persist
     * @param isMetadataLoad if this operation is the result of a metadata load
     * @throws InvalidPathException if the path for the inode is invalid
     * @throws FileDoesNotExistException if the path for the inode is invalid
     */
    public void syncPersistExistingDirectory(Supplier<JournalContext> context, InodeDirectoryView dir,
          boolean isMetadataLoad)
        throws IOException, InvalidPathException, FileDoesNotExistException {
      RetryPolicy retry =
          new ExponentialBackoffRetry(PERSIST_WAIT_BASE_SLEEP_MS, PERSIST_WAIT_MAX_SLEEP_MS,
              PERSIST_WAIT_MAX_RETRIES);
      while (retry.attempt()) {
        if (dir.getPersistenceState() == PersistenceState.PERSISTED) {
          // The directory is persisted
          return;
        }
        Optional<Scoped> persisting = mInodeLockManager.tryAcquirePersistingLock(dir.getId());
        if (!persisting.isPresent()) {
          // Someone else is doing this persist. Continue and wait for them to finish.
          continue;
        }
        try (Scoped s = persisting.get()) {
          if (dir.getPersistenceState() == PersistenceState.PERSISTED) {
            // The directory is persisted
            return;
          }
          mState.applyAndJournal(context, UpdateInodeEntry.newBuilder()
              .setId(dir.getId())
              .setPersistenceState(PersistenceState.TO_BE_PERSISTED.name())
              .build());
          UpdateInodeEntry.Builder entry = UpdateInodeEntry.newBuilder()
              .setId(dir.getId());
          syncPersistDirectory(dir, isMetadataLoad).ifPresent(status -> {
            if (isRootId(dir.getId())) {
              // Don't load the root dir metadata from UFS
              return;
            }
            entry.setOwner(status.getOwner())
                .setGroup(status.getGroup())
                .setMode(status.getMode());
  
            Map<String, byte[]> xattr = status.getXAttr();
            if (xattr != null) {
              entry.putAllXAttr(CommonUtils.convertToByteString(xattr));
            }
  
            Long lastModificationTime = status.getLastModifiedTime();
            if (lastModificationTime != null) {
              entry.setLastModificationTimeMs(lastModificationTime)
                  .setOverwriteModificationTime(true);
            }
          });
          LOG.debug("Sync persisted dir {}, was caused by metadata load: {}", dir.getName(),
              isMetadataLoad);
          entry.setPersistenceState(PersistenceState.PERSISTED.name());
  
          mState.applyAndJournal(context, entry.build());
          return;
        }
      }
      throw new IOException(
          MessageFormat.format("Failed to create {0} in the under file system", dir.getName()));
    }
  
    /**
     * Synchronously persists an inode directory to the UFS.
     *
     * This method does not handle concurrent modification to the given inode, so the inode must not
     * yet be added to the inode tree.
     *
     * @param dir the inode directory to persist
     * @param createContext the create directory context
     */
    public void syncPersistNewDirectory(
        MutableInodeDirectory dir, CreatePathContext<?, ?> createContext)
        throws InvalidPathException, FileDoesNotExistException, IOException {
      dir.setPersistenceState(PersistenceState.TO_BE_PERSISTED);
      syncPersistDirectory(dir, createContext.isMetadataLoad()).ifPresent(status -> {
        // If the directory already exists in the UFS, update our metadata to match the UFS.
        dir.setOwner(status.getOwner().intern())
            .setGroup(status.getGroup().intern())
            .setMode(status.getMode())
            .setXAttr(status.getXAttr());
  
        Long lastModificationTime = status.getLastModifiedTime();
        if (lastModificationTime != null) {
          dir.setLastModificationTimeMs(lastModificationTime, true);
          // TODO(feng): update last access time from UFS
          dir.setLastAccessTimeMs(lastModificationTime, true);
        }
      });
      dir.setPersistenceState(PersistenceState.PERSISTED);
    }
  
    /**
     * Persists the directory to the UFS, returning the UFS status if the directory is found to
     * already exist in the UFS.
     *
     * @param dir the directory to persist
     * @param isMetadataLoad if this operation is the result of a metadata load
     * @return optional ufs status if the directory already existed
     */
    private Optional<UfsStatus> syncPersistDirectory(InodeDirectoryView dir, boolean isMetadataLoad)
        throws FileDoesNotExistException, IOException, InvalidPathException {
      AlluxioURI uri = getPath(dir);
      MountTable.Resolution resolution = mMountTable.resolve(uri);
      String ufsUri = resolution.getUri().toString();
      try (CloseableResource<UnderFileSystem> ufsResource = resolution.acquireUfsResource()) {
        UnderFileSystem ufs = ufsResource.get();
        MkdirsOptions mkdirsOptions =
            MkdirsOptions.defaults(Configuration.global()).setCreateParent(false)
            .setOwner(dir.getOwner()).setGroup(dir.getGroup()).setMode(new Mode(dir.getMode()));
        if (!ufs.mkdirs(ufsUri, mkdirsOptions)) {
          // Directory might already exist. Try loading the status from ufs.
          UfsStatus status;
          try {
            status = ufs.getStatus(ufsUri);
          } catch (Exception e) {
            throw new IOException(String.format("Cannot create or load UFS directory %s: %s.",
                ufsUri, e.toString()), e);
          }
          if (status.isFile()) {
            throw new InvalidPathException(String.format(
                "Error persisting directory. A file exists at the UFS location %s.", ufsUri));
          }
          return Optional.of(status);
        }
      }
      return Optional.empty();
    }
  
    /**
     * Close resources associated with this tree instance.
     *
     * @throws IOException
     */
    public void close() throws IOException {
      mInodeStore.close();
    }
  }
  