salesforce-lib/src/main/java/com/streamsets/pipeline/stage/origin/salesforce/ForceSource.java
@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.min(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;
  }
@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.min(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;
  }
{@inheritDoc}