basic-lib/src/main/java/com/streamsets/pipeline/stage/processor/http/HttpProcessor.java
private Record processResponse(
      Record record,
      Future<Response> responseFuture,
      long maxRequestCompletionSecs,
      boolean failOn403
  ) throws StageException {

    Response response = null;
    try {
      response = responseFuture.get(maxRequestCompletionSecs, TimeUnit.SECONDS);
      InputStream responseBody = null;
      if (response.hasEntity()) {
        responseBody = response.readEntity(InputStream.class);
      }
      int responseStatus = response.getStatus();
      if (conf.client.useOAuth2 && response.getStatus() == 403 && !failOn403) {
        HttpStageUtil.getNewOAuth2Token(conf.client.oauth2, httpClientCommon.getClient());
        return null;
      } else if (responseStatus < 200 || responseStatus >= 300) {
        resolvedRecords.remove(record);
        throw new OnRecordErrorException(
            record,
            Errors.HTTP_01,
            response.getStatus(),
            response.getStatusInfo().getReasonPhrase() + " " + responseBody
        );
      }
      resolvedRecords.remove(record);
      Record parsedResponse = parseResponse(responseBody);
      if (parsedResponse != null) {
        record.set(conf.outputField, parsedResponse.get());
        addResponseHeaders(record, response);
      } else if (responseBody == null && responseStatus != 204) {
        throw new OnRecordErrorException(record, Errors.HTTP_34);
      }
      return record;
    } catch (InterruptedException | ExecutionException e) {
      LOG.error(Errors.HTTP_03.getMessage(), e.toString(), e);
      throw new OnRecordErrorException(record, Errors.HTTP_03, e.toString());
    } catch (TimeoutException e) {
      LOG.error("HTTP request future timed out", e.toString(), e);
      throw new OnRecordErrorException(record, Errors.HTTP_03, e.toString());
    } finally {
      if (response != null) {
        response.close();
      }
    }
  }
private Record processResponse(
      Record record,
      Future<Response> responseFuture,
      long maxRequestCompletionSecs,
      boolean failOn403
  ) throws StageException {

    Response response = null;
    try {
      response = responseFuture.get(maxRequestCompletionSecs, TimeUnit.SECONDS);
      InputStream responseBody = null;
      if (response.hasEntity()) {
        responseBody = response.readEntity(InputStream.class);
      }
      int responseStatus = response.getStatus();
      if (conf.client.useOAuth2 && response.getStatus() == 403 && !failOn403) {
        HttpStageUtil.getNewOAuth2Token(conf.client.oauth2, httpClientCommon.getClient());
        return null;
      } else if (responseStatus < 200 || responseStatus >= 300) {
        resolvedRecords.remove(record);
        throw new OnRecordErrorException(
            record,
            Errors.HTTP_01,
            response.getStatus(),
            response.getStatusInfo().getReasonPhrase() + " " + responseBody
        );
      }
      resolvedRecords.remove(record);
      Record parsedResponse = parseResponse(responseBody);
      if (parsedResponse != null) {
        record.set(conf.outputField, parsedResponse.get());
        addResponseHeaders(record, response);
      } else if (responseBody == null && responseStatus != 204) {
        throw new OnRecordErrorException(record, Errors.HTTP_34);
      }
      return record;
    } catch (InterruptedException | ExecutionException e) {
      LOG.error(Errors.HTTP_03.getMessage(), e.toString(), e);
      throw new OnRecordErrorException(record, Errors.HTTP_03, e.toString());
    } catch (TimeoutException e) {
      LOG.error("HTTP request future timed out", e.toString(), e);
      throw new OnRecordErrorException(record, Errors.HTTP_03, e.toString());
    } finally {
      if (response != null) {
        response.close();
      }
    }
  }
private Record processResponse ( Record record , Future < Response > responseFuture , long maxRequestCompletionSecs , boolean failOn403 ) throws StageException { Response response = null ; try { response = responseFuture . get ( maxRequestCompletionSecs , TimeUnit . SECONDS ) ; InputStream responseBody = null ; if ( response . hasEntity ( ) ) { responseBody = response . readEntity ( InputStream . class ) ; } int responseStatus = response . getStatus ( ) ; if ( conf . client . useOAuth2 && response . getStatus ( ) == 403 && ! failOn403 ) { HttpStageUtil . getNewOAuth2Token ( conf . client . oauth2 , httpClientCommon . getClient ( ) ) ; return null ; } else if ( responseStatus < 200 || responseStatus >= 300 ) { resolvedRecords . remove ( record ) ; throw new OnRecordErrorException ( record , Errors . HTTP_01 , response . getStatus ( ) , response . getStatusInfo ( ) . getReasonPhrase ( ) + " " + responseBody ) ; } resolvedRecords . remove ( record ) ; Record parsedResponse = parseResponse ( responseBody ) ; if ( parsedResponse != null ) { record . set ( conf . outputField , parsedResponse . get ( ) ) ; addResponseHeaders ( record , response ) ; } else if ( responseBody == null && responseStatus != 204 ) { throw new OnRecordErrorException ( record , Errors . HTTP_34 ) ; } return record ; } catch ( InterruptedException | ExecutionException e ) { LOG . error ( Errors . HTTP_03 . getMessage ( ) , e . toString ( ) , e ) ; throw new OnRecordErrorException ( record , Errors . HTTP_03 , e . toString ( ) ) ; } catch ( TimeoutException e ) { LOG . error ( "HTTP request future timed out" , e . toString ( ) , e ) ; throw new OnRecordErrorException ( record , Errors . HTTP_03 , e . toString ( ) ) ; } finally { if ( response != null ) { response . close ( ) ; } } }
Waits for the Jersey client to complete an asynchronous request, checks the response code
and continues to parse the response if it is deemed ok.

@param record the current record to set in context for any expression evaluation
@param responseFuture the async HTTP request future
@param maxRequestCompletionSecs maximum time to wait for request completion (start to finish)
@return parsed record from the request
@throws StageException if the request fails, times out, or cannot be parsed
Waits for the Jersey client to complete an asynchronous request checks the response code and continues to parse the response if it is deemed ok .