basic-lib/src/main/java/com/streamsets/pipeline/stage/processor/http/HttpProcessor.java
@Override
  public void process(Batch batch, SingleLaneBatchMaker batchMaker) throws StageException {
    List<Future<Response>> responses = new ArrayList<>();
    resolvedRecords.clear();

    Iterator<Record> records = batch.getRecords();
    while (records.hasNext()) {

      Record record = records.next();
      String resolvedUrl = httpClientCommon.getResolvedUrl(conf.resourceUrl, record);
      WebTarget target = httpClientCommon.getClient().target(resolvedUrl);

      LOG.debug("Resolved HTTP Client URL: '{}'",resolvedUrl);

      // If the request (headers or body) contain a known sensitive EL and we're not using https then fail the request.
      if (httpClientCommon.requestContainsSensitiveInfo(conf.headers, conf.requestBody) &&
          !target.getUri().getScheme().toLowerCase().startsWith("https")) {
        throw new StageException(Errors.HTTP_07);
      }

      // from HttpStreamConsumer
      final MultivaluedMap<String, Object> resolvedHeaders = httpClientCommon.resolveHeaders(conf.headers, record);

      String contentType = HttpStageUtil.getContentTypeWithDefault(resolvedHeaders, conf.defaultRequestContentType);

      final AsyncInvoker asyncInvoker = target.request()
          .property(OAuth1ClientSupport.OAUTH_PROPERTY_ACCESS_TOKEN, httpClientCommon.getAuthToken())
          .headers(resolvedHeaders)
          .async();

      HttpMethod method = httpClientCommon.getHttpMethod(conf.httpMethod, conf.methodExpression, record);

      rateLimiter.acquire();
      if (conf.requestBody != null && !conf.requestBody.isEmpty() && method != HttpMethod.GET) {
        RecordEL.setRecordInContext(bodyVars, record);
        final String requestBody = bodyEval.eval(bodyVars, conf.requestBody, String.class);
        resolvedRecords.put(record, new HeadersAndBody(resolvedHeaders, requestBody, contentType, method, target));
        responses.add(asyncInvoker.method(method.getLabel(), Entity.entity(requestBody, contentType)));
      } else {
        resolvedRecords.put(record, new HeadersAndBody(resolvedHeaders, null, null, method, target));
        responses.add(asyncInvoker.method(method.getLabel()));
      }
    }

    records = batch.getRecords();
    int recordNum = 0;
    while (records.hasNext()) {
      try {
        Record record = processResponse(records.next(), responses.get(recordNum), conf.maxRequestCompletionSecs, false);
        if (record != null) {
          batchMaker.addRecord(record);
        }
      } catch (OnRecordErrorException e) {
        errorRecordHandler.onError(e);
      } finally {
        ++recordNum;
      }
    }
    if (!resolvedRecords.isEmpty()) {
      reprocessIfRequired(batchMaker);
    }
  }
@Override
  public void process(Batch batch, SingleLaneBatchMaker batchMaker) throws StageException {
    List<Future<Response>> responses = new ArrayList<>();
    resolvedRecords.clear();

    Iterator<Record> records = batch.getRecords();
    while (records.hasNext()) {

      Record record = records.next();
      String resolvedUrl = httpClientCommon.getResolvedUrl(conf.resourceUrl, record);
      WebTarget target = httpClientCommon.getClient().target(resolvedUrl);

      LOG.debug("Resolved HTTP Client URL: '{}'",resolvedUrl);

      // If the request (headers or body) contain a known sensitive EL and we're not using https then fail the request.
      if (httpClientCommon.requestContainsSensitiveInfo(conf.headers, conf.requestBody) &&
          !target.getUri().getScheme().toLowerCase().startsWith("https")) {
        throw new StageException(Errors.HTTP_07);
      }

      // from HttpStreamConsumer
      final MultivaluedMap<String, Object> resolvedHeaders = httpClientCommon.resolveHeaders(conf.headers, record);

      String contentType = HttpStageUtil.getContentTypeWithDefault(resolvedHeaders, conf.defaultRequestContentType);

      final AsyncInvoker asyncInvoker = target.request()
          .property(OAuth1ClientSupport.OAUTH_PROPERTY_ACCESS_TOKEN, httpClientCommon.getAuthToken())
          .headers(resolvedHeaders)
          .async();

      HttpMethod method = httpClientCommon.getHttpMethod(conf.httpMethod, conf.methodExpression, record);

      rateLimiter.acquire();
      if (conf.requestBody != null && !conf.requestBody.isEmpty() && method != HttpMethod.GET) {
        RecordEL.setRecordInContext(bodyVars, record);
        final String requestBody = bodyEval.eval(bodyVars, conf.requestBody, String.class);
        resolvedRecords.put(record, new HeadersAndBody(resolvedHeaders, requestBody, contentType, method, target));
        responses.add(asyncInvoker.method(method.getLabel(), Entity.entity(requestBody, contentType)));
      } else {
        resolvedRecords.put(record, new HeadersAndBody(resolvedHeaders, null, null, method, target));
        responses.add(asyncInvoker.method(method.getLabel()));
      }
    }

    records = batch.getRecords();
    int recordNum = 0;
    while (records.hasNext()) {
      try {
        Record record = processResponse(records.next(), responses.get(recordNum), conf.maxRequestCompletionSecs, false);
        if (record != null) {
          batchMaker.addRecord(record);
        }
      } catch (OnRecordErrorException e) {
        errorRecordHandler.onError(e);
      } finally {
        ++recordNum;
      }
    }
    if (!resolvedRecords.isEmpty()) {
      reprocessIfRequired(batchMaker);
    }
  }
{@inheritDoc}