@Override
  public List<FileMetadata> listFilesWithMetadata(URI fileUri, boolean recursive)
      throws IOException {
    ImmutableList.Builder<FileMetadata> listBuilder = ImmutableList.builder();
    GcsUri gcsFileUri = new GcsUri(fileUri);
    String prefix = gcsFileUri.getPrefix();
    String bucketName = gcsFileUri.getBucketName();
    visitFiles(gcsFileUri, recursive, blob -> {
      if (!blob.getName().equals(prefix)) {
        // Note: isDirectory flag is only set when listing with BlobListOption.currentDirectory() i.e non-recursively.
        // For simplicity, we check if a path is directory by checking if it ends with '/', as done in S3PinotFS.
        boolean isDirectory = blob.getName().endsWith(GcsUri.DELIMITER);
        FileMetadata.Builder fileBuilder =
            new FileMetadata.Builder().setFilePath(GcsUri.createGcsUri(bucketName, blob.getName()).toString())
                .setLength(blob.getSize()).setIsDirectory(isDirectory);
        if (!isDirectory) {
          // Note: if it's a directory, updateTime is set to null, and calling this getter leads to NPE.
          // public Long getUpdateTime() { return updateTime; }. So skip this for directory.
          fileBuilder.setLastModifiedTime(blob.getUpdateTime());
        }
        listBuilder.add(fileBuilder.build());
      }
    });
    ImmutableList<FileMetadata> listedFiles = listBuilder.build();
    LOGGER.info("Listed {} files from URI: {}, is recursive: {}", listedFiles.size(), gcsFileUri, recursive);
    return listedFiles;
  }
@Test
  public void testListFilesWithMetadata()
      throws Exception {
    skipIfNotConfigured();

    // Create empty file
    Path emptyFile = _localTmpDir.resolve("empty");
    emptyFile.toFile().createNewFile();

    // Create 5 subfolders with files inside.
    int count = 5;
    Set<String> expectedNonRecursive = new HashSet<>();
    Set<String> expectedRecursive = new HashSet<>();
    for (int i = 0; i < count; i++) {
      GcsUri testDir = _dataDir.resolve("testDir" + i);
      _pinotFS.mkdir(testDir.getUri());
      expectedNonRecursive.add(appendSlash(testDir).toString());

      GcsUri testFile = testDir.resolve("testFile" + i);
      // Create the file by copying an empty file there.
      _pinotFS.copyFromLocalFile(emptyFile.toFile(), testFile.getUri());
      expectedRecursive.add(appendSlash(testDir).toString());
      expectedRecursive.add(testFile.toString());
    }
    GcsUri testDirEmpty = _dataDir.resolve("testDirEmpty");
    _pinotFS.mkdir(testDirEmpty.getUri());
    expectedNonRecursive.add(appendSlash(testDirEmpty).toString());
    expectedRecursive.add(appendSlash(testDirEmpty).toString());

    GcsUri testRootFile = _dataDir.resolve("testRootFile");
    _pinotFS.copyFromLocalFile(emptyFile.toFile(), testRootFile.getUri());
    expectedNonRecursive.add(testRootFile.toString());
    expectedRecursive.add(testRootFile.toString());

    // Assert that recursive list files and nonrecursive list files are as expected
    String[] files = _pinotFS.listFiles(_dataDir.getUri(), false);
    Assert.assertEquals(files.length, count + 2);
    Assert.assertTrue(expectedNonRecursive.containsAll(Arrays.asList(files)), Arrays.toString(files));
    files = _pinotFS.listFiles(_dataDir.getUri(), 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 = _pinotFS.listFilesWithMetadata(_dataDir.getUri(), 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 = _pinotFS.listFilesWithMetadata(_dataDir.getUri(), 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());
  }