FilePatternMatchingShardedFile implements ShardedFile { @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; } return readLines(files); } catch (IOException e) { lastException = e; LOG.warn("Error in file reading. Ignore and retry."); } } while (BackOffUtils.next(sleeper, backOff)); throw new IOException( String.format("Unable to read file(s) after retrying %d times", MAX_READ_RETRIES), lastException); } FilePatternMatchingShardedFile(String filePattern); @Override List<String> readFilesWithRetries(Sleeper sleeper, BackOff backOff); List<String> readFilesWithRetries(); @Override String toString();  }
@Test public void testReadMultipleShardsWithoutShardNumber() throws Exception { String contents1 = "To be or not to be, "; String contents2 = "it is not a question."; File tmpFile1 = tmpFolder.newFile("result"); File tmpFile2 = tmpFolder.newFile("tmp"); Files.write(contents1, tmpFile1, StandardCharsets.UTF_8); Files.write(contents2, tmpFile2, StandardCharsets.UTF_8); FilePatternMatchingShardedFile shardedFile = new FilePatternMatchingShardedFile(filePattern); assertThat(shardedFile.readFilesWithRetries(), containsInAnyOrder(contents1, contents2)); }
@Test public void testReadEmpty() throws Exception { File emptyFile = tmpFolder.newFile("result-000-of-001"); Files.write("", emptyFile, StandardCharsets.UTF_8); FilePatternMatchingShardedFile shardedFile = new FilePatternMatchingShardedFile(filePattern); assertThat(shardedFile.readFilesWithRetries(), empty()); }
@Test public void testReadWithRetriesFailsWhenOutputDirEmpty() throws Exception { FilePatternMatchingShardedFile shardedFile = new FilePatternMatchingShardedFile(filePattern); thrown.expect(IOException.class); thrown.expectMessage(containsString("Unable to read file(s) after retrying")); shardedFile.readFilesWithRetries(fastClock, backOff); }