NumberedShardedFile implements ShardedFile { @Override public List<String> readFilesWithRetries(Sleeper sleeper, BackOff backOff) throws IOException, InterruptedException { IOException lastException = null; do { try { 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; } 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); } NumberedShardedFile(String filePattern);  NumberedShardedFile(String filePattern, Pattern shardTemplate); String getFilePattern(); @Override List<String> readFilesWithRetries(Sleeper sleeper, BackOff backOff); List<String> readFilesWithRetries(); @Override String toString();  }
@Test public void testReadEmpty() throws Exception { File emptyFile = tmpFolder.newFile("result-000-of-001"); Files.write("", emptyFile, StandardCharsets.UTF_8); NumberedShardedFile shardedFile = new NumberedShardedFile(filePattern); assertThat(shardedFile.readFilesWithRetries(), empty()); }
@Test public void testReadCustomTemplate() throws Exception { String contents1 = "To be or not to be, ", contents2 = "it is not a question."; File tmpFile1 = tmpFolder.newFile("result0-total2"); File tmpFile2 = tmpFolder.newFile("result1-total2"); Files.write(contents1, tmpFile1, StandardCharsets.UTF_8); Files.write(contents2, tmpFile2, StandardCharsets.UTF_8); Pattern customizedTemplate = Pattern.compile("(?x) result (?<shardnum>\\d+) - total (?<numshards>\\d+)"); NumberedShardedFile shardedFile = new NumberedShardedFile(filePattern, customizedTemplate); assertThat(shardedFile.readFilesWithRetries(), containsInAnyOrder(contents1, contents2)); }
@Test public void testReadWithRetriesFailsWhenTemplateIncorrect() throws Exception { File tmpFile = tmpFolder.newFile(); Files.write("Test for file checksum verifier.", tmpFile, StandardCharsets.UTF_8); NumberedShardedFile shardedFile = new NumberedShardedFile(filePattern, Pattern.compile("incorrect-template")); thrown.expect(IOException.class); thrown.expectMessage( containsString( "Unable to read file(s) after retrying " + NumberedShardedFile.MAX_READ_RETRIES)); shardedFile.readFilesWithRetries(fastClock, backOff); }
@Test public void testReadWithRetriesFailsWhenOutputDirEmpty() throws Exception { 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); }
@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); }