nl_tokens: {
pl_tokens: @ Override public String produce ( String lastSourceOffset , int maxBatchSize , BatchMaker batchMaker ) throws StageException { // Offsets can vary depending on the data source. Here we use an integer as an example only. long nextSourceOffset = 0 ; lastSourceOffset = lastSourceOffset == null ? "" : lastSourceOffset ; if ( ! lastSourceOffset . equals ( "" ) ) { nextSourceOffset = Long . parseLong ( lastSourceOffset ) ; } int recordCounter = 0 ; long startTime = System . currentTimeMillis ( ) ; int maxRecords = Math . <mask> ( maxBatchSize , conf . maxBatchSize ) ; while ( recordCounter < maxRecords && ( startTime + conf . maxWaitTime ) > System . currentTimeMillis ( ) ) { String message = buffer . poll ( ) ; if ( null == message ) { try { Thread . sleep ( 100 ) ; } catch ( Exception e ) { LOG . debug ( e . getMessage ( ) , e ) ; break ; } } else { List < Record > records = processRedisMessage ( "id::" + nextSourceOffset , message ) ; for ( Record record : records ) { batchMaker . addRecord ( record ) ; } recordCounter += records . size ( ) ; ++ nextSourceOffset ; } } return lastSourceOffset ; }