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

    do {
      try {
        // Match inputPath which may contains glob
        Collection<Metadata> files =
            Iterables.getOnlyElement(FileSystems.match(Collections.singletonList(filePattern)))
                .metadata();

        LOG.debug("Found {} file(s) by matching the path: {}", files.size(), filePattern);

        if (files.isEmpty() || !checkTotalNumOfFiles(files)) {
          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 testReadWithRetriesFailsWhenRedundantFileLoaded() throws Exception {
    tmpFolder.newFile("result-000-of-001");
    tmpFolder.newFile("tmp-result-000-of-001");

    NumberedShardedFile shardedFile = new NumberedShardedFile(filePattern);

    thrown.expect(IOException.class);
    thrown.expectMessage(
        containsString(
            "Unable to read file(s) after retrying " + NumberedShardedFile.MAX_READ_RETRIES));
    shardedFile.readFilesWithRetries(fastClock, backOff);
  }