nl_tokens: {
pl_tokens: @ Override public String produce ( String lastSourceOffset , int maxBatchSize , BatchMaker batchMaker ) throws StageException { String nextSourceOffset = null ; LOG . debug ( "lastSourceOffset: {}" , lastSourceOffset ) ; // send event only once for each time we run out of data. if ( shouldSendNoMoreDataEvent ) { CommonEvents . NO_MORE_DATA . create ( getContext ( ) ) . createAndSend ( ) ; shouldSendNoMoreDataEvent = false ; return lastSourceOffset ; } int batchSize = Math . <mask> ( conf . basicConfig . maxBatchSize , maxBatchSize ) ; if ( ! conf . queryExistingData || ( null != lastSourceOffset && lastSourceOffset . startsWith ( EVENT_ID_OFFSET_PREFIX ) ) ) { if ( conf . subscribeToStreaming ) { nextSourceOffset = streamingProduce ( lastSourceOffset , batchSize , batchMaker ) ; } else { // We're done reading existing data, but we don't want to subscribe to Streaming API return null ; } } else if ( conf . queryExistingData ) { if ( ! queryInProgress ( ) ) { long now = System . currentTimeMillis ( ) ; long delay = Math . max ( 0 , ( lastQueryCompletedTime + ( 1000 * conf . queryInterval ) ) - now ) ; if ( delay > 0 ) { // Sleep in one second increments so we don't tie up the app. LOG . info ( "{}ms remaining until next fetch." , delay ) ; ThreadUtil . sleep ( Math . min ( delay , 1000 ) ) ; return lastSourceOffset ; } } if ( conf . useBulkAPI ) { nextSourceOffset = bulkProduce ( lastSourceOffset , batchSize , batchMaker ) ; } else { nextSourceOffset = soapProduce ( lastSourceOffset , batchSize , batchMaker ) ; } } else if ( conf . subscribeToStreaming ) { // No offset, and we're not querying existing data, so switch to streaming nextSourceOffset = READ_EVENTS_FROM_NOW ; } LOG . debug ( "nextSourceOffset: {}" , nextSourceOffset ) ; return nextSourceOffset ; }