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

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;

import alluxio.AlluxioURI;
import alluxio.exception.FileDoesNotExistException;
import alluxio.exception.InvalidPathException;
import alluxio.grpc.DeletePOptions;
import alluxio.grpc.ListStatusPOptions;
import alluxio.grpc.ListStatusPartialPOptions;
import alluxio.grpc.LoadMetadataPType;
import alluxio.master.file.contexts.DeleteContext;
import alluxio.master.file.contexts.ListStatusContext;
import alluxio.master.file.contexts.RenameContext;
import alluxio.master.metastore.InodeStore;
import alluxio.wire.FileInfo;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

@RunWith(Parameterized.class)
public class FileSystemMasterPartialListingTest extends FileSystemMasterTestBase {

  public FileSystemMasterPartialListingTest(InodeStore.Factory factory) {
    mInodeStoreFactory = factory;
  }

  private ListStatusContext genListStatusPrefix(String prefix, boolean recursive) {
    return ListStatusContext.mergeFrom(
        ListStatusPartialPOptions.newBuilder().setPrefix(prefix).setOptions(
            ListStatusPOptions.newBuilder()
                .setLoadMetadataType(LoadMetadataPType.NEVER)
                .setRecursive(recursive)));
  }

  private ListStatusContext genListStatusStartAfter(String startAfter, boolean recursive) {
    return ListStatusContext.mergeFrom(
        ListStatusPartialPOptions.newBuilder().setStartAfter(startAfter).setOptions(
            ListStatusPOptions.newBuilder()
                .setLoadMetadataType(LoadMetadataPType.NEVER)
                .setRecursive(recursive)));
  }

  private ListStatusContext genListStatusPrefixStartAfter(
      String prefix, String startAfter, boolean recursive) {
    return ListStatusContext.mergeFrom(
        ListStatusPartialPOptions.newBuilder().setPrefix(prefix)
            .setStartAfter(startAfter).setOptions(ListStatusPOptions.newBuilder()
                .setLoadMetadataType(LoadMetadataPType.NEVER)
                .setRecursive(recursive)));
  }

  private ListStatusContext genListStatusPrefixOffsetCount(
      int batchSize, int offsetCount, boolean recursive, String prefix) {
    return ListStatusContext.mergeFrom(
        ListStatusPartialPOptions.newBuilder().setPrefix(prefix).setBatchSize(batchSize)
            .setOffsetCount(offsetCount).setOptions(ListStatusPOptions.newBuilder()
                .setLoadMetadataType(LoadMetadataPType.NEVER)
                .setRecursive(recursive)));
  }

  private ListStatusContext genListStatusPartial(
      int batchSize, long offsetId, boolean recursive, String prefix, String startAfter) {
    assertTrue("only one of offsetId and startAfter can be set",
        offsetId == 0 || startAfter.isEmpty());
    ListStatusPartialPOptions.Builder builder = ListStatusPartialPOptions.newBuilder()
        .setBatchSize(batchSize)
        .setPrefix(prefix).setOptions(
            ListStatusPOptions.newBuilder()
                .setLoadMetadataType(LoadMetadataPType.NEVER)
                .setRecursive(recursive));
    if (offsetId != 0) {
      builder.setOffsetId(offsetId);
    } else {
      builder.setStartAfter(startAfter);
    }
    return ListStatusContext.mergeFrom(builder);
  }

  @Test
  public void listStatusPartialDelete() throws Exception {
    long offset = 0;
    List<FileInfo> infos;

    createFileWithSingleBlock(ROOT_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE2_URI);
    createFileWithSingleBlock(NESTED_DIR_URI);

    // List two file from NESTED_URI, getting its offset
    ListStatusContext context = genListStatusPartial(2, offset, true, "", "");
    infos = mFileSystemMaster.listStatus(NESTED_URI, context);
    assertEquals(2, infos.size());
    assertEquals(NESTED_DIR_URI.toString(), infos.get(0).getPath());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(1).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(1).getFileId();

    // The next file should be NESTED_FILE_URI2
    context = genListStatusPartial(2, offset, true, "", "");
    infos = mFileSystemMaster.listStatus(NESTED_URI, context);
    assertEquals(1, infos.size());
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(0).getPath());

    // Now delete NESTED_FILE_URI, so the offset is no longer valid
    mFileSystemMaster.delete(NESTED_FILE_URI,
        DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)));

    // Should throw an inode does not exist exception
    final long throwOffset = offset;
    assertThrows(FileDoesNotExistException.class, () -> mFileSystemMaster.listStatus(
        NESTED_URI, genListStatusPartial(1, throwOffset, true, "", "")));

    // Insert a new file with the same name
    createFileWithSingleBlock(NESTED_FILE_URI);

    // An exception should still be thrown because it has a new inode id
    assertThrows(FileDoesNotExistException.class, () -> mFileSystemMaster.listStatus(
        NESTED_URI, genListStatusPartial(1, throwOffset, true, "", "")));
  }

  @Test
  public void listStatusPartialRename() throws Exception {
    long offset = 0;
    List<FileInfo> infos;

    createFileWithSingleBlock(ROOT_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE2_URI);
    createFileWithSingleBlock(NESTED_DIR_URI);

    ListStatusContext context = genListStatusPartial(2, offset, true, "", "");
    // List two file from NESTED_URI, getting its offset
    infos = mFileSystemMaster.listStatus(NESTED_URI, context);
    assertEquals(2, infos.size());
    assertEquals(NESTED_DIR_URI.toString(), infos.get(0).getPath());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(1).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(1).getFileId();

    // The next file should still be NESTED_FILE_URI2
    context = genListStatusPartial(2, offset, true, "", "");
    infos = mFileSystemMaster.listStatus(NESTED_URI, context);
    assertEquals(1, infos.size());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(0).getPath());

    AlluxioURI renameTo = new AlluxioURI(NESTED_FILE_URI + "1");
    // Now rename NESTED_FILE_URI, but in the same directory
    mFileSystemMaster.rename(NESTED_FILE_URI, renameTo,
        RenameContext.defaults());

    // The next file should still be NESTED_FILE_URI2
    context = genListStatusPartial(2, offset, true, "", "");
    infos = mFileSystemMaster.listStatus(NESTED_URI, context);
    assertEquals(1, infos.size());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(0).getPath());

    // Now rename the renamed file into a different directory
    mFileSystemMaster.rename(renameTo, new AlluxioURI("/moveHere"),
        RenameContext.defaults());
    // Should throw an exception since we are no longer in the same path as NESTED_URI
    final long finalOffset = offset;
    assertThrows(FileDoesNotExistException.class,
        () -> mFileSystemMaster.listStatus(NESTED_URI, genListStatusPartial(
            1, finalOffset, true, "", "")));
  }

  @Test
  public void listStatusStartAfter() throws Exception {
    List<FileInfo> infos;

    createFileWithSingleBlock(ROOT_FILE_URI);
    createFileWithSingleBlock(ROOT_AFILE_URI);
    createFileWithSingleBlock(NESTED_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE2_URI);
    createFileWithSingleBlock(NESTED_DIR_URI);
    createFileWithSingleBlock(NESTED_TEST_FILE_URI);

    // list without recursion, and start after "/file",
    // the results should be sorted by name
    ListStatusContext context = genListStatusStartAfter(
        ROOT_FILE_URI.getPath(), false);
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(3, context.getTotalListings());
    assertFalse(context.isTruncated());
    assertEquals(NESTED_BASE_URI.toString(), infos.get(0).getPath());

    // list with recursion, and start after "/file",
    // the results should be sorted by name
    context = genListStatusStartAfter(
        ROOT_FILE_URI.getPath(), true);
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(6, infos.size());
    assertEquals(NESTED_BASE_URI.toString(), infos.get(0).getPath());
    assertEquals(NESTED_URI.toString(), infos.get(1).getPath());
    assertEquals(NESTED_DIR_URI.toString(), infos.get(2).getPath());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(3).getPath());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(4).getPath());
    assertEquals(NESTED_TEST_FILE_URI.toString(), infos.get(5).getPath());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());

    // list "/nested/test" with recursion, and start after "/file",
    // the results should be sorted by name
    context = genListStatusStartAfter("file", true);
    infos = mFileSystemMaster.listStatus(NESTED_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(0).getPath());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());

    // list "/nested/test" with recursion, and start after "/dir",
    // the results should be sorted by name
    context = genListStatusStartAfter("di", true);
    infos = mFileSystemMaster.listStatus(NESTED_URI, context);
    assertEquals(3, infos.size());
    assertEquals(NESTED_DIR_URI.toString(), infos.get(0).getPath());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(1).getPath());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(2).getPath());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());

    // list "/nested/test" with recursion, and start after "/dir",
    // the results should be sorted by name
    context = genListStatusStartAfter("dir", true);
    infos = mFileSystemMaster.listStatus(NESTED_URI, context);
    assertEquals(2, infos.size());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(0).getPath());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(1).getPath());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());
  }

  @Test
  public void listStatusPrefixNestedStartAfter() throws Exception {
    List<FileInfo> infos;

    createFileWithSingleBlock(ROOT_FILE_URI);
    createFileWithSingleBlock(TEST_URI);
    createFileWithSingleBlock(ROOT_AFILE_URI);
    createFileWithSingleBlock(NESTED_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE2_URI);
    createFileWithSingleBlock(NESTED_DIR_URI);
    createFileWithSingleBlock(NESTED_TEST_FILE_URI);

    // list without recursion, with prefix "/file" and startAfter "/fi"
    ListStatusContext context = genListStatusPrefixStartAfter(
        ROOT_FILE_URI.getPath(), "/fi", false);
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(ROOT_FILE_URI.toString(), infos.get(0).getPath());
    assertEquals(4, context.getTotalListings());
    assertFalse(context.isTruncated());

    // list without recursion, with prefix "/file" and startAfter "/file"
    context = genListStatusPrefixStartAfter(
        ROOT_FILE_URI.getPath(), ROOT_FILE_URI.getPath(), false);
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(0, infos.size());
    assertEquals(4, context.getTotalListings());
    assertFalse(context.isTruncated());

    // list with recursion, with prefix "/fi" and startAfter "/file"
    context = genListStatusPrefixStartAfter(
        "/fi", ROOT_FILE_URI.getPath(), false);
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(0, infos.size());
    assertEquals(4, context.getTotalListings());
    assertFalse(context.isTruncated());

    // list with recursion, with prefix "/fi" and startAfter "/file"
    context = genListStatusPrefixStartAfter(
        "/fi", ROOT_FILE_URI.getPath(), true);
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(0, infos.size());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());

    // list with recursion, with prefix "/ne" and startAfter "/file"
    context = genListStatusPrefixStartAfter(
        "/ne", ROOT_FILE_URI.getPath(), true);
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(6, infos.size());
    assertEquals(NESTED_BASE_URI.toString(), infos.get(0).getPath());
    assertEquals(NESTED_URI.toString(), infos.get(1).getPath());
    assertEquals(NESTED_DIR_URI.toString(), infos.get(2).getPath());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(3).getPath());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(4).getPath());
    assertEquals(NESTED_TEST_FILE_URI.toString(), infos.get(5).getPath());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());

    // list with recursion, with prefix "/ne" and startAfter "/nested/test"
    context = genListStatusPrefixStartAfter(
        "/ne", "/nested/test", true);
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(4, infos.size());
    assertEquals(NESTED_DIR_URI.toString(), infos.get(0).getPath());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(1).getPath());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(2).getPath());
    assertEquals(NESTED_TEST_FILE_URI.toString(), infos.get(3).getPath());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());
  }

  @Test
  public void listStatusPartialPrefixNestedStartAfter() throws Exception {
    List<FileInfo> infos;
    long offset;

    createFileWithSingleBlock(ROOT_FILE_URI);
    createFileWithSingleBlock(TEST_URI);
    createFileWithSingleBlock(ROOT_AFILE_URI);
    createFileWithSingleBlock(NESTED_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE2_URI);
    createFileWithSingleBlock(NESTED_DIR_URI);
    createFileWithSingleBlock(NESTED_TEST_FILE_URI);

    // list without recursion with prefix "/file" and start after "/fi",
    // the results should be sorted by name
    ListStatusContext context = genListStatusPartial(
        1, 0, false, ROOT_FILE_URI.getPath(), "/fi");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(ROOT_FILE_URI.toString(), infos.get(0).getPath());
    assertEquals(4, context.getTotalListings());
    assertFalse(context.isTruncated());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, false, ROOT_FILE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(4, context.getTotalListings());
    assertFalse(context.isTruncated());
    assertEquals(0, infos.size());

    // list with recursion from "/nested/test/" with prefix "/file" and start after "/fi",
    // the results should be sorted by name
    context = genListStatusPartial(
        1, 0, true, ROOT_FILE_URI.getPath(), "fi");
    infos = mFileSystemMaster.listStatus(NESTED_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(0).getPath());
    assertEquals(-1, context.getTotalListings());
    assertTrue(context.isTruncated());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, ROOT_FILE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(NESTED_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(0).getPath());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, ROOT_FILE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(NESTED_URI, context);
    assertEquals(0, infos.size());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());

    // list with recursion from "/" with prefix "/nest" and start after "/nested/d",
    // the results should be sorted by name
    context = genListStatusPartial(
        1, 0, true, NESTED_BASE_URI.getPath(), "nested/test/d");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_DIR_URI.toString(), infos.get(0).getPath());
    assertEquals(-1, context.getTotalListings());
    assertTrue(context.isTruncated());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_TEST_FILE_URI.toString(), infos.get(0).getPath());
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    assertEquals(0, infos.size());
  }

  @Test
  public void listStatusPrefixNested() throws Exception {
    List<FileInfo> infos;

    createFileWithSingleBlock(ROOT_FILE_URI);
    createFileWithSingleBlock(ROOT_AFILE_URI);
    createFileWithSingleBlock(NESTED_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE2_URI);
    createFileWithSingleBlock(NESTED_DIR_URI);

    // list one at a time without recursion, and prefix "/file",
    // the results should be sorted by name
    ListStatusContext context = genListStatusPrefix(
        ROOT_FILE_URI.getPath(), false);
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(ROOT_FILE_URI.toString(), infos.get(0).getPath());
    assertFalse(context.isTruncated());
    assertEquals(3, context.getTotalListings());

    // list without recursion, with a prefix that is longer than the result
    context = genListStatusPrefix(
        NESTED_FILE_URI.getPath(), true);
    infos = mFileSystemMaster.listStatus(NESTED_BASE_URI, context);
    assertEquals(0, infos.size());
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());

    // list one at a time without recursion, and prefix "/nested",
    // the results should be sorted by name
    context = genListStatusPrefix(
        NESTED_BASE_URI.getPath(), false);
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_BASE_URI.toString(), infos.get(0).getPath());
    assertFalse(context.isTruncated());
    assertEquals(3, context.getTotalListings());

    // start listing without recursion from "/nested", with a prefix of "/test/file"
    context = genListStatusPrefix("/test/file", false);
    infos = mFileSystemMaster.listStatus(NESTED_BASE_URI, context);
    assertEquals(0, infos.size());
    assertFalse(context.isTruncated());
    assertEquals(1, context.getTotalListings());

    // list one at a time with recursion, and prefix "/nested",
    // the results should be sorted by name
    context = genListStatusPrefix(
        NESTED_BASE_URI.getPath(), true);
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(5, infos.size());
    assertEquals(NESTED_BASE_URI.toString(), infos.get(0).getPath());
    assertEquals(NESTED_URI.toString(), infos.get(1).getPath());
    assertEquals(NESTED_DIR_URI.toString(), infos.get(2).getPath());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(3).getPath());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(4).getPath());
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());

    // start listing with recursion from "/nested", with a prefix of "/test/file"
    context = genListStatusPrefix("/test/file", true);
    infos = mFileSystemMaster.listStatus(NESTED_BASE_URI, context);
    assertEquals(2, infos.size());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(0).getPath());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(1).getPath());
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
  }

  @Test
  public void listStatusPartialPrefixNested() throws Exception {
    long offset;
    List<FileInfo> infos;

    createFileWithSingleBlock(ROOT_FILE_URI);
    createFileWithSingleBlock(ROOT_AFILE_URI);
    createFileWithSingleBlock(NESTED_FILE_URI);
    createFileWithSingleBlock(NESTED_TEST_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE2_URI);
    createFileWithSingleBlock(NESTED_DIR_URI);

    // list one at a time without recursion, and prefix "/file",
    // the results should be sorted by name
    ListStatusContext context = genListStatusPartial(
        1, 0, false, ROOT_FILE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(ROOT_FILE_URI.toString(), infos.get(0).getPath());
    assertFalse(context.isTruncated());
    assertEquals(3, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, false, ROOT_FILE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(3, context.getTotalListings());
    assertEquals(0, infos.size());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 1, false, ROOT_FILE_URI.getPath());
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(3, context.getTotalListings());
    assertEquals(0, infos.size());

    // list one at a time without recursion, and prefix "/nested",
    // the results should be sorted by name
    context = genListStatusPartial(
        1, 0, false, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_BASE_URI.toString(), infos.get(0).getPath());
    assertEquals(3, context.getTotalListings());
    assertFalse(context.isTruncated());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, false, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(3, context.getTotalListings());
    assertEquals(0, infos.size());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 1, false, NESTED_BASE_URI.getPath());
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(3, context.getTotalListings());
    assertEquals(0, infos.size());

    // list one at a time without recursion from "/nested", and prefix "/test",
    // the results should be sorted by name
    context = genListStatusPartial(
        1, 0, false, "/test", "");
    infos = mFileSystemMaster.listStatus(NESTED_BASE_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_URI.toString(), infos.get(0).getPath());
    assertEquals(2, context.getTotalListings());
    assertTrue(context.isTruncated());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, false, "/test", "");
    infos = mFileSystemMaster.listStatus(NESTED_BASE_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_TEST_FILE_URI.toString(), infos.get(0).getPath());
    assertFalse(context.isTruncated());
    assertEquals(2, context.getTotalListings());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 1, false, "/test");
    infos = mFileSystemMaster.listStatus(NESTED_BASE_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_TEST_FILE_URI.toString(), infos.get(0).getPath());
    assertFalse(context.isTruncated());
    assertEquals(2, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, false, "/test", "");
    infos = mFileSystemMaster.listStatus(NESTED_BASE_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(2, context.getTotalListings());
    assertEquals(0, infos.size());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 2, false, "/test");
    infos = mFileSystemMaster.listStatus(NESTED_BASE_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(2, context.getTotalListings());
    assertEquals(0, infos.size());

    // list one at a time with recursion with prefix "/file",
    // the results should be sorted by name,
    // and returned in a depth first manner
    context = genListStatusPartial(
        1, 0, true, ROOT_FILE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(ROOT_FILE_URI.toString(), infos.get(0).getPath());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());
    offset = infos.get(0).getFileId();

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 1, true, ROOT_FILE_URI.getPath());
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    assertEquals(0, infos.size());

    context = genListStatusPartial(
        1, offset, true, ROOT_FILE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    assertEquals(0, infos.size());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 2, true, ROOT_FILE_URI.getPath());
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    assertEquals(0, infos.size());

    // list one at a time with recursion with prefix "/nested",
    // the results should be sorted by name,
    context = genListStatusPartial(
        1, 0, true, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_BASE_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 1, true, NESTED_BASE_URI.getPath());
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    // give an invalid nested prefix during the partial listing
    long finalOffset = offset;
    assertThrows(InvalidPathException.class, () ->
        mFileSystemMaster.listStatus(ROOT_URI, genListStatusPartial(
            1, finalOffset, true, ROOT_FILE_URI.getPath(), "")));

    context = genListStatusPartial(
        1, offset, true, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_DIR_URI.toString(), infos.get(0).getPath());
    assertEquals(-1, context.getTotalListings());
    assertTrue(context.isTruncated());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 2, true, NESTED_BASE_URI.getPath());
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_DIR_URI.toString(), infos.get(0).getPath());
    assertEquals(-1, context.getTotalListings());
    assertTrue(context.isTruncated());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 3, true, NESTED_BASE_URI.getPath());
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(0).getPath());
    assertEquals(-1, context.getTotalListings());
    assertTrue(context.isTruncated());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 4, true, NESTED_BASE_URI.getPath());
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(0).getPath());
    assertEquals(-1, context.getTotalListings());
    assertTrue(context.isTruncated());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_TEST_FILE_URI.toString(), infos.get(0).getPath());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 5, true, NESTED_BASE_URI.getPath());
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_TEST_FILE_URI.toString(), infos.get(0).getPath());
    assertEquals(-1, context.getTotalListings());
    assertFalse(context.isTruncated());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, NESTED_BASE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    assertEquals(0, infos.size());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 6, true, NESTED_BASE_URI.getPath());
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    assertEquals(0, infos.size());

    // list one at a time with recursion with prefix "/nested/test/file",
    // the results should be sorted by name,
    // and returned in a depth first manner
    context = genListStatusPartial(
        1, 0, true, NESTED_FILE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, NESTED_FILE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(0).getPath());
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 1, true, NESTED_FILE_URI.getPath());
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(0).getPath());
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, NESTED_FILE_URI.getPath(), "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    assertEquals(0, infos.size());

    // use an offset count instead of offset id, the results should be the same
    context = genListStatusPrefixOffsetCount(
        1, 2, true, NESTED_FILE_URI.getPath());
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    assertEquals(0, infos.size());
  }

  @Test
  public void listStatusPartialNested() throws Exception {
    long offset;
    List<FileInfo> infos;

    createFileWithSingleBlock(ROOT_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE_URI);
    createFileWithSingleBlock(NESTED_TEST_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE2_URI);
    createFileWithSingleBlock(NESTED_DIR_URI);

    // list one at a time without recursion, the results should be sorted by name
    ListStatusContext context = genListStatusPartial(
        1, 0, false, "", "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(ROOT_FILE_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(2, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, false, "", "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_BASE_URI.toString(), infos.get(0).getPath());
    assertFalse(context.isTruncated());
    assertEquals(2, context.getTotalListings());
    offset = infos.get(0).getFileId();

    // use an offset count of 1 instead of offset id, should be the same as previous
    context = genListStatusPrefixOffsetCount(1, 1, false, "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_BASE_URI.toString(), infos.get(0).getPath());
    assertFalse(context.isTruncated());
    assertEquals(2, context.getTotalListings());

    context = genListStatusPartial(
        1, offset, false, "", "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(2, context.getTotalListings());
    assertEquals(0, infos.size());

    // use an offset count of 2 instead of offset id, should be the same as previous
    context = genListStatusPrefixOffsetCount(
        1, 2, false, "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(2, context.getTotalListings());
    assertEquals(0, infos.size());

    // list one at a time with recursion, the results should be sorted by name
    context = genListStatusPartial(
        1, 0, true, "", "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(ROOT_FILE_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, "", "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_BASE_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());

    // use an offset count of 1 instead of offset id, the result should be the same
    context = genListStatusPrefixOffsetCount(
        1, 1, true, "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_BASE_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, "", "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());

    // use an offset count of 2 instead of offset id, the result should be the same
    context = genListStatusPrefixOffsetCount(
        1, 2, true, "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, "", "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_DIR_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());

    // use an offset count of 3 instead of offset id, the result should be the same
    context = genListStatusPrefixOffsetCount(
        1, 3, true, "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_DIR_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, "", "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());

    // use an offset count of 4 instead of offset id, the result should be the same
    context = genListStatusPrefixOffsetCount(
        1, 4, true, "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, "", "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());

    // use an offset count of 5 instead of offset id, the result should be the same
    context = genListStatusPrefixOffsetCount(
        1, 5, true, "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_FILE2_URI.toString(), infos.get(0).getPath());
    assertTrue(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, "", "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_TEST_FILE_URI.toString(), infos.get(0).getPath());
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());

    // use an offset count of 6 instead of offset id, the result should be the same
    context = genListStatusPrefixOffsetCount(
        1, 6, true, "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertEquals(1, infos.size());
    assertEquals(NESTED_TEST_FILE_URI.toString(), infos.get(0).getPath());
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    offset = infos.get(0).getFileId();

    context = genListStatusPartial(
        1, offset, true, "", "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    assertEquals(0, infos.size());

    // use an offset count of 7 instead of offset id, the result should be the same
    context = genListStatusPrefixOffsetCount(
        1, 7, true, "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertFalse(context.isTruncated());
    assertEquals(-1, context.getTotalListings());
    assertEquals(0, infos.size());
  }

  @Test
  public void listStatusPartial() throws Exception {
    final int files = 13;
    final int batchSize = 5;
    List<FileInfo> infos;
    List<String> filenames;

    // Test files in root directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(ROOT_URI.join("file" + String.format("%05d", i)));
    }

    long offset = 0;
    for (int i = 0; i < files; i += batchSize) {
      ListStatusContext context = genListStatusPartial(batchSize, offset,
          false, "", "");
      infos = mFileSystemMaster.listStatus(ROOT_URI, context);
      assertEquals(files, context.getTotalListings());
      if (i + infos.size() == files) {
        assertFalse(context.isTruncated());
      } else {
        assertTrue(context.isTruncated());
      }
      // Copy out filenames to use List contains.
      filenames = infos.stream().map(FileInfo::getPath).collect(
          Collectors.toCollection(ArrayList::new));
      // Compare all filenames.
      for (int j = i; j < Math.min(i + batchSize, files); j++) {
        assertTrue(
            filenames.contains(ROOT_URI.join("file" + String.format("%05d", j))
                .toString()));
      }
      assertEquals(Math.min(i + batchSize, files) - i, filenames.size());
      // start from the offset
      offset = infos.get(infos.size() - 1).getFileId();
    }

    // test a non-existing file
    assertThrows(FileDoesNotExistException.class, () -> mFileSystemMaster.listStatus(
        new AlluxioURI("/doesNotExist"), genListStatusPartial(
            batchSize, 0, false, "", "")));
  }

  @Test
  public void listStatusStartAfterFromPrevious() throws Exception {
    final int files = 13;
    final int batchSize = 5;
    List<FileInfo> infos;
    List<String> filenames;

    // Test files in root directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(ROOT_URI.join("file" + String.format("%05d", i)));
    }

    String startAfter = "";
    for (int i = 0; i < files; i += batchSize) {
      ListStatusContext context = genListStatusPartial(batchSize, 0,
          false, "", startAfter);
      infos = mFileSystemMaster.listStatus(ROOT_URI, context);
      assertEquals(files, context.getTotalListings());
      if (i + infos.size() == files) {
        assertFalse(context.isTruncated());
      } else {
        assertTrue(context.isTruncated());
      }
      // Copy out filenames to use List contains.
      filenames = infos.stream().map(FileInfo::getPath).collect(
          Collectors.toCollection(ArrayList::new));
      // Compare all filenames.
      for (int j = i; j < Math.min(i + batchSize, files); j++) {
        assertTrue(
            filenames.contains(ROOT_URI.join("file" + String.format("%05d", j))
                .toString()));
      }
      assertEquals(Math.min(i + batchSize, files) - i, filenames.size());
      // start from the startAfter
      startAfter = infos.get(infos.size() - 1).getPath();
    }
  }

  @Test
  public void listStatusPartialOffsetCount() throws Exception {
    final int files = 13;
    final int batchSize = 5;
    List<FileInfo> infos;
    List<String> filenames;

    // Test files in root directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(ROOT_URI.join("file" + String.format("%05d", i)));
    }

    int offsetCount = 0;
    for (int i = 0; i < files; i += batchSize) {
      ListStatusContext context = genListStatusPrefixOffsetCount(batchSize, offsetCount,
          false, "");
      infos = mFileSystemMaster.listStatus(ROOT_URI, context);
      assertEquals(files, context.getTotalListings());
      if (i + infos.size() == files) {
        assertFalse(context.isTruncated());
      } else {
        assertTrue(context.isTruncated());
      }
      // Copy out filenames to use List contains.
      filenames = infos.stream().map(FileInfo::getPath).collect(
          Collectors.toCollection(ArrayList::new));
      // Compare all filenames.
      for (int j = i; j < Math.min(i + batchSize, files); j++) {
        assertTrue(
            filenames.contains(ROOT_URI.join("file" + String.format("%05d", j))
                .toString()));
      }
      assertEquals(Math.min(i + batchSize, files) - i, filenames.size());
      // increase the offset by the returned count
      offsetCount += infos.size();
    }

    // a larger offset than number of files should return an empty result
    ListStatusContext context = genListStatusPrefixOffsetCount(batchSize, offsetCount + 1,
        false, "");
    infos = mFileSystemMaster.listStatus(ROOT_URI, context);
    assertTrue(infos.isEmpty());
    assertEquals(files, context.getTotalListings());
  }

  @Test
  public void listStatusPartialBatchRecursive() throws Exception {
    final int files = 13;
    final int batchSize = 5;
    final int depthSize = 5;
    List<FileInfo> infos;
    // we will start a listing from each depth
    // so for each depth there will be an arraylist of files that are at that depth or greater
    ArrayList<ArrayList<String>> filenames = new ArrayList<>(depthSize);
    for (int i = 0; i < depthSize; i++) {
      filenames.add(new ArrayList<>());
    }

    // First add number files in each depth
    ArrayList<String> parent = new ArrayList<>();
    for (int j = 0; j < depthSize; j++) {
      // The directory listing will be at the start of those with smaller depth
      for (int k = 0; k < j; k++) {
        filenames.get(k).add(parent.stream().skip(1).reduce("/", String::concat) + "nxt");
      }
      for (int i = 0; i < files; i++) {
        AlluxioURI nxt = new AlluxioURI(parent.stream().reduce(
            "/", String::concat) + "file" + String.format("%05d", i));
        // the file will be in the listing of every path that has an equal or smaller depth
        for (int k = 0; k <= j; k++) {
          filenames.get(k).add(nxt.getPath());
        }
        createFileWithSingleBlock(nxt);
      }
      parent.add("nxt/");
    }

    // Start a partial listing from each depth
    StringBuilder parentPath = new StringBuilder();
    for (int i = 0; i < depthSize; i++) {
      long offset = 0;
      ArrayList<String> myFileNames = filenames.get(i);
      // do the partial listing for each batch at this depth and be sure all files are listed
      for (int j = 0; j < myFileNames.size(); j += batchSize) {
        ListStatusContext context = genListStatusPartial(
            batchSize, offset, true, "", "");
        infos = mFileSystemMaster.listStatus(new AlluxioURI("/" + parentPath), context);
        assertEquals(Math.min(myFileNames.size() - j, batchSize), infos.size());
        assertEquals(-1, context.getTotalListings());
        if (j + infos.size() == myFileNames.size()) {
          assertFalse(context.isTruncated());
        } else {
          assertTrue(context.isTruncated());
        }
        for (int k = 0; k < infos.size(); k++) {
          assertEquals(myFileNames.get(k + j), infos.get(k).getPath());
        }
        offset = infos.get(infos.size() - 1).getFileId();
      }
      // there should be no more files to list
      ListStatusContext context = genListStatusPartial(batchSize, offset, true, "", "");
      infos = mFileSystemMaster.listStatus(new AlluxioURI("/" + parentPath), context);
      assertEquals(0, infos.size());
      assertFalse(context.isTruncated());
      parentPath.append("nxt/");
    }
  }

  @Test
  public void listStatusPartialBatch() throws Exception {
    final int files = 13;
    final int batchSize = 5;
    final int depthSize = 5;
    List<FileInfo> infos;
    List<String> filenames = new ArrayList<>();

    StringBuilder parent = new StringBuilder();
    for (int j = 0; j < depthSize; j++) {
      // Test files in root directory.
      for (int i = 0; i < files; i++) {
        AlluxioURI nxt = new AlluxioURI(parent + "/file" + String.format("%05d", i));
        createFileWithSingleBlock(nxt);
      }
      parent.append("/nxt");
    }

    // go through each file without recursion
    parent = new StringBuilder();
    for (int j = 0; j < depthSize; j++) {
      long offset = 0;
      // the number of remaining files to list for this directory
      int remain;
      long listingCount;
      if (j == depthSize - 1) {
        remain = files;
        listingCount = files;
      } else {
        // +1 for the nested directory
        remain = files + 1;
        listingCount = files + 1;
      }
      for (int i = 0; i < files; i += batchSize) {
        ListStatusContext context = genListStatusPartial(
            batchSize, offset, false, "", "");
        infos = mFileSystemMaster.listStatus(new AlluxioURI("/" + parent), context);
        assertEquals(listingCount, context.getTotalListings());
        if (remain > batchSize) {
          assertTrue(context.isTruncated());
        } else {
          assertFalse(context.isTruncated());
        }
        // Copy out filenames to use List contains.
        filenames = infos.stream().map(FileInfo::getPath).collect(
            Collectors.toCollection(ArrayList::new));
        // check all the remaining files are listed
        assertEquals(Math.min(batchSize, remain), filenames.size());
        for (int k = i; k < Math.min(files, i + batchSize); k++) {
          assertTrue(filenames.contains("/" + parent + "file" + String.format("%05d", k)));
        }
        offset = infos.get(infos.size() - 1).getFileId();
        remain -= batchSize;
      }
      // listing of the nested directory
      if (j != depthSize - 1) {
        assertEquals("/" + parent + "nxt", filenames.get(filenames.size() - 1));
      }
      // all files should have been listed
      ListStatusContext context = genListStatusPartial(
          batchSize, offset, false, "", "");
      infos = mFileSystemMaster.listStatus(new AlluxioURI("/" + parent), context);
      assertFalse(context.isTruncated());
      assertEquals(0, infos.size());

      parent.append("nxt/");
    }
  }

  @Test
  public void listStatusStartAfterBatch() throws Exception {
    final int files = 13;
    final int batchSize = 5;
    final int depthSize = 5;
    List<FileInfo> infos;
    List<String> filenames = new ArrayList<>();

    StringBuilder parent = new StringBuilder();
    for (int j = 0; j < depthSize; j++) {
      // Test files in root directory.
      for (int i = 0; i < files; i++) {
        AlluxioURI nxt = new AlluxioURI(parent + "/file" + String.format("%05d", i));
        createFileWithSingleBlock(nxt);
      }
      parent.append("/nxt");
    }

    // go through each file without recursion
    parent = new StringBuilder();
    for (int j = 0; j < depthSize; j++) {
      String startAfter = "";
      // the number of remaining files to list for this directory
      int remain;
      long listingCount;
      if (j == depthSize - 1) {
        remain = files;
        listingCount = files;
      } else {
        // +1 for the nested directory
        remain = files + 1;
        listingCount = files + 1;
      }
      for (int i = 0; i < files; i += batchSize) {
        ListStatusContext context = genListStatusPartial(
            batchSize, 0, false, "", startAfter);
        infos = mFileSystemMaster.listStatus(new AlluxioURI("/" + parent), context);
        assertEquals(listingCount, context.getTotalListings());
        if (remain > batchSize) {
          assertTrue(context.isTruncated());
        } else {
          assertFalse(context.isTruncated());
        }
        // Copy out filenames to use List contains.
        filenames = infos.stream().map(FileInfo::getPath).collect(
            Collectors.toCollection(ArrayList::new));
        // check all the remaining files are listed
        assertEquals(Math.min(batchSize, remain), filenames.size());
        for (int k = i; k < Math.min(files, i + batchSize); k++) {
          assertTrue(filenames.contains("/" + parent + "file" + String.format("%05d", k)));
        }
        startAfter = infos.get(infos.size() - 1).getPath();
        remain -= batchSize;
      }
      // listing of the nested directory
      if (j != depthSize - 1) {
        assertEquals("/" + parent + "nxt", filenames.get(filenames.size() - 1));
      }
      // all files should have been listed
      ListStatusContext context = genListStatusPartial(
          batchSize, 0, false, "", startAfter);
      infos = mFileSystemMaster.listStatus(new AlluxioURI("/" + parent), context);
      assertFalse(context.isTruncated());
      assertEquals(0, infos.size());

      parent.append("nxt/");
    }
  }
}
