@Override
  public List<String> readFilesWithRetries(Sleeper sleeper, BackOff backOff)
      throws IOException, InterruptedException {
    IOException lastException = null;

    do {
      try {
        Collection<Metadata> files = FileSystems.match(filePattern).metadata();
        LOG.debug(
            "Found file(s) {} by matching the path: {}",
            files.stream()
                .map(Metadata::resourceId)
                .map(ResourceId::getFilename)
                .collect(Collectors.joining(",")),
            filePattern);
        if (files.isEmpty()) {
          continue;
        }
        // Read data from file paths
        return readLines(files);
      } catch (IOException e) {
        // Ignore and retry
        lastException = e;
        LOG.warn("Error in file reading. Ignore and retry.");
      }
    } while (BackOffUtils.next(sleeper, backOff));
    // Failed after max retries
    throw new IOException(
        String.format("Unable to read file(s) after retrying %d times", MAX_READ_RETRIES),
        lastException);
  }
@Test
  public void testReadEmpty() throws Exception {
    File emptyFile = tmpFolder.newFile("result-000-of-001");
    Files.asCharSink(emptyFile, StandardCharsets.UTF_8).write("");
    FilePatternMatchingShardedFile shardedFile = new FilePatternMatchingShardedFile(filePattern);

    assertThat(shardedFile.readFilesWithRetries(), empty());
  }