class DynamicFileListRecordReader extends RecordReader<K, V> {

  @Override
  public boolean nextKeyValue() throws IOException, InterruptedException {
    currentValue = null;

    // Check if we already have a reader in-progress.
    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()) {
      // Open the file and see if it's the 0-record end of dataset marker:
      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()) {
        // we found the end of dataset marker.
        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.");

    // No files ready and we shouldn't expect any more.
    return false;
  }

  public  DynamicFileListRecordReader(
      DelegateRecordReaderFactory<K, V> delegateRecordReaderFactory);

}