nl_tokens: / * When we start with a file ( empty or not ) the file offset is zero . If the file is a rolled file the file will be EOF immediately triggering a close of the reader and setting the offset to Long . MAX_VALUE ( this happens in the MultiDirectoryReader class ) . This is the signal that in the next read a directory scan should be triggered to get the next rolled file or the live file if we were scanning the last rolled file . If the file you are starting is the live file we don t get an EOF as we expect data to be appended . We just return null chunks while there is no data . If the file is rolled we ll detect that and then do what is described in the previous paragraph .
pl_tokens: @ Override public String produce ( String lastSourceOffset , int maxBatchSize , BatchMaker batchMaker ) throws StageException { int recordCounter = 0 ; long startTime = System . currentTimeMillis ( ) ; maxBatchSize = Math . <mask> ( conf . batchSize , maxBatchSize ) ; // deserializing offsets of all directories Map < String , String > offsetMap = deserializeOffsetMap ( lastSourceOffset ) ; boolean offsetSet = false ; while ( ! offsetSet ) { try { multiDirReader . setOffsets ( offsetMap ) ; offsetSet = true ; } catch ( IOException ex ) { LOG . warn ( "Error while creating reading previous offset: {}" , ex . toString ( ) , ex ) ; multiDirReader . purge ( ) ; } } while ( recordCounter < maxBatchSize && ! isTimeout ( startTime ) ) { LiveFileChunk chunk = multiDirReader . next ( getRemainingWaitTime ( startTime ) ) ; if ( chunk != null ) { String tag = chunk . getTag ( ) ; tag = ( tag != null && tag . isEmpty ( ) ) ? null : tag ; String liveFileStr = chunk . getFile ( ) . serialize ( ) ; List < FileLine > lines = chunk . getLines ( ) ; int truncatedLine = chunk . isTruncated ( ) ? lines . size ( ) - 1 : - 1 ; for ( int i = 0 ; i < lines . size ( ) ; i ++ ) { FileLine line = lines . get ( i ) ; String sourceId = liveFileStr + "::" + line . getFileOffset ( ) ; try ( DataParser parser = parserFactory . getParser ( sourceId , line . getText ( ) ) ) { if ( i == truncatedLine ) { //set truncated parser . setTruncated ( ) ; } Record record = parser . parse ( ) ; if ( record != null ) { if ( tag != null ) { record . getHeader ( ) . setAttribute ( "tag" , tag ) ; } record . getHeader ( ) . setAttribute ( HeaderAttributeConstants . FILE , chunk . getFile ( ) . getPath ( ) . toString ( ) ) ; record . getHeader ( ) . setAttribute ( HeaderAttributeConstants . FILE_NAME , chunk . getFile ( ) . getPath ( ) . getFileName ( ) . toString ( ) ) ; record . getHeader ( ) . setAttribute ( HeaderAttributeConstants . OFFSET , String . valueOf ( line . getFileOffset ( ) ) ) ; record . getHeader ( ) . setAttribute ( HeaderAttributeConstants . LAST_MODIFIED_TIME , String . valueOf ( Files . getLastModifiedTime ( chunk . getFile ( ) . getPath ( ) ) . toMillis ( ) ) ) ; batchMaker . addRecord ( record , outputLane ) ; recordCounter ++ ; } } catch ( IOException | DataParserException ex ) { errorRecordHandler . onError ( Errors . TAIL_12 , sourceId , ex . toString ( ) , ex ) ; } } } } boolean metadataGenerationFailure = false ; Date now = new Date ( startTime ) ; for ( FileEvent event : multiDirReader . getEvents ( ) ) { try { LiveFile file = event . getFile ( ) . refresh ( ) ; Record metadataRecord = getContext ( ) . createRecord ( "" ) ; Map < String , Field > map = new HashMap <> ( ) ; map . put ( "fileName" , Field . create ( file . getPath ( ) . toString ( ) ) ) ; map . put ( "inode" , Field . create ( file . getINode ( ) ) ) ; map . put ( "time" , Field . createDate ( now ) ) ; map . put ( "event" , Field . create ( ( event . getAction ( ) . name ( ) ) ) ) ; metadataRecord . set ( Field . create ( map ) ) ; batchMaker . addRecord ( metadataRecord , metadataLane ) ; // We're also sending the same information on event lane String eventRecordSourceId = Utils . format ( "event:{}:{}:{}" , event . getAction ( ) . name ( ) , 1 , file . getPath ( ) . toString ( ) ) ; EventRecord eventRecord = getContext ( ) . createEventRecord ( event . getAction ( ) . name ( ) , 1 , eventRecordSourceId ) ; eventRecord . set ( Field . create ( map ) ) ; getContext ( ) . toEvent ( eventRecord ) ; } catch ( IOException ex ) { LOG . warn ( "Error while creating metadata records: {}" , ex . toString ( ) , ex ) ; metadataGenerationFailure = true ; } } if ( metadataGenerationFailure ) { multiDirReader . purge ( ) ; } boolean offsetExtracted = false ; while ( ! offsetExtracted ) { try { offsetMap = multiDirReader . getOffsets ( ) ; offsetExtracted = true ; } catch ( IOException ex ) { LOG . warn ( "Error while creating creating new offset: {}" , ex . toString ( ) , ex ) ; multiDirReader . purge ( ) ; } } //Calculate Offset lag Metric. calculateOffsetLagMetric ( offsetMap ) ; //Calculate Pending Files Metric calculatePendingFilesMetric ( ) ; // serializing offsets of all directories return serializeOffsetMap ( offsetMap ) ; }