@ 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 ( ) ; while ( needRefresh ) { logger . atFine ( ) . log ( "No files available, but more are expected; refreshing..." ) ; refreshFileList ( ) ; needRefresh = ! isNextFileReady ( ) && shouldExpectMoreFiles ( ) ; if ( needRefresh ) { logger . atFine ( ) . log ( "No new files found, sleeping before trying again..." ) ; try { sleeper . sleep ( pollIntervalMs ) ; context . progress ( ) ; } catch ( InterruptedException ie ) { logger . atWarning ( ) . withCause ( ie ) . log ( "Interrupted while sleeping." ) ; } } } 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 ; }
Reads the next key value pair . Gets next line and parses Json object . May hang for a long time waiting for more files to appear in this reader s directory .