basic-lib/src/main/java/com/streamsets/pipeline/stage/origin/http/HttpClientSource.java
@Override
  public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) throws StageException {
    long start = System.currentTimeMillis();
    int chunksToFetch = Math.min(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);
  }
@Override
  public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) throws StageException {
    long start = System.currentTimeMillis();
    int chunksToFetch = Math.min(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);
  }
{@inheritDoc}