/*
 * 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 org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;

import alluxio.AlluxioTestDirectory;
import alluxio.AlluxioURI;
import alluxio.ConfigurationRule;
import alluxio.Constants;
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.grpc.CreateDirectoryPOptions;
import alluxio.grpc.CreateFilePOptions;
import alluxio.master.CoreMasterContext;
import alluxio.master.MasterRegistry;
import alluxio.master.MasterTestUtils;
import alluxio.master.block.BlockMaster;
import alluxio.master.block.BlockMasterFactory;
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.file.meta.InodeTree.LockPattern;
import alluxio.master.file.meta.options.MountInfo;
import alluxio.master.journal.NoopJournalContext;
import alluxio.master.metastore.InodeStore;
import alluxio.master.metastore.caching.CachingInodeStore;
import alluxio.master.metastore.heap.HeapInodeStore;
import alluxio.master.metastore.rocks.RocksInodeStore;
import alluxio.master.metrics.MetricsMaster;
import alluxio.master.metrics.MetricsMasterFactory;
import alluxio.proto.journal.Journal;
import alluxio.resource.CloseableIterator;
import alluxio.security.authorization.Mode;
import alluxio.underfs.UfsManager;
import alluxio.util.CommonUtils;
import alluxio.util.StreamUtils;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Sets;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.rules.ExpectedException;
import org.junit.rules.TemporaryFolder;
import org.junit.runners.Parameterized.Parameters;

import java.io.IOException;
import java.time.Clock;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

/**
 * Unit tests for {@link InodeTree}.
 * @deprecated
 */
//@RunWith(Parameterized.class)
@Deprecated
public final class InodeTreeTest {
  private static final String TEST_PATH = "test";
  private static final AlluxioURI TEST_URI = new AlluxioURI("/test");
  private static final AlluxioURI NESTED_URI = new AlluxioURI("/nested/test");
  private static final AlluxioURI NESTED_DIR_URI = new AlluxioURI("/nested/test/dir");
  private static final AlluxioURI NESTED_DIR_FILE_URI = new AlluxioURI("/nested/test/dir/file1");
  private static final AlluxioURI NESTED_FILE_URI = new AlluxioURI("/nested/test/file");
  private static final AlluxioURI NESTED_MULTIDIR_FILE_URI
      = new AlluxioURI("/nested/test/dira/dirb/file");
  public static final String TEST_OWNER = "user1";
  public static final String TEST_GROUP = "group1";
  public static final Mode TEST_DIR_MODE = new Mode((short) 0755);
  public static final Mode TEST_FILE_MODE = new Mode((short) 0644);
  private static CreateFileContext sFileContext;
  private static CreateDirectoryContext sDirectoryContext;
  private static CreateFileContext sNestedFileContext;
  private static CreateDirectoryContext sNestedDirectoryContext;
  private InodeStore mInodeStore;
  private InodeTree mTree;
  private MasterRegistry mRegistry;
  private MetricsMaster mMetricsMaster;

  private ExecutorService mThreadPool;

  @Parameters
  public static Iterable<Supplier<InodeStore>> parameters() throws Exception {
    String dir =
            AlluxioTestDirectory.createTemporaryDirectory("inode-tree-test").getAbsolutePath();
    return Arrays.asList(
        () -> new CachingInodeStore(new RocksInodeStore(dir), new InodeLockManager()),
        () -> new CachingInodeStore(new HeapInodeStore(), new InodeLockManager()),
        HeapInodeStore::new,
        () -> new RocksInodeStore(dir));
  }

  public InodeTreeTest(Supplier<InodeStore> supplier) {
    mInodeStore = supplier.get();
  }

  /** Rule to create a new temporary folder during each test. */
  @Rule
  public TemporaryFolder mTestFolder = new TemporaryFolder();

  /** The exception expected to be thrown. */
  @Rule
  public ExpectedException mThrown = ExpectedException.none();

  @Rule
  public ConfigurationRule mConfigurationRule =
      new ConfigurationRule(new ImmutableMap.Builder<PropertyKey, Object>()
          .put(PropertyKey.SECURITY_AUTHORIZATION_PERMISSION_ENABLED, true)
          .put(PropertyKey.SECURITY_AUTHORIZATION_PERMISSION_SUPERGROUP, "test-supergroup")
          .build(), Configuration.modifiableGlobal());

  /**
   * Sets up all dependencies before a test runs.
   */
  @Before
  public void before() throws Exception {
    mRegistry = new MasterRegistry();
    CoreMasterContext context = MasterTestUtils.testMasterContext();
    mMetricsMaster = new MetricsMasterFactory().create(mRegistry, context);
    mRegistry.add(MetricsMaster.class, mMetricsMaster);
    BlockMaster blockMaster = new BlockMasterFactory().create(mRegistry, context);
    InodeDirectoryIdGenerator directoryIdGenerator =
        new InodeDirectoryIdGenerator(blockMaster);
    UfsManager ufsManager = mock(UfsManager.class);
    MountTable mountTable = new MountTable(ufsManager, mock(MountInfo.class), Clock.systemUTC());
    InodeLockManager lockManager = new InodeLockManager();
    mTree = new InodeTree(mInodeStore, blockMaster, directoryIdGenerator, mountTable, lockManager);

    mRegistry.start(true);

    mTree.initializeRoot(TEST_OWNER, TEST_GROUP, TEST_DIR_MODE, NoopJournalContext.INSTANCE);
    mThreadPool = Executors.newCachedThreadPool();
  }

  @After
  public void after() throws Exception {
    mRegistry.stop();
    mInodeStore.close();
    mThreadPool.shutdown();
  }

  /**
   * Sets up dependencies before a single test runs.
   */
  @BeforeClass
  public static void beforeClass() throws Exception {
    sFileContext =
        CreateFileContext
            .mergeFrom(CreateFilePOptions.newBuilder().setBlockSizeBytes(Constants.KB)
                .setMode(TEST_FILE_MODE.toProto()))
            .setOwner(TEST_OWNER).setGroup(TEST_GROUP);
    sDirectoryContext = CreateDirectoryContext
        .mergeFrom(CreateDirectoryPOptions.newBuilder().setMode(TEST_DIR_MODE.toProto()))
        .setOwner(TEST_OWNER).setGroup(TEST_GROUP);
    sNestedFileContext = CreateFileContext
        .mergeFrom(CreateFilePOptions.newBuilder().setBlockSizeBytes(Constants.KB)
            .setMode(TEST_FILE_MODE.toProto()).setRecursive(true))
        .setOwner(TEST_OWNER).setGroup(TEST_GROUP);
    sNestedDirectoryContext = CreateDirectoryContext.mergeFrom(
        CreateDirectoryPOptions.newBuilder().setMode(TEST_DIR_MODE.toProto()).setRecursive(true))
        .setOwner(TEST_OWNER).setGroup(TEST_GROUP);
  }

  /**
   * Tests that initializing the root twice results in the same root.
   * @deprecated
   */
  @Deprecated
  public void initializeRootTwice() throws Exception {
    MutableInode<?> root = getInodeByPath(new AlluxioURI("/"));
    // initializeRoot call does nothing
    mTree.initializeRoot(TEST_OWNER, TEST_GROUP, TEST_DIR_MODE, NoopJournalContext.INSTANCE);
    assertEquals(TEST_OWNER, root.getOwner());
    MutableInode<?> newRoot = getInodeByPath(new AlluxioURI("/"));
    assertEquals(root, newRoot);
  }

  /**
   * Tests the {@link InodeTree#createPath(RpcContext, LockedInodePath, CreatePathContext)}
   * method for creating directories.
   * @deprecated
   */
  @Deprecated
  public void createDirectory() throws Exception {
    // create directory
    createPath(mTree, TEST_URI, sDirectoryContext);
    assertTrue(mTree.inodePathExists(TEST_URI));
    MutableInode<?> test = getInodeByPath(TEST_URI);
    assertEquals(TEST_PATH, test.getName());
    assertTrue(test.isDirectory());
    assertEquals("user1", test.getOwner());
    assertEquals("group1", test.getGroup());
    assertEquals(TEST_DIR_MODE.toShort(), test.getMode());

    // create nested directory
    createPath(mTree, NESTED_URI, sNestedDirectoryContext);
    assertTrue(mTree.inodePathExists(NESTED_URI));
    MutableInode<?> nested = getInodeByPath(NESTED_URI);
    assertEquals(TEST_PATH, nested.getName());
    assertEquals(2, nested.getParentId());
    assertTrue(test.isDirectory());
    assertEquals("user1", test.getOwner());
    assertEquals("group1", test.getGroup());
    assertEquals(TEST_DIR_MODE.toShort(), test.getMode());
  }

  /**
   * Tests that an exception is thrown when trying to create an already existing directory with the
   * {@code allowExists} flag set to {@code false}.
   * @deprecated
   */
  @Deprecated
  public void createExistingDirectory() throws Exception {
    // create directory
    createPath(mTree, TEST_URI, sDirectoryContext);

    // create again with allowExists true
    createPath(mTree, TEST_URI, CreateDirectoryContext
        .mergeFrom(CreateDirectoryPOptions.newBuilder().setAllowExists(true)));

    // create again with allowExists false
    mThrown.expect(FileAlreadyExistsException.class);
    mThrown
        .expectMessage("Not allowed to create directory because path already exists: " + TEST_URI);
    createPath(mTree, TEST_URI, CreateDirectoryContext
        .mergeFrom(CreateDirectoryPOptions.newBuilder().setAllowExists(false)));
  }

  /**
   * Tests that creating a file under a pinned directory works.
   * @deprecated
   */
  @Deprecated
  public void createFileUnderPinnedDirectory() throws Exception {
    // create nested directory
    createPath(mTree, NESTED_URI, sNestedDirectoryContext);

    // pin nested folder
    try (
        LockedInodePath inodePath = mTree.lockFullInodePath(
            NESTED_URI, LockPattern.WRITE_INODE, NoopJournalContext.INSTANCE)
    ) {
      mTree.setPinned(RpcContext.NOOP, inodePath, true, Collections.emptyList(), 0);
    }

    // create nested file under pinned folder
    createPath(mTree, NESTED_FILE_URI, sNestedFileContext);

    // the nested file is pinned
    assertEquals(1, mTree.getPinIdSet().size());
  }

  /**
   * Tests the {@link InodeTree#createPath(RpcContext, LockedInodePath, CreatePathContext)}
   * method for creating a file.
   * @deprecated
   */
  @Deprecated
  public void createFile() throws Exception {
    // created nested file
    createPath(mTree, NESTED_FILE_URI, sNestedFileContext);
    MutableInode<?> nestedFile = getInodeByPath(NESTED_FILE_URI);
    assertEquals("file", nestedFile.getName());
    assertEquals(2, nestedFile.getParentId());
    assertTrue(nestedFile.isFile());
    assertEquals("user1", nestedFile.getOwner());
    assertEquals("group1", nestedFile.getGroup());
    assertEquals(TEST_FILE_MODE.toShort(), nestedFile.getMode());
  }

  /**
   * Tests the {@link InodeTree#createPath(RpcContext, LockedInodePath, CreatePathContext)}
   * method.
   * @deprecated
   */
  @Deprecated
  public void createPathTest() throws Exception {
    // save the last mod time of the root
    long lastModTime = mTree.getRoot().getLastModificationTimeMs();
    // sleep to ensure a different last modification time
    CommonUtils.sleepMs(10);

    // Need to use updated options to set the correct last mod time.
    CreateDirectoryContext dirContext = CreateDirectoryContext.mergeFrom(
        CreateDirectoryPOptions.newBuilder().setRecursive(true).setMode(TEST_DIR_MODE.toProto()))
            .setOwner(TEST_OWNER).setGroup(TEST_GROUP);

    // create nested directory
    List<Inode> created = createPath(mTree, NESTED_URI, dirContext);
    // 1 modified directory
    assertNotEquals(lastModTime,
        getInodeByPath(NESTED_URI.getParent()).getLastModificationTimeMs());
    // 2 created directories
    assertEquals(2, created.size());
    assertEquals("nested", created.get(0).getName());
    assertEquals("test", created.get(1).getName());
    // save the last mod time of 'test'
    lastModTime = created.get(1).getLastModificationTimeMs();
    // sleep to ensure a different last modification time
    CommonUtils.sleepMs(10);

    // creating the directory path again results in no new inodes.
    try {
      createPath(mTree, NESTED_URI, dirContext);
      fail("createPath should throw FileAlreadyExistsException");
    } catch (FileAlreadyExistsException e) {
      assertEquals("Not allowed to create directory because path already exists: " + NESTED_URI,
          e.getMessage());
    }

    // create a file
    CreateFileContext options = CreateFileContext.mergeFrom(
        CreateFilePOptions.newBuilder().setBlockSizeBytes(Constants.KB).setRecursive(true));
    created = createPath(mTree, NESTED_FILE_URI, options);

    // test directory was modified
    assertNotEquals(lastModTime, getInodeByPath(NESTED_URI).getLastModificationTimeMs());
    // file was created
    assertEquals(1, created.size());
    assertEquals("file", created.get(0).getName());

    // creating the file path again results in no new inodes.
    try {
      createPath(mTree, NESTED_FILE_URI, options);
      fail("createPath should throw FileAlreadyExistsException");
    } catch (FileAlreadyExistsException e) {
      assertEquals("Not allowed to create file because path already exists: " + NESTED_FILE_URI,
          e.getMessage());
    }
  }

  /**
   * Tests the {@link InodeTree#createPath(RpcContext, LockedInodePath, CreatePathContext)} method
   * for inheriting owner and group when empty.
   * @deprecated
   */
  @Deprecated
  public void createPathInheritanceTest() throws Exception {
    // create nested directory
    CreateDirectoryContext dirContext = CreateDirectoryContext.mergeFrom(
        CreateDirectoryPOptions.newBuilder().setRecursive(true).setMode(TEST_DIR_MODE.toProto()))
            .setOwner(TEST_OWNER).setGroup(TEST_GROUP);
    List<Inode> created = createPath(mTree, NESTED_URI, dirContext);
    assertEquals(2, created.size());

    // 1. create a nested directory with empty owner and group
    CreateDirectoryContext nestedDirContext = CreateDirectoryContext.mergeFrom(
        CreateDirectoryPOptions.newBuilder().setRecursive(true).setMode(TEST_DIR_MODE.toProto()))
        .setOwner("").setGroup("");
    created = createPath(mTree, NESTED_DIR_URI, nestedDirContext);
    assertEquals(1, created.size());
    assertEquals("dir", created.get(0).getName());
    assertEquals(TEST_OWNER, created.get(0).getOwner());
    assertEquals(TEST_GROUP, created.get(0).getGroup());

    // 2. create a file with empty owner and group
    CreateFileContext nestedDirFileContext = CreateFileContext
        .mergeFrom(
            CreateFilePOptions.newBuilder().setBlockSizeBytes(Constants.KB).setRecursive(true))
        .setOwner("").setGroup("");
    created = createPath(mTree, NESTED_DIR_FILE_URI, nestedDirFileContext);
    assertEquals(1, created.size());
    assertEquals("file1", created.get(0).getName());
    assertEquals(TEST_OWNER, created.get(0).getOwner());
    assertEquals(TEST_GROUP, created.get(0).getGroup());
  }

  /**
   * Tests that an exception is thrown when trying to create the root path twice.
   * @deprecated
   */
  @Deprecated
  public void createRootPath() throws Exception {
    mThrown.expect(FileAlreadyExistsException.class);
    mThrown.expectMessage("/");

    createPath(mTree, new AlluxioURI("/"), sFileContext);
  }

  /**
   * Tests that an exception is thrown when trying to create a file with invalid block size.
   * @deprecated
   */
  @Deprecated
  public void createFileWithInvalidBlockSize() throws Exception {
    mThrown.expect(BlockInfoException.class);
    mThrown.expectMessage("Invalid block size 0");

    CreateFileContext context =
        CreateFileContext.mergeFrom(CreateFilePOptions.newBuilder().setBlockSizeBytes(0));
    createPath(mTree, TEST_URI, context);
  }

  /**
   * Tests that an exception is thrown when trying to create a file with a negative block size.
   * @deprecated
   */
  @Deprecated
  public void createFileWithNegativeBlockSize() throws Exception {
    mThrown.expect(BlockInfoException.class);
    mThrown.expectMessage("Invalid block size -1");

    CreateFileContext context =
        CreateFileContext.mergeFrom(CreateFilePOptions.newBuilder().setBlockSizeBytes(-1));
    createPath(mTree, TEST_URI, context);
  }

  /**
   * Tests that an exception is thrown when trying to create a file under a non-existing directory.
   * @deprecated
   */
  @Deprecated
  public void createFileUnderNonexistingDir() throws Exception {
    mThrown.expect(FileDoesNotExistException.class);
    mThrown.expectMessage("File /nested/test creation failed. Component 1(nested) does not exist");

    createPath(mTree, NESTED_URI, sFileContext);
  }

  /**
   * Tests that an exception is thrown when trying to create a file twice.
   * @deprecated
   */
  @Deprecated
  public void createFileTwice() throws Exception {
    mThrown.expect(FileAlreadyExistsException.class);
    mThrown.expectMessage("/nested/test");

    createPath(mTree, NESTED_URI, sNestedFileContext);
    createPath(mTree, NESTED_URI, sNestedFileContext);
  }

  /**
   * Tests that an exception is thrown when trying to create a file under a file path.
   * @deprecated
   */
  @Deprecated
  public void createFileUnderFile() throws Exception {
    createPath(mTree, NESTED_URI, sNestedFileContext);

    mThrown.expect(InvalidPathException.class);
    mThrown.expectMessage("Traversal failed for path /nested/test/test. "
        + "Component 2(test) is a file, not a directory");
    createPath(mTree, new AlluxioURI("/nested/test/test"), sNestedFileContext);
  }

  /**
   * Tests {@link InodeTree#inodeIdExists(long)}.
   * @deprecated
   */
  @Deprecated
  public void inodeIdExists() throws Exception {
    assertTrue(mTree.inodeIdExists(0));
    assertFalse(mTree.inodeIdExists(1));

    createPath(mTree, TEST_URI, sFileContext);
    MutableInode<?> inode = getInodeByPath(TEST_URI);
    assertTrue(mTree.inodeIdExists(inode.getId()));

    deleteInodeByPath(mTree, TEST_URI);
    assertFalse(mTree.inodeIdExists(inode.getId()));
  }

  /**
   * Tests {@link InodeTree#inodePathExists(AlluxioURI)}.
   * @deprecated
   */
  @Deprecated
  public void inodePathExists() throws Exception {
    assertFalse(mTree.inodePathExists(TEST_URI));

    createPath(mTree, TEST_URI, sFileContext);
    assertTrue(mTree.inodePathExists(TEST_URI));

    deleteInodeByPath(mTree, TEST_URI);
    assertFalse(mTree.inodePathExists(TEST_URI));
  }

  /**
   * Tests that an exception is thrown when trying to get an Inode by a non-existing path.
   * @deprecated
   */
  @Deprecated
  public void getInodeByNonexistingPath() throws Exception {
    mThrown.expect(FileDoesNotExistException.class);
    mThrown.expectMessage("Path \"/test\" does not exist");

    assertFalse(mTree.inodePathExists(TEST_URI));
    getInodeByPath(TEST_URI);
  }

  /**
   * Tests that an exception is thrown when trying to get an Inode by a non-existing, nested
   * path.
   * @deprecated
   */
  @Deprecated
  public void getInodeByNonexistingNestedPath() throws Exception {
    mThrown.expect(FileDoesNotExistException.class);
    mThrown.expectMessage("Path \"/nested/test/file\" does not exist");

    createPath(mTree, NESTED_URI, sNestedDirectoryContext);
    assertFalse(mTree.inodePathExists(NESTED_FILE_URI));
    getInodeByPath(NESTED_FILE_URI);
  }

  /**
   * Tests that an exception is thrown when trying to get an Inode with an invalid id.
   * @deprecated
   */
  @Deprecated
  public void getInodeByInvalidId() throws Exception {
    mThrown.expect(FileDoesNotExistException.class);
    mThrown.expectMessage(ExceptionMessage.INODE_DOES_NOT_EXIST.getMessage(1));

    assertFalse(mTree.inodeIdExists(1));
    try (LockedInodePath inodePath =
             mTree.lockFullInodePath(1, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      // inode exists
    }
  }

  /**
   * Tests the {@link InodeTree#isRootId(long)} method.
   * @deprecated
   */
  @Deprecated
  public void isRootId() {
    assertTrue(mTree.isRootId(0));
    assertFalse(mTree.isRootId(1));
  }

  /**
   * Tests the {@link InodeTree#getPath(InodeView)} method.
   * @deprecated
   */
  @Deprecated
  public void getPath() throws Exception {
    try (LockedInodePath inodePath =
             mTree.lockFullInodePath(0, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      Inode root = inodePath.getInode();
      // test root path
      assertEquals(new AlluxioURI("/"), mTree.getPath(root));
    }

    // test one level
    createPath(mTree, TEST_URI, sDirectoryContext);
    try (LockedInodePath inodePath =
             mTree.lockFullInodePath(TEST_URI, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      assertEquals(new AlluxioURI("/test"), mTree.getPath(inodePath.getInode()));
    }

    // test nesting
    createPath(mTree, NESTED_URI, sNestedDirectoryContext);
    try (LockedInodePath inodePath = mTree.lockFullInodePath(
        NESTED_URI, LockPattern.READ, NoopJournalContext.INSTANCE)
    ) {
      assertEquals(new AlluxioURI("/nested/test"), mTree.getPath(inodePath.getInode()));
    }
  }

  List<AlluxioURI> setupBaseFiles(int fileCount, String parent) throws Exception {
    List<AlluxioURI> files = new ArrayList<>();
    for (int i = 0; i < fileCount; i++) {
      files.add(new AlluxioURI(String.format("%s/%d", parent, i)));
    }

    // insert in reverse order so the inodes are created not in sorted order
    for (int i = files.size() - 1; i >= 0; i--) {
      createPath(mTree, files.get(i), sNestedFileContext);
    }
    return files;
  }

  public void checkChildren(List<AlluxioURI> files) throws Exception {
    int fileCount = files.size();
    int remain = fileCount;
    for (AlluxioURI nxt : files) {
      try (LockedInodePath inodePath = mTree.lockFullInodePath(
          nxt, LockPattern.READ, NoopJournalContext.INSTANCE)) {
        long parentId = inodePath.getInode().getParentId();
        Inode childInode = inodePath.getInode();
        Inode[] childInodes;
        try (CloseableIterator<? extends Inode> fromIter = mInodeStore.getChildrenFrom(parentId,
            childInode.getName())) {
          childInodes = iteratorToStream(fromIter).toArray(Inode[]::new);
        }
        assertEquals(remain, childInodes.length);
        ArrayList<AlluxioURI> remainFiles = files.stream().skip(fileCount - remain)
            .collect(Collectors.toCollection(ArrayList::new));
        for (Inode inode : childInodes) {
          AlluxioURI path = mTree.getPath(inode.getId());
          assertTrue(remainFiles.contains(path));
        }
      }
      remain--;
    }
  }

  @Deprecated
  public void getChildrenPrefix() throws Exception {
    // Test a single file prefix
    String prefix = "afile";
    AlluxioURI file = new AlluxioURI("/" + prefix);
    createPath(mTree, file, sNestedDirectoryContext);
    try (LockedInodePath inodePath = mTree.lockFullInodePath(
        file, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      long parentId = inodePath.getInode().getParentId();
      Inode[] childInodes;
      try (CloseableIterator<? extends Inode> fromIter = mInodeStore.getChildrenPrefix(parentId,
          prefix)) {
        childInodes = iteratorToStream(fromIter).toArray(Inode[]::new);
      }
      assertEquals(1, childInodes.length);
      assertEquals(file, mTree.getPath(childInodes[0].getId()));
    }

    // Test multiple files with the prefix
    prefix = "bfile";
    List<AlluxioURI> files = new ArrayList<>(Arrays.asList(new AlluxioURI("/" + prefix),
        new AlluxioURI("/" + prefix + "1"),
        new AlluxioURI("/" + prefix + "2")));
    // add the files out of order
    for (int i = files.size() - 1; i >= 0; i--) {
      createPath(mTree, files.get(i), sNestedFileContext);
    }
    // add a nested directory that matches the prefix
    String nestedPath = "/" + prefix + "3";
    createPath(mTree, new AlluxioURI(nestedPath + "/nested"), sNestedDirectoryContext);
    files.add(new AlluxioURI(nestedPath));
    // add a file that does not match the prefix
    createPath(mTree, new AlluxioURI("/cfile"), sNestedFileContext);
    try (LockedInodePath inodePath = mTree.lockFullInodePath(
        file, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      long parentId = inodePath.getInode().getParentId();
      Inode[] childInodes;
      try (CloseableIterator<? extends Inode> fromIter = mInodeStore.getChildrenPrefix(parentId,
          prefix)) {
        childInodes = iteratorToStream(fromIter).toArray(Inode[]::new);
      }
      assertEquals(files.size(), childInodes.length);
      for (int i = 0; i < files.size(); i++) {
        assertEquals(files.get(i), mTree.getPath(childInodes[i].getId()));
      }
    }
  }

  @Deprecated
  public void getChildrenPrefixAfter() throws Exception {
    // Test a single file prefix
    String prefix = "afile";
    AlluxioURI file = new AlluxioURI("/" + prefix);
    createPath(mTree, file, sNestedDirectoryContext);
    try (LockedInodePath inodePath = mTree.lockFullInodePath(
        file, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      long parentId = inodePath.getInode().getParentId();
      Inode[] childInodes;
      try (CloseableIterator<? extends Inode> fromIter = mInodeStore.getChildrenPrefixFrom(parentId,
          prefix, "afile")) {
        childInodes = iteratorToStream(fromIter).toArray(Inode[]::new);
      }
      assertEquals(1, childInodes.length);
      assertEquals(file, mTree.getPath(childInodes[0].getId()));
    }

    // Test multiple files with the prefix
    prefix = "bfile";
    List<AlluxioURI> files = new ArrayList<>(Arrays.asList(new AlluxioURI("/" + prefix),
        new AlluxioURI("/" + prefix + "1"),
        new AlluxioURI("/" + prefix + "2")));
    // add the files out of order
    for (int i = files.size() - 1; i >= 0; i--) {
      createPath(mTree, files.get(i), sNestedFileContext);
    }
    // add a nested directory that matches the prefix
    String nestedPath = "/" + prefix + "3";
    createPath(mTree, new AlluxioURI(nestedPath + "/nested"), sNestedDirectoryContext);
    files.add(new AlluxioURI(nestedPath));
    // add a file that does not match the prefix
    createPath(mTree, new AlluxioURI("/cfile"), sNestedFileContext);
    try (LockedInodePath inodePath = mTree.lockFullInodePath(
        file, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      long parentId = inodePath.getInode().getParentId();
      Inode[] childInodes;
      // read from "bfile1" with prefix
      try (CloseableIterator<? extends Inode> fromIter = mInodeStore.getChildrenPrefixFrom(parentId,
          prefix, prefix + "1")) {
        childInodes = iteratorToStream(fromIter).toArray(Inode[]::new);
      }
      assertEquals(files.size() - 1, childInodes.length);
      for (int i = 1; i < files.size(); i++) {
        assertEquals(files.get(i), mTree.getPath(childInodes[i - 1].getId()));
      }
    }
  }

  @Deprecated
  public void getChildrenAfterId() throws Exception {
    int fileCount = 10;
    String parent = "/nxt";
    List<AlluxioURI> files = setupBaseFiles(fileCount, parent);
    checkChildren(files);
  }

  <T> Stream<T> iteratorToStream(Iterator<T> iter) {
    return StreamSupport.stream(Spliterators.spliteratorUnknownSize(
        iter, Spliterator.ORDERED), false);
  }

  @Deprecated
  public void getChildAfterIdDeleted() throws Exception {
    int fileCount = 10;
    String parent = "/nxt";
    List<AlluxioURI> files = setupBaseFiles(fileCount, parent);

    // delete the file in the middle
    int middle = fileCount / 2;
    AlluxioURI toRemove = files.remove(middle);
    deleteInodeByPath(mTree, toRemove);
    checkChildren(files);
  }

  @Deprecated
  public void getChildrenAfterIdNested() throws Exception {
    int fileCount = 10;
    int nestDepth = 10;
    String parent = "";
    List<AlluxioURI> files = new ArrayList<>();
    for (int i = 0; i < nestDepth; i++) {
      for (int j = 0; j < fileCount; j++) {
        files.add(new AlluxioURI(String.format("%s/%d", parent, j)));
      }
      parent += "/nxt";
    }

    for (AlluxioURI nxt : files) {
      createPath(mTree, nxt, sNestedFileContext);
    }

    // Go through each file and be sure there are the proper inodes following it in the listing
    parent = "";
    for (int i = 0; i < nestDepth; i++) {
      for (int j = 0; j < fileCount; j++) {
        // fileCount plus 1 for remaining nested directory (except on the bottom depth)
        int remain = fileCount - j;
        boolean notAtBottom = i < nestDepth - 1;
        if (notAtBottom) {
          remain++;
        }
        AlluxioURI nxt = new AlluxioURI(String.format("%s/%d", parent, j));
        try (LockedInodePath inodePath = mTree.lockFullInodePath(
            nxt, LockPattern.READ, NoopJournalContext.INSTANCE)) {
          long parentId = inodePath.getInode().getParentId();
          Inode childInode = inodePath.getInode();
          Inode[] childInodes;
          try (CloseableIterator<? extends Inode> fromIter = mInodeStore.getChildrenFrom(parentId,
              childInode.getName())) {
            childInodes = iteratorToStream(fromIter).toArray(Inode[]::new);
          }
          for (int k = 0; k < remain; k++) {
            if (notAtBottom && k == remain - 1) {
              // the last entry should be the remaining directory
              assertEquals(new AlluxioURI(String.format("%s/nxt", parent)),
                  mTree.getPath(childInodes[k].getId()));
            } else {
              assertEquals(new AlluxioURI(String.format("%s/%d", parent, k + j)),
                  mTree.getPath(childInodes[k].getId()));
            }
          }
          assertEquals(remain, childInodes.length);
        }
      }
      parent += "/nxt";
    }
  }

  @Deprecated
  public void getPathById() throws Exception {
    // test nesting
    long id;
    createPath(mTree, NESTED_URI, sNestedDirectoryContext);
    try (LockedInodePath inodePath = mTree.lockFullInodePath(
        NESTED_URI, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      assertEquals(NESTED_URI, mTree.getPath(inodePath.getInode()));
      id = inodePath.getInode().getId();
    }
    assertEquals(NESTED_URI, mTree.getPath(id));

    // test path not exists
    assertThrows(FileDoesNotExistException.class, () -> mTree.getPath(id + 1));
  }

  @Deprecated
  public void getInodesById() throws Exception {
    // test nesting
    long id;
    ArrayList<String> pathIds;
    createPath(mTree, NESTED_URI, sNestedDirectoryContext);
    try (LockedInodePath inodePath = mTree.lockFullInodePath(
        NESTED_URI, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      pathIds = inodePath.getInodeList().stream().map(Inode::getName).collect(
          Collectors.toCollection(ArrayList::new));
      assertEquals(pathIds, mTree.getPathInodeNames(inodePath.getInode()));
      id = inodePath.getInode().getId();
    }
    assertEquals(pathIds, mTree.getPathInodeNames(id));

    // test path not exists
    assertThrows(FileDoesNotExistException.class, () -> mTree.getPathInodeNames(id + 1));
  }

  @Deprecated
  public void getInodeChildrenRecursive() throws Exception {
    createPath(mTree, TEST_URI, sDirectoryContext);
    createPath(mTree, NESTED_URI, sNestedDirectoryContext);
    // add nested file
    createPath(mTree, NESTED_FILE_URI, sNestedFileContext);

    // all inodes under root
    try (LockedInodePath inodePath =
             mTree.lockFullInodePath(0, LockPattern.WRITE_INODE, NoopJournalContext.INSTANCE)) {
      // /test, /nested, /nested/test, /nested/test/file
      assertEquals(4, mTree.getDescendants(inodePath).getInodePathList().size());
    }
  }

  /**
   * Tests deleting a nested inode.
   * @deprecated
   */
  @Deprecated
  public void deleteInode() throws Exception {
    createPath(mTree, NESTED_URI, sNestedDirectoryContext);

    // all inodes under root
    try (LockedInodePath inodePath =
             mTree.lockFullInodePath(0, LockPattern.WRITE_INODE, NoopJournalContext.INSTANCE)) {
      // /nested, /nested/test
      assertEquals(2, mTree.getDescendants(inodePath).getInodePathList().size());

      // delete the nested inode
      deleteInodeByPath(mTree, NESTED_URI);

      // only /nested left
      assertEquals(1, mTree.getDescendants(inodePath).getInodePathList().size());
    }
  }

  @Deprecated
  public void setPinned() throws Exception {
    createPath(mTree, NESTED_URI, sNestedDirectoryContext);
    createPath(mTree, NESTED_FILE_URI, sNestedFileContext);

    // no inodes pinned
    assertEquals(0, mTree.getPinIdSet().size());

    // pin nested folder
    try (
        LockedInodePath inodePath = mTree.lockFullInodePath(
            NESTED_URI, LockPattern.WRITE_INODE, NoopJournalContext.INSTANCE)
    ) {
      mTree.setPinned(RpcContext.NOOP, inodePath, true, Collections.emptyList(), 0);
    }
    // nested file pinned
    assertEquals(1, mTree.getPinIdSet().size());

    // unpin nested folder
    try (
        LockedInodePath inodePath = mTree.lockFullInodePath(
            NESTED_URI, LockPattern.WRITE_INODE, NoopJournalContext.INSTANCE)
    ) {
      mTree.setPinned(RpcContext.NOOP, inodePath, false, Collections.emptyList(), 0);
    }
    assertEquals(0, mTree.getPinIdSet().size());
  }

  /**
   * Tests that streaming to a journal checkpoint works.
   * @deprecated
   */
  @Deprecated
  public void streamToJournalCheckpoint() throws Exception {
    verifyJournal(mTree, Arrays.asList(getInodeByPath("/")));

    // test nested URI
    createPath(mTree, NESTED_FILE_URI, sNestedFileContext);

    verifyJournal(mTree, StreamUtils.map(path -> getInodeByPath(path),
        Arrays.asList("/", "/nested", "/nested/test", "/nested/test/file")));

    // add a sibling of test and verify journaling is in correct order (breadth first)
    createPath(mTree, new AlluxioURI("/nested/test1/file1"), sNestedFileContext);
    verifyJournal(mTree, StreamUtils.map(path -> getInodeByPath(path), Arrays.asList("/",
        "/nested", "/nested/test", "/nested/test1", "/nested/test/file", "/nested/test1/file1")));
  }

  @Deprecated
  public void addInodeFromJournal() throws Exception {
    createPath(mTree, NESTED_FILE_URI, sNestedFileContext);
    createPath(mTree, new AlluxioURI("/nested/test1/file1"), sNestedFileContext);
    MutableInode<?> root = getInodeByPath("/");
    MutableInode<?> nested = getInodeByPath("/nested");
    MutableInode<?> test = getInodeByPath("/nested/test");
    MutableInode<?> file = getInodeByPath("/nested/test/file");
    MutableInode<?> test1 = getInodeByPath("/nested/test1");
    MutableInode<?> file1 = getInodeByPath("/nested/test1/file1");
    // reset the tree
    mTree.processJournalEntry(root.toJournalEntry());
    // re-init the root since the tree was reset above
    mTree.getRoot();
    try (LockedInodePath inodePath =
             mTree.lockFullInodePath(
                 new AlluxioURI("/"), LockPattern.WRITE_INODE, NoopJournalContext.INSTANCE)
    ) {
      assertEquals(0, mTree.getDescendants(inodePath).getInodePathList().size());
      mTree.processJournalEntry(nested.toJournalEntry());
      verifyChildrenNames(mTree, inodePath, Sets.newHashSet("nested"));
      mTree.processJournalEntry(test.toJournalEntry());
      verifyChildrenNames(mTree, inodePath, Sets.newHashSet("nested", "test"));
      mTree.processJournalEntry(test1.toJournalEntry());
      verifyChildrenNames(mTree, inodePath, Sets.newHashSet("nested", "test", "test1"));
      mTree.processJournalEntry(file.toJournalEntry());
      verifyChildrenNames(mTree, inodePath, Sets.newHashSet("nested", "test", "test1", "file"));
      mTree.processJournalEntry(file1.toJournalEntry());
      verifyChildrenNames(mTree, inodePath,
          Sets.newHashSet("nested", "test", "test1", "file", "file1"));
    }
  }

  @Deprecated
  public void addInodeModeFromJournalWithEmptyOwnership() throws Exception {
    createPath(mTree, NESTED_FILE_URI, sNestedFileContext);
    MutableInode<?> root = getInodeByPath("/");
    MutableInode<?> nested = getInodeByPath("/nested");
    MutableInode<?> test = getInodeByPath("/nested/test");
    MutableInode<?> file = getInodeByPath("/nested/test/file");
    MutableInode<?>[] inodeChildren = {nested, test, file};
    for (MutableInode<?> child : inodeChildren) {
      child.setOwner("");
      child.setGroup("");
      child.setMode((short) 0600);
    }
    // reset the tree
    mTree.processJournalEntry(root.toJournalEntry());
    // re-init the root since the tree was reset above
    mTree.getRoot();
    try (LockedInodePath inodePath =
             mTree.lockFullInodePath(
                 new AlluxioURI("/"), LockPattern.WRITE_INODE, NoopJournalContext.INSTANCE)
    ) {
      assertEquals(0, mTree.getDescendants(inodePath).getInodePathList().size());
      mTree.processJournalEntry(nested.toJournalEntry());
      mTree.processJournalEntry(test.toJournalEntry());
      mTree.processJournalEntry(file.toJournalEntry());
      List<LockedInodePath> descendants = mTree.getDescendants(inodePath).getInodePathList();
      assertEquals(inodeChildren.length, descendants.size());
      for (LockedInodePath childPath : descendants) {
        Inode child = childPath.getInodeOrNull();
        Assert.assertNotNull(child);
        Assert.assertEquals("", child.getOwner());
        Assert.assertEquals("", child.getGroup());
        Assert.assertEquals((short) 0600, child.getMode());
      }
    }
  }

  @Deprecated
  public void getInodePathById() throws Exception {
    try (LockedInodePath rootPath
             = mTree.lockFullInodePath(0, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      assertEquals(0, rootPath.getInode().getId());
    }

    List<Inode> created = createPath(mTree, NESTED_FILE_URI, sNestedFileContext);

    for (Inode inode : created) {
      long id = inode.getId();
      try (LockedInodePath inodePath
               = mTree.lockFullInodePath(id, LockPattern.READ, NoopJournalContext.INSTANCE)) {
        assertEquals(id, inodePath.getInode().getId());
      }
    }
  }

  @Deprecated
  public void getInodePathByPath() throws Exception {
    try (LockedInodePath rootPath =
        mTree.lockFullInodePath(
            new AlluxioURI("/"), LockPattern.READ, NoopJournalContext.INSTANCE)
    ) {
      assertTrue(mTree.isRootId(rootPath.getInode().getId()));
    }

    // Create a nested file.
    createPath(mTree, NESTED_FILE_URI, sNestedFileContext);

    AlluxioURI uri = new AlluxioURI("/nested");
    try (LockedInodePath inodePath =
             mTree.lockFullInodePath(uri, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      assertEquals(uri.getName(), inodePath.getInode().getName());
    }

    uri = NESTED_URI;
    try (LockedInodePath inodePath =
             mTree.lockFullInodePath(uri, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      assertEquals(uri.getName(), inodePath.getInode().getName());
    }

    uri = NESTED_FILE_URI;
    try (LockedInodePath inodePath =
             mTree.lockFullInodePath(uri, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      assertEquals(uri.getName(), inodePath.getInode().getName());
    }
  }

  // Helper to create a path.
  private List<Inode> createPath(InodeTree root, AlluxioURI path, CreatePathContext<?, ?> context)
      throws FileAlreadyExistsException, BlockInfoException, InvalidPathException, IOException,
      FileDoesNotExistException {
    try (LockedInodePath inodePath =
             root.lockInodePath(path, LockPattern.WRITE_EDGE, NoopJournalContext.INSTANCE)) {
      return root.createPath(RpcContext.NOOP, inodePath, context);
    }
  }

  // Helper to get an inode by path. The inode is unlocked before returning.
  private MutableInode<?> getInodeByPath(String path) {
    try {
      return getInodeByPath(new AlluxioURI(path));
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  // Helper to get an inode by path. The inode is unlocked before returning.
  private MutableInode<?> getInodeByPath(AlluxioURI path) throws Exception {
    try (LockedInodePath inodePath =
             mTree.lockFullInodePath(path, LockPattern.READ, NoopJournalContext.INSTANCE)) {
      return mInodeStore.getMutable(inodePath.getInode().getId()).get();
    }
  }

  // Helper to delete an inode by path.
  private static void deleteInodeByPath(InodeTree root, AlluxioURI path) throws Exception {
    try (LockedInodePath inodePath
             = root.lockFullInodePath(path, LockPattern.WRITE_EDGE, NoopJournalContext.INSTANCE)) {
      root.deleteInode(RpcContext.NOOP, inodePath, System.currentTimeMillis());
    }
  }

  // helper for verifying that correct objects were journaled to the output stream
  private static void verifyJournal(InodeTree root, List<MutableInode<?>> journaled) {
    try (CloseableIterator<Journal.JournalEntry> it = root.getJournalEntryIterator()) {
      // Read entries from InodeTree.
      List<Journal.JournalEntry> treeEntries = new LinkedList<>();
      while (it.get().hasNext()) {
        treeEntries.add(it.get().next());
      }

      // Validate InodeTree entries match given entries.
      for (MutableInode<?> node : journaled) {
        assertTrue(treeEntries.contains(node.toJournalEntry()));
      }
    }
  }

  // verify that the tree has the given children
  private static void verifyChildrenNames(InodeTree tree, LockedInodePath inodePath,
      Set<String> childNames) throws Exception {
    List<LockedInodePath> descendants = tree.getDescendants(inodePath).getInodePathList();
    assertEquals(childNames.size(), descendants.size());
    for (LockedInodePath childPath : descendants) {
      assertTrue(childNames.contains(childPath.getInode().getName()));
    }
  }
}
