nl_tokens: {
pl_tokens: @ Override public String produce ( String lastSourceOffset , int maxBatchSize , BatchMaker batchMaker ) throws StageException { long start = System . currentTimeMillis ( ) ; int chunksToFetch = Math . <mask> ( conf . basic . maxBatchSize , maxBatchSize ) ; Optional < String > newSourceOffset = Optional . empty ( ) ; recordCount = 0 ; setPageOffset ( lastSourceOffset ) ; setResolvedUrl ( resolveInitialUrl ( lastSourceOffset ) ) ; WebTarget target = client . target ( getResolvedUrl ( ) ) ; // If the request (headers or body) contain a known sensitive EL and we're not using https then fail the request. if ( requestContainsSensitiveInfo ( ) && ! target . getUri ( ) . getScheme ( ) . toLowerCase ( ) . startsWith ( "https" ) ) { LOG . error ( Errors . HTTP_07 . getMessage ( ) ) ; throw new StageException ( Errors . HTTP_07 ) ; } boolean uninterrupted = true ; while ( ! waitTimeExpired ( start ) && uninterrupted && ( recordCount < chunksToFetch ) ) { if ( parser != null ) { // We already have an response that we haven't finished reading. newSourceOffset = Optional . of ( parseResponse ( start , chunksToFetch , batchMaker ) ) ; } else if ( shouldMakeRequest ( ) ) { if ( conf . pagination . mode != PaginationMode . NONE ) { target = client . target ( resolveNextPageUrl ( newSourceOffset . orElse ( null ) ) ) ; // Pause between paging requests so we don't get rate limited. uninterrupted = ThreadUtil . sleep ( conf . pagination . rateLimit ) ; } makeRequest ( target ) ; if ( lastRequestTimedOut ) { String actionName = conf . responseTimeoutActionConfig . getAction ( ) . name ( ) ; LOG . warn ( "HTTPClient timed out after waiting {} ms for response from server;" + " reconnecting client and proceeding as per configured {} action" , conf . client . readTimeoutMillis , actionName ) ; reconnectClient ( ) ; return nonTerminating ( lastSourceOffset ) ; } else { newSourceOffset = processResponse ( start , chunksToFetch , batchMaker ) ; } } else if ( conf . httpMode == HttpClientMode . BATCH ) { // We are done. return null ; } else { // In polling mode, waiting for the next polling interval. uninterrupted = ThreadUtil . sleep ( SLEEP_TIME_WAITING_FOR_BATCH_SIZE_MS ) ; } } return newSourceOffset . orElse ( lastSourceOffset ) ; }