@Override
  public List<FileMetadata> listFilesWithMetadata(URI fileUri, boolean recursive)
      throws IOException {
    ImmutableList.Builder<FileMetadata> listBuilder = ImmutableList.builder();
    visitFiles(fileUri, recursive, f -> {
      FileMetadata.Builder fileBuilder =
          new FileMetadata.Builder().setFilePath(f.getPath().toString()).setLastModifiedTime(f.getModificationTime())
              .setLength(f.getLen()).setIsDirectory(f.isDirectory());
      listBuilder.add(fileBuilder.build());
    });
    ImmutableList<FileMetadata> listedFiles = listBuilder.build();
    LOGGER.debug("Listed {} files from URI: {}, is recursive: {}", listedFiles.size(), fileUri, recursive);
    return listedFiles;
  }
@Test
  public void testListFilesWithMetadata()
      throws IOException {
    URI baseURI = URI.create(TMP_DIR + "/testListFilesWithMetadata");
    try (HadoopPinotFS hadoopFS = new HadoopPinotFS()) {
      hadoopFS.init(new PinotConfiguration());

      // Create a testDir and file underneath directory
      int count = 5;
      List<String> expectedNonRecursive = new ArrayList<>();
      List<String> expectedRecursive = new ArrayList<>();
      for (int i = 0; i < count; i++) {
        URI testDir = new Path(baseURI.getPath(), "testDir" + i).toUri();
        hadoopFS.mkdir(testDir);
        expectedNonRecursive.add((testDir.getScheme() == null ? "file:" : "") + testDir);

        URI testFile = new Path(testDir.getPath(), "testFile" + i).toUri();
        hadoopFS.touch(testFile);
        expectedRecursive.add((testDir.getScheme() == null ? "file:" : "") + testDir);
        expectedRecursive.add((testDir.getScheme() == null ? "file:" : "") + testFile);
      }
      URI testDirEmpty = new Path(baseURI.getPath(), "testDirEmpty").toUri();
      hadoopFS.mkdir(testDirEmpty);
      expectedNonRecursive.add((testDirEmpty.getScheme() == null ? "file:" : "") + testDirEmpty);
      expectedRecursive.add((testDirEmpty.getScheme() == null ? "file:" : "") + testDirEmpty);

      URI testRootFile = new Path(baseURI.getPath(), "testRootFile").toUri();
      hadoopFS.touch(testRootFile);
      expectedNonRecursive.add((testRootFile.getScheme() == null ? "file:" : "") + testRootFile);
      expectedRecursive.add((testRootFile.getScheme() == null ? "file:" : "") + testRootFile);

      // Assert that recursive list files and nonrecursive list files are as expected
      String[] files = hadoopFS.listFiles(baseURI, false);
      Assert.assertEquals(files.length, count + 2);
      Assert.assertTrue(expectedNonRecursive.containsAll(Arrays.asList(files)), Arrays.toString(files));
      files = hadoopFS.listFiles(baseURI, true);
      Assert.assertEquals(files.length, count * 2 + 2);
      Assert.assertTrue(expectedRecursive.containsAll(Arrays.asList(files)), Arrays.toString(files));

      // Assert that recursive list files and nonrecursive list files with file info are as expected
      List<FileMetadata> fileMetadata = hadoopFS.listFilesWithMetadata(baseURI, false);
      Assert.assertEquals(fileMetadata.size(), count + 2);
      Assert.assertEquals(fileMetadata.stream().filter(FileMetadata::isDirectory).count(), count + 1);
      Assert.assertEquals(fileMetadata.stream().filter(f -> !f.isDirectory()).count(), 1);
      Assert.assertTrue(expectedNonRecursive.containsAll(
          fileMetadata.stream().map(FileMetadata::getFilePath).collect(Collectors.toSet())), fileMetadata.toString());
      fileMetadata = hadoopFS.listFilesWithMetadata(baseURI, true);
      Assert.assertEquals(fileMetadata.size(), count * 2 + 2);
      Assert.assertEquals(fileMetadata.stream().filter(FileMetadata::isDirectory).count(), count + 1);
      Assert.assertEquals(fileMetadata.stream().filter(f -> !f.isDirectory()).count(), count + 1);
      Assert.assertTrue(expectedRecursive.containsAll(
          fileMetadata.stream().map(FileMetadata::getFilePath).collect(Collectors.toSet())), fileMetadata.toString());
    }
  }