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 static alluxio.conf.PropertyKey.MASTER_FILE_SYSTEM_OPERATION_RETRY_CACHE_ENABLED;
  import static alluxio.conf.PropertyKey.MASTER_FILE_SYSTEM_OPERATION_RETRY_CACHE_SIZE;
  import static alluxio.conf.PropertyKey.MASTER_METRICS_FILE_SIZE_DISTRIBUTION_BUCKETS;
  
  import alluxio.ProcessUtils;
  import alluxio.conf.Configuration;
  import alluxio.master.file.RpcContext;
  import alluxio.master.journal.JournalContext;
  import alluxio.master.journal.JournalUtils;
  import alluxio.master.journal.Journaled;
  import alluxio.master.journal.checkpoint.CheckpointInputStream;
  import alluxio.master.journal.checkpoint.CheckpointName;
  import alluxio.master.metastore.InodeStore;
  import alluxio.proto.journal.File.AsyncPersistRequestEntry;
  import alluxio.proto.journal.File.CompleteFileEntry;
  import alluxio.proto.journal.File.DeleteFileEntry;
  import alluxio.proto.journal.File.InodeDirectoryEntry;
  import alluxio.proto.journal.File.InodeFileEntry;
  import alluxio.proto.journal.File.InodeLastModificationTimeEntry;
  import alluxio.proto.journal.File.NewBlockEntry;
  import alluxio.proto.journal.File.PersistDirectoryEntry;
  import alluxio.proto.journal.File.RenameEntry;
  import alluxio.proto.journal.File.SetAclEntry;
  import alluxio.proto.journal.File.SetAttributeEntry;
  import alluxio.proto.journal.File.UpdateInodeDirectoryEntry;
  import alluxio.proto.journal.File.UpdateInodeEntry;
  import alluxio.proto.journal.File.UpdateInodeEntry.Builder;
  import alluxio.proto.journal.File.UpdateInodeFileEntry;
  import alluxio.proto.journal.Journal.JournalEntry;
  import alluxio.resource.CloseableIterator;
  import alluxio.resource.LockResource;
  import alluxio.security.authorization.AclEntry;
  import alluxio.security.authorization.DefaultAccessControlList;
  import alluxio.util.BucketCounter;
  import alluxio.util.FormatUtils;
  import alluxio.util.StreamUtils;
  import alluxio.util.proto.ProtoUtils;
  import alluxio.wire.OperationId;
  
  import com.google.common.base.Preconditions;
  import com.google.common.cache.Cache;
  import com.google.common.cache.CacheBuilder;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.File;
  import java.io.IOException;
  import java.io.OutputStream;
  import java.nio.file.Path;
  import java.nio.file.Paths;
  import java.util.ArrayDeque;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Queue;
  import java.util.Set;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.ExecutorService;
  import java.util.function.Supplier;
  import java.util.stream.Collectors;
  import java.util.stream.Stream;
  import javax.annotation.Nullable;
  
  /**
   * Class for managing persistent inode tree state.
   *
   * This class owns all persistent inode tree state, and all inode tree modifications must go through
   * this class. To modify the inode tree, create a journal entry and call one of the applyAndJournal
   * methods.
   */
  public class InodeTreePersistentState implements Journaled {
    private static final Logger LOG = LoggerFactory.getLogger(InodeTreePersistentState.class);
  
    private final InodeStore mInodeStore;
    private final InodeLockManager mInodeLockManager;
  
    private final boolean mRetryCacheEnabled;
    /**
     * This cache is used by this persistent state and InodeTree
     * for keeping track of recently applied operations on inodes.
     */
    private final Cache<OperationId, Boolean> mOpIdCache;
  
    /**
     * A set of inode ids representing pinned inode files. These are not part of the journaled state,
     * but this class keeps the set of pinned inodes up to date whenever the inode tree is modified.
     *
     * This class owns this set, and no other class can modify the set.
     */
    private final PinnedInodeFileIds mPinnedInodeFileIds = new PinnedInodeFileIds();
  
    /** A set of inode ids whose replication max value is non-default. */
    private final ReplicationLimitedFileIds mReplicationLimitedFileIds =
        new ReplicationLimitedFileIds();
  
    /** Counter for tracking how many inodes we have. */
    private final InodeCounter mInodeCounter = new InodeCounter();
  
    /** A set of inode ids whose persistence state is {@link PersistenceState#TO_BE_PERSISTED}. */
    private final ToBePersistedFileIds mToBePersistedIds = new ToBePersistedFileIds();
  
    /**
     * TTL bucket list. The list is owned by InodeTree, and is only shared with
     * InodeTreePersistentState so that the list can be updated whenever inode tree state changes.
     */
    // TODO(andrew): Move ownership of the ttl bucket list to this class
    private final TtlBucketList mTtlBuckets;
  
    private final BucketCounter mBucketCounter;
  
    /**
     * @param inodeStore file store which holds inode metadata
     * @param lockManager manager for inode locks
     * @param ttlBucketList reference to the ttl bucket list so that the list can be updated when the
     *        inode tree is modified
     */
    public InodeTreePersistentState(InodeStore inodeStore, InodeLockManager lockManager,
        TtlBucketList ttlBucketList) {
      mInodeStore = inodeStore;
      mInodeLockManager = lockManager;
      mTtlBuckets = ttlBucketList;
      mBucketCounter = new BucketCounter(
          Configuration.getList(MASTER_METRICS_FILE_SIZE_DISTRIBUTION_BUCKETS)
              .stream().map(FormatUtils::parseSpaceSize).collect(Collectors.toList()));
      mRetryCacheEnabled =
          Configuration.getBoolean(MASTER_FILE_SYSTEM_OPERATION_RETRY_CACHE_ENABLED);
      mOpIdCache = CacheBuilder.newBuilder()
          .maximumSize(Configuration.getInt(MASTER_FILE_SYSTEM_OPERATION_RETRY_CACHE_SIZE))
          .build();
    }
  
    /**
     * 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 mRetryCacheEnabled && opId != null && (null != mOpIdCache.getIfPresent(opId));
    }
  
    /**
     * Used to mark an operation as complete in retry-cache.
     * @param opId the operation id
     */
    public void cacheOperation(@Nullable OperationId opId) {
      if (mRetryCacheEnabled && opId != null) {
        mOpIdCache.put(opId, true);
      }
    }
  
    /**
     * @return an unmodifiable view of the replication limited file ids
     */
    public Set<Long> getReplicationLimitedFileIds() {
      return Collections.unmodifiableSet(mReplicationLimitedFileIds);
    }
  
    /**
     * @return the root of the inode tree
     */
    public InodeDirectory getRoot() {
      return mInodeStore.get(0).map(Inode::asDirectory).orElse(null);
    }
  
    /**
     * @return the pinned inode file ids;
     */
    public Set<Long> getPinnedInodeFileIds() {
      return Collections.unmodifiableSet(mPinnedInodeFileIds);
    }
  
    /**
     * @return the number of inodes in the tree
     */
    public long getInodeCount() {
      return mInodeCounter.longValue();
    }
  
    /**
     * @return the file size distribution in the tree
     */
    public Map<Long, Number> getFileSizeHistogram() {
      return mBucketCounter.getCounters();
    }
  
    /**
     * @return an unmodifiable view of the files with persistence state
     *         {@link PersistenceState#TO_BE_PERSISTED}
     */
    public Set<Long> getToBePersistedIds() {
      return Collections.unmodifiableSet(mToBePersistedIds);
    }
  
    /**
     * @return the list of TTL buckets for tracking inode TTLs
     */
    public TtlBucketList getTtlBuckets() {
      return mTtlBuckets;
    }
  
    ////
    /// The applyAndJournal() methods make sure the in-memory metadata state and the journal are
    /// BOTH updated. Any exception seen here will crash the master! So if an exception should be
    /// tolerated, make sure it is caught and handled properly.
    /// However, the journal flushing is async and happens when the JournalContext is closed.
    /// That means when applyAndJournal() returns, the metadata states are updated but the
    /// journal entries are not necessarily persisted or sent to standby masters.
    /// Therefore, to ensure atomicity, callers should make sure the JournalContext is closed
    /// after calling the applyAndJournal() methods.
    ////
  
    /**
     * Deletes an inode (may be either a file or directory).
     *
     * @param context journal context supplier
     * @param entry delete file entry
     */
    public void applyAndJournal(Supplier<JournalContext> context, DeleteFileEntry entry) {
      // Unlike most entries, the delete file entry must be applied *before* making the in-memory
      // change. This is because delete file and create file are performed with only a read lock on
      // the parent directory. As soon as we do the in-memory-delete, another thread could re-create a
      // directory with the same name, and append a journal entry for the inode creation. This would
      // ruin journal replay because we would see two create file entries in a row for the same file
      // name. The opposite order is safe. We will never append the delete entry for a file before its
      // creation entry because delete requires a write lock on the deleted file, but the create
      // operation holds that lock until after it has appended to the journal.
      try {
        JournalEntry.Builder builder = JournalEntry.newBuilder().setDeleteFile(entry);
        OperationId opId = getOpId(context);
        if (opId != null) {
          builder.setOperationId(opId.toJournalProto());
        }
        context.get().append(builder.build());
        applyDelete(entry);
      } catch (Throwable t) {
        // Delete entries should always apply cleanly, but if it somehow fails, we are in a state
        // where we've journaled the delete, but failed to make the in-memory update. We don't yet
        // have a way to recover from this, so we give a fatal error.
        ProcessUtils.fatalError(LOG, t, "Failed to apply entry %s", entry);
      }
    }
  
    /**
     * Allocates and returns the next block ID for the indicated inode.
     *
     * @param context journal context supplier
     * @param entry new block entry
     * @return the new block id
     */
    public long applyAndJournal(Supplier<JournalContext> context, NewBlockEntry entry) {
      try {
        long id = applyNewBlock(entry);
        context.get().append(JournalEntry.newBuilder().setNewBlock(entry).build());
        return id;
      } catch (Throwable t) {
        ProcessUtils.fatalError(LOG, t, "Failed to apply %s", entry);
        throw t; // fatalError will usually system.exit
      }
    }
  
    /**
     * Renames an inode.
     *
     * @param context journal context supplier
     * @param entry rename entry
     */
    public void applyAndJournal(Supplier<JournalContext> context, RenameEntry entry) {
      try {
        applyRename(entry);
        JournalEntry.Builder builder = JournalEntry.newBuilder().setRename(entry);
        OperationId opId = getOpId(context);
        if (opId != null) {
          builder.setOperationId(opId.toJournalProto());
        }
        context.get().append(builder.build());
      } catch (Throwable t) {
        ProcessUtils.fatalError(LOG, t, "Failed to apply %s", entry);
        throw t; // fatalError will usually system.exit
      }
    }
  
    /**
     * Sets an ACL for an inode.
     *
     * @param context journal context supplier
     * @param entry set acl entry
     */
    public void applyAndJournal(Supplier<JournalContext> context, SetAclEntry entry) {
      try {
        applySetAcl(entry);
        context.get().append(JournalEntry.newBuilder().setSetAcl(entry).build());
      } catch (Throwable t) {
        ProcessUtils.fatalError(LOG, t, "Failed to apply %s", entry);
        throw t; // fatalError will usually system.exit
      }
    }
  
    /**
     * Updates an inode's state. This is used for state common to both files and directories.
     *
     * @param context journal context supplier
     * @param entry update inode entry
     */
    public void applyAndJournal(Supplier<JournalContext> context, UpdateInodeEntry entry) {
      try {
        applyUpdateInode(entry);
        JournalEntry.Builder builder = JournalEntry.newBuilder().setUpdateInode(entry);
        OperationId opId = getOpId(context);
        if (opId != null) {
          builder.setOperationId(opId.toJournalProto());
        }
        context.get().append(builder.build());
      } catch (Throwable t) {
        ProcessUtils.fatalError(LOG, t, "Failed to apply %s", entry);
        throw t; // fatalError will usually system.exit
      }
    }
  
    /**
     * Updates an inode directory's state.
     *
     * @param context journal context supplier
     * @param entry update inode directory entry
     */
    public void applyAndJournal(Supplier<JournalContext> context, UpdateInodeDirectoryEntry entry) {
      try {
        applyUpdateInodeDirectory(entry);
        context.get().append(JournalEntry.newBuilder().setUpdateInodeDirectory(entry).build());
      } catch (Throwable t) {
        ProcessUtils.fatalError(LOG, t, "Failed to apply %s", entry);
        throw t; // fatalError will usually system.exit
      }
    }
  
    /**
     * Updates an inode file's state.
     *
     * @param context journal context supplier
     * @param entry update inode file entry
     */
    public void applyAndJournal(Supplier<JournalContext> context, UpdateInodeFileEntry entry) {
      try {
        applyUpdateInodeFile(entry);
        context.get().append(JournalEntry.newBuilder().setUpdateInodeFile(entry).build());
      } catch (Throwable t) {
        ProcessUtils.fatalError(LOG, t, "Failed to apply %s", entry);
        throw t; // fatalError will usually system.exit
      }
    }
  
    /**
     * Adds an inode to the inode tree.
     *
     * @param context journal context supplier
     * @param inode an inode to add and create a journal entry for
     * @param path path of the new inode
     */
    public void applyAndJournal(Supplier<JournalContext> context, MutableInode<?> inode,
        String path) {
      try {
        applyCreateInode(inode);
        JournalEntry.Builder builder =
            inode.toJournalEntry(Preconditions.checkNotNull(path)).toBuilder();
        OperationId opId = getOpId(context);
        if (opId != null) {
          builder.setOperationId(opId.toJournalProto());
        }
        context.get().append(builder.build());
      } catch (Throwable t) {
        ProcessUtils.fatalError(LOG, t, "Failed to apply %s", inode);
        throw t; // fatalError will usually system.exit
      }
    }
  
    /**
     * Updates last access time for Inode without journaling. The caller should apply the journal
     * entry separately.
     *
     * @param inodeId the id of the target inode
     * @param accessTime the new value for last access time
     * @return the journal entry that represents the update
     */
    public UpdateInodeEntry applyInodeAccessTime(long inodeId, long accessTime) {
      UpdateInodeEntry entry = UpdateInodeEntry.newBuilder().setId(inodeId)
          .setLastAccessTimeMs(accessTime).build();
      applyUpdateInode(entry);
      return entry;
    }
  
    ////
    /// Apply Implementations. These methods are used for journal replay, so they are not allowed to
    /// fail. They are also used when making metadata changes during regular operation.
    /// If the method throws an exception, the caller applyAndJournal() method will crash the master.
    ////
  
    private void applyDelete(DeleteFileEntry entry) {
      long id = entry.getId();
      Inode inode = mInodeStore.get(id).get();
  
      // The recursive option is only used by old versions.
      if (inode.isDirectory() && entry.getRecursive()) {
        Queue<InodeDirectory> dirsToDelete = new ArrayDeque<>();
        dirsToDelete.add(inode.asDirectory());
        while (!dirsToDelete.isEmpty()) {
          InodeDirectory dir = dirsToDelete.poll();
          mInodeStore.removeInodeAndParentEdge(inode);
          mInodeCounter.decrement();
          try (CloseableIterator<? extends Inode> it = mInodeStore.getChildren(dir)) {
            while (it.hasNext()) {
              Inode child = it.next();
              if (child.isDirectory()) {
                dirsToDelete.add(child.asDirectory());
              } else {
                mInodeStore.removeInodeAndParentEdge(inode);
                mInodeCounter.decrement();
              }
            }
          }
        }
      } else {
        mInodeStore.removeInodeAndParentEdge(inode);
        mInodeCounter.decrement();
      }
      if (inode.isFile()) {
        mBucketCounter.remove(inode.asFile().getLength());
      }
      updateTimestampsAndChildCount(inode.getParentId(), entry.getOpTimeMs(), -1);
      mPinnedInodeFileIds.remove(id);
      mReplicationLimitedFileIds.remove(id);
      mToBePersistedIds.remove(id);
      mTtlBuckets.remove(inode);
    }
  
    private void applyCreateDirectory(InodeDirectoryEntry entry) {
      applyCreateInode(MutableInodeDirectory.fromJournalEntry(entry));
    }
  
    private void applyCreateFile(InodeFileEntry entry) {
      applyCreateInode(MutableInodeFile.fromJournalEntry(entry));
    }
  
    private long applyNewBlock(NewBlockEntry entry) {
      MutableInodeFile inode = mInodeStore.getMutable(entry.getId()).get().asFile();
      long newBlockId = inode.getNewBlockId();
      mInodeStore.writeInode(inode);
      return newBlockId;
    }
  
    private void applySetAcl(SetAclEntry entry) {
      MutableInode<?> inode = mInodeStore.getMutable(entry.getId()).get();
      List<AclEntry> entries = StreamUtils.map(ProtoUtils::fromProto, entry.getEntriesList());
      switch (entry.getAction()) {
        case REPLACE:
          // fully replace the acl for the path
          inode.replaceAcl(entries);
          break;
        case MODIFY:
          inode.setAcl(entries);
          break;
        case REMOVE:
          inode.removeAcl(entries);
          break;
        case REMOVE_ALL:
          inode.removeExtendedAcl();
          break;
        case REMOVE_DEFAULT:
          inode.setDefaultACL(new DefaultAccessControlList(inode.getACL()));
          break;
        default:
          LOG.warn("Unrecognized acl action: " + entry.getAction());
      }
      mInodeStore.writeInode(inode);
    }
  
    private void applyUpdateInode(UpdateInodeEntry entry) {
      Optional<MutableInode<?>> inodeOpt = mInodeStore.getMutable(entry.getId());
      if (!inodeOpt.isPresent()) {
        if (isJournalUpdateAsync(entry)) {
          // do not throw if the entry is journaled asynchronously
          return;
        }
        throw new IllegalStateException("Inode " + entry.getId() + " not found");
      }
      MutableInode<?> inode = inodeOpt.get();
      if (entry.hasTtl()) {
        // Remove before updating the inode. #remove relies on the inode having the same
        // TTL as when it was inserted.
        mTtlBuckets.remove(inode);
      }
      inode.updateFromEntry(entry);
      if (entry.hasTtl()) {
        mTtlBuckets.insert(Inode.wrap(inode));
      }
      if (inode.isFile() && entry.hasPinned()) {
        setReplicationForPin(inode, entry.getPinned());
      }
      mInodeStore.writeInode(inode);
      updateToBePersistedIds(inode);
    }
  
    private void setReplicationForPin(MutableInode<?> inode, boolean pinned) {
      if (inode.isFile()) {
        MutableInodeFile file = inode.asFile();
        if (pinned) {
          // when we pin a file with default min replication (zero), we bump the min replication
          // to one in addition to setting pinned flag, and adjust the max replication if it is
          // smaller than min replication.
          file.setPinned(true);
          if (file.getReplicationMin() == 0) {
            file.setReplicationMin(1);
          }
          if (file.getReplicationMax() == 0) {
            file.setReplicationMax(alluxio.Constants.REPLICATION_MAX_INFINITY);
          }
          mPinnedInodeFileIds.add(inode.getId());
        } else {
          // when we unpin a file, set the min replication to zero too.
          file.setReplicationMin(0);
          mPinnedInodeFileIds.remove(file.getId());
        }
        if (file.getReplicationMax() != alluxio.Constants.REPLICATION_MAX_INFINITY) {
          mReplicationLimitedFileIds.add(file.getId());
        }
      }
    }
  
    private OperationId getOpId(Supplier<JournalContext> context) {
      if (context instanceof RpcContext) {
        return ((RpcContext) context).getOpId();
      }
      return null;
    }
  
    /**
     * @param entry the update inode journal entry to be checked
     * @return whether the journal entry might be applied asynchronously out of order
     */
    private boolean isJournalUpdateAsync(UpdateInodeEntry entry) {
      return entry.getAllFields().size() == 2 && entry.hasId() && entry.hasLastAccessTimeMs();
    }
  
    private void applyUpdateInodeDirectory(UpdateInodeDirectoryEntry entry) {
      MutableInode<?> inode = mInodeStore.getMutable(entry.getId()).get();
      Preconditions.checkState(inode.isDirectory(),
          "Encountered non-directory id in update directory entry %s", entry);
  
      inode.asDirectory().updateFromEntry(entry);
      mInodeStore.writeInode(inode);
    }
  
    private void applyUpdateInodeFile(UpdateInodeFileEntry entry) {
      MutableInode<?> inode = mInodeStore.getMutable(entry.getId()).get();
      Preconditions.checkState(inode.isFile(), "Encountered non-file id in update file entry %s",
          entry);
      if (entry.hasReplicationMax()) {
        if (entry.getReplicationMax() == alluxio.Constants.REPLICATION_MAX_INFINITY) {
          mReplicationLimitedFileIds.remove(inode.getId());
        } else {
          mReplicationLimitedFileIds.add(inode.getId());
        }
      }
      if (inode.asFile().isCompleted()) {
        mBucketCounter.remove(inode.asFile().getLength());
      }
      inode.asFile().updateFromEntry(entry);
      mInodeStore.writeInode(inode);
      mBucketCounter.insert(inode.asFile().getLength());
    }
  
    ////
    /// Deprecated Entries
    ////
  
    private void applyAsyncPersist(AsyncPersistRequestEntry entry) {
      applyUpdateInode(UpdateInodeEntry.newBuilder().setId(entry.getFileId())
          .setPersistenceState(PersistenceState.TO_BE_PERSISTED.name()).build());
    }
  
    private void applyCompleteFile(CompleteFileEntry entry) {
      applyUpdateInode(UpdateInodeEntry.newBuilder().setId(entry.getId())
          .setLastModificationTimeMs(entry.getOpTimeMs()).setOverwriteModificationTime(true)
          .setUfsFingerprint(entry.getUfsFingerprint()).build());
      applyUpdateInodeFile(UpdateInodeFileEntry.newBuilder().setId(entry.getId())
          .setLength(entry.getLength()).addAllSetBlocks(entry.getBlockIdsList()).build());
    }
  
    private void applyInodeLastModificationTime(InodeLastModificationTimeEntry entry) {
      // This entry is deprecated, use UpdateInode instead.
      applyUpdateInode(UpdateInodeEntry.newBuilder().setId(entry.getId())
          .setLastModificationTimeMs(entry.getLastModificationTimeMs()).build());
    }
  
    private void applyPersistDirectory(PersistDirectoryEntry entry) {
      // This entry is deprecated, use UpdateInode instead.
      applyUpdateInode(UpdateInodeEntry.newBuilder().setId(entry.getId())
          .setPersistenceState(PersistenceState.PERSISTED.name()).build());
    }
  
    private void applySetAttribute(SetAttributeEntry entry) {
      Builder builder = UpdateInodeEntry.newBuilder();
      builder.setId(entry.getId());
      if (entry.hasGroup()) {
        builder.setGroup(entry.getGroup());
      }
      if (entry.hasOpTimeMs()) {
        builder.setLastModificationTimeMs(entry.getOpTimeMs());
      }
      if (entry.hasOwner()) {
        builder.setOwner(entry.getOwner());
      }
      if (entry.hasPermission()) {
        builder.setMode((short) entry.getPermission());
      }
      if (entry.hasPersisted()) {
        if (entry.getPersisted()) {
          builder.setPersistenceState(PersistenceState.PERSISTED.name());
        } else {
          builder.setPersistenceState(PersistenceState.NOT_PERSISTED.name());
        }
      }
      if (entry.hasPinned()) {
        builder.setPinned(entry.getPinned());
      }
      if (entry.hasTtl()) {
        builder.setTtl(entry.getTtl());
        builder.setTtlAction(entry.getTtlAction());
      }
      if (entry.hasUfsFingerprint()) {
        builder.setUfsFingerprint(entry.getUfsFingerprint());
      }
      applyUpdateInode(builder.build());
    }
  
    ////
    // Helper methods
    ////
  
    private void applyCreateInode(MutableInode<?> inode) {
      if (inode.isDirectory() && inode.getName().equals(InodeTree.ROOT_INODE_NAME)) {
        // This is the root inode. Clear all the state, and set the root.
        mInodeStore.clear();
        mInodeStore.writeNewInode(inode);
        mInodeCounter.reset();
        mInodeCounter.increment();
        mPinnedInodeFileIds.clear();
        mReplicationLimitedFileIds.clear();
        mToBePersistedIds.clear();
  
        updateToBePersistedIds(inode);
        return;
      }
      // inode should be added to the inode store before getting added to its parent list, because it
      // becomes visible at this point.
      mInodeStore.writeNewInode(inode);
      mInodeCounter.increment();
      mInodeStore.addChild(inode.getParentId(), inode);
      // Only update size, last modified time is updated separately.
      updateTimestampsAndChildCount(inode.getParentId(), Long.MIN_VALUE, 1);
      if (inode.isFile()) {
        boolean pinned = inode.asFile().isPinned() || inode.asFile().getReplicationMin() > 0;
        setReplicationForPin(inode, pinned);
      }
      // Add the file to TTL buckets, the insert automatically rejects files w/ Constants.NO_TTL
      mTtlBuckets.insert(Inode.wrap(inode));
      updateToBePersistedIds(inode);
      if (inode.isFile() && inode.asFile().isCompleted()) {
        mBucketCounter.insert(inode.asFile().getLength());
      }
    }
  
    private void applyRename(RenameEntry entry) {
      if (entry.hasDstPath()) {
        entry = rewriteDeprecatedRenameEntry(entry);
      }
  
      MutableInode<?> inode = mInodeStore.getMutable(entry.getId()).get();
      long oldParent = inode.getParentId();
      long newParent = entry.getNewParentId();
  
      mInodeStore.removeChild(oldParent, inode.getName());
      inode.setName(entry.getNewName());
      mInodeStore.addChild(newParent, inode);
      inode.setParentId(newParent);
      mInodeStore.writeInode(inode);
  
      if (oldParent == newParent) {
        updateTimestampsAndChildCount(oldParent, entry.getOpTimeMs(), 0);
      } else {
        updateTimestampsAndChildCount(oldParent, entry.getOpTimeMs(), -1);
        updateTimestampsAndChildCount(newParent, entry.getOpTimeMs(), 1);
      }
    }
  
    /**
     * Updates the last modification time and last access time for the indicated inode directory,
     * and updates its child count.
     *
     * If the inode's timestamps are already greater than the specified time, the inode's timestamps
     * will not be changed.
     *
     * @param id the inode to update
     * @param opTimeMs the time of the operation that modified the inode
     * @param deltaChildCount the change in inode directory child count
     */
    private void updateTimestampsAndChildCount(long id, long opTimeMs, long deltaChildCount) {
      try (LockResource lr = mInodeLockManager.lockUpdate(id)) {
        MutableInodeDirectory inode = mInodeStore.getMutable(id).get().asDirectory();
        boolean madeUpdate = false;
        if (inode.getLastModificationTimeMs() < opTimeMs) {
          inode.setLastModificationTimeMs(opTimeMs);
          madeUpdate = true;
        }
        if (inode.getLastAccessTimeMs() < opTimeMs) {
          inode.setLastAccessTimeMs(opTimeMs);
          madeUpdate = true;
        }
        if (deltaChildCount != 0) {
          inode.setChildCount(inode.getChildCount() + deltaChildCount);
          madeUpdate = true;
        }
        if (madeUpdate) {
          mInodeStore.writeInode(inode);
        }
      }
    }
  
    private void updateToBePersistedIds(MutableInode<?> inode) {
      if (inode.getPersistenceState() == PersistenceState.TO_BE_PERSISTED) {
        mToBePersistedIds.add(inode.getId());
      } else {
        mToBePersistedIds.remove(inode.getId());
      }
    }
  
    private RenameEntry rewriteDeprecatedRenameEntry(RenameEntry entry) {
      Preconditions.checkState(!entry.hasNewName(),
          "old-style rename entries should not have the newName field set");
      Preconditions.checkState(!entry.hasNewParentId(),
          "old-style rename entries should not have the newParentId field set");
      Path path = Paths.get(entry.getDstPath());
      Path parent = path.getParent();
      Path filename = path.getFileName();
      if (parent == null) {
        throw new NullPointerException("path parent cannot be null");
      }
      if (filename == null) {
        throw new NullPointerException("path filename cannot be null");
      }
  
      return RenameEntry.newBuilder().setId(entry.getId())
          .setNewParentId(getIdFromPath(parent))
          .setNewName(filename.toString())
          .setOpTimeMs(entry.getOpTimeMs()).build();
    }
  
    private long getIdFromPath(Path path) {
      Inode curr = getRoot();
      for (Path component : path) {
        curr = mInodeStore.getChild(curr.asDirectory(), component.toString()).get();
      }
      return curr.getId();
    }
  
    @Override
    public boolean processJournalEntry(JournalEntry entry) {
      // Apply entry.
      if (entry.hasDeleteFile()) {
        applyDelete(entry.getDeleteFile());
      } else if (entry.hasInodeDirectory()) {
        applyCreateDirectory(entry.getInodeDirectory());
      } else if (entry.hasInodeFile()) {
        applyCreateFile(entry.getInodeFile());
      } else if (entry.hasNewBlock()) {
        applyNewBlock(entry.getNewBlock());
      } else if (entry.hasRename()) {
        applyRename(entry.getRename());
      } else if (entry.hasSetAcl()) {
        applySetAcl(entry.getSetAcl());
      } else if (entry.hasUpdateInode()) {
        applyUpdateInode(entry.getUpdateInode());
      } else if (entry.hasUpdateInodeDirectory()) {
        applyUpdateInodeDirectory(entry.getUpdateInodeDirectory());
      } else if (entry.hasUpdateInodeFile()) {
        applyUpdateInodeFile(entry.getUpdateInodeFile());
        // Deprecated entries
      } else if (entry.hasAsyncPersistRequest()) {
        applyAsyncPersist(entry.getAsyncPersistRequest());
      } else if (entry.hasCompleteFile()) {
        applyCompleteFile(entry.getCompleteFile());
      } else if (entry.hasInodeLastModificationTime()) {
        applyInodeLastModificationTime(entry.getInodeLastModificationTime());
      } else if (entry.hasPersistDirectory()) {
        applyPersistDirectory(entry.getPersistDirectory());
      } else if (entry.hasSetAttribute()) {
        applySetAttribute(entry.getSetAttribute());
      } else {
        return false;
      }
  
      // Account for entry in retry-cache before returning.
      if (entry.hasOperationId()) {
        cacheOperation(OperationId.fromJournalProto(entry.getOperationId()));
      }
      return true;
    }
  
    @Override
    public void resetState() {
      mInodeStore.clear();
      mReplicationLimitedFileIds.clear();
      mPinnedInodeFileIds.clear();
      mOpIdCache.invalidateAll();
    }
  
    @Override
    public CompletableFuture<Void> writeToCheckpoint(File directory,
                                                     ExecutorService executorService) {
      return CompletableFuture.allOf(Stream.of(mInodeStore, mPinnedInodeFileIds,
          mReplicationLimitedFileIds, mToBePersistedIds, mTtlBuckets, mInodeCounter)
          .map(journaled -> journaled.writeToCheckpoint(directory, executorService))
          .toArray(CompletableFuture[]::new));
    }
  
    @Override
    public void writeToCheckpoint(OutputStream output) throws IOException, InterruptedException {
      // mTtlBuckets must come after mInodeStore so that it can query the inode store to resolve inode
      // ids to inodes.
      JournalUtils.writeToCheckpoint(output, Arrays.asList(mInodeStore, mPinnedInodeFileIds,
          mReplicationLimitedFileIds, mToBePersistedIds, mTtlBuckets, mInodeCounter));
    }
  
    @Override
    public CompletableFuture<Void> restoreFromCheckpoint(File directory,
                                                         ExecutorService executorService) {
      return CompletableFuture.allOf(Stream.of(mInodeStore, mPinnedInodeFileIds,
          mReplicationLimitedFileIds, mToBePersistedIds, mTtlBuckets, mInodeCounter)
          .map(journaled -> journaled.restoreFromCheckpoint(directory, executorService))
          .toArray(CompletableFuture[]::new));
    }
  
    @Override
    public void restoreFromCheckpoint(CheckpointInputStream input) throws IOException {
      // mTtlBuckets must come after mInodeStore so that it can query the inode store to resolve inode
      // ids to inodes.
      JournalUtils.restoreFromCheckpoint(input, Arrays.asList(mInodeStore, mPinnedInodeFileIds,
          mReplicationLimitedFileIds, mToBePersistedIds, mTtlBuckets, mInodeCounter));
    }
  
    @Override
    public CloseableIterator<JournalEntry> getJournalEntryIterator() {
      return InodeTreeBufferedIterator.create(mInodeStore, getRoot());
    }
  
    @Override
    public CheckpointName getCheckpointName() {
      return CheckpointName.INODE_TREE;
    }
  }
  