DynamicFileListRecordReader extends RecordReader<K, V> { @Override public boolean nextKeyValue() throws IOException, InterruptedException { currentValue = null; if (delegateReader != null) { if (delegateReader.nextKeyValue()) { populateCurrentKeyValue(); return true; } else { delegateReader.close(); delegateReader = null; } } boolean needRefresh = !isNextFileReady() && shouldExpectMoreFiles(); int pollAttempt = 0; while (needRefresh && (maxPollAttempts < 0 || pollAttempt < maxPollAttempts)) { logger.atFine().log( "No files available after %d attempt(s), but more are expected; refreshing ...", pollAttempt + 1); refreshFileList(); needRefresh = !isNextFileReady() && shouldExpectMoreFiles(); if (needRefresh) { logger.atFine().log("No new files found, sleeping before trying again..."); sleeper.sleep(pollIntervalMs); context.progress(); pollAttempt++; } } if (needRefresh) { throw new IllegalStateException( String.format( "Couldn't obtain any files after %d attempt(s). This could happen if in the first" + " failed task attempt BigQuery returned 0 records, but didn't create 0-record" + " file, in this case the second task attempt record reader will wait" + " indefinitely, but no files will appear.", pollAttempt)); } if (isNextFileReady()) { FileStatus newFile = moveToNextFile(); logger.atInfo().log( "Moving to next file '%s' which has %s bytes. Records read so far: %s", newFile.getPath(), newFile.getLen(), recordsRead); InputSplit split = new FileSplit(newFile.getPath(), 0, newFile.getLen(), new String[0]); delegateReader = delegateRecordReaderFactory.createDelegateRecordReader( split, context.getConfiguration()); delegateReader.initialize(split, context); if (!delegateReader.nextKeyValue()) { setEndFileMarkerFile(newFile.getPath().getName()); return nextKeyValue(); } else { populateCurrentKeyValue(); return true; } } Preconditions.checkState( !shouldExpectMoreFiles(), "Should not have exited the refresh loop shouldExpectMoreFiles = true " + "and no files ready to read."); return false; } DynamicFileListRecordReader(
      DelegateRecordReaderFactory<K, V> delegateRecordReaderFactory); @Override void initialize(InputSplit genericSplit, TaskAttemptContext context); @Override boolean nextKeyValue(); @Override K getCurrentKey(); @Override V getCurrentValue(); @Override float getProgress(); @Override void close();  }
@Test public void testEmptyFileIsOnlyFileAndNotZeroIndex() throws IOException { fileSystem.createNewFile(new Path(shardPath.getParent(), "data-001.json")); checkNextKeyValueWouldBlock(); fileSystem.createNewFile(new Path(shardPath.getParent(), "data-002.json")); assertThrows(IllegalStateException.class, () -> recordReader.nextKeyValue()); }
@Test public void testBadFilename() throws Exception { String outOfBounds = String.format("data-%d.json", 1L + Integer.MAX_VALUE); fileSystem.createNewFile(new Path(shardPath.getParent(), outOfBounds)); assertThrows(IndexOutOfBoundsException.class, () -> recordReader.nextKeyValue()); }