redis-lib/src/main/java/com/streamsets/pipeline/stage/origin/redis/RedisSubscriptionSource.java
@Override
  public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) throws StageException {
    // Offsets can vary depending on the data source. Here we use an integer as an example only.
    long nextSourceOffset = 0;
    lastSourceOffset = lastSourceOffset == null ? "" : lastSourceOffset;
    if (!lastSourceOffset.equals("")) {
      nextSourceOffset = Long.parseLong(lastSourceOffset);
    }

    int recordCounter = 0;
    long startTime = System.currentTimeMillis();
    int maxRecords = Math.min(maxBatchSize, conf.maxBatchSize);

    while (recordCounter < maxRecords && (startTime + conf.maxWaitTime) > System.currentTimeMillis()) {
      String message = buffer.poll();
      if (null == message) {
        try {
          Thread.sleep(100);
        } catch (Exception e) {
          LOG.debug(e.getMessage(), e);
          break;
        }
      } else {
        List<Record> records = processRedisMessage("id::" + nextSourceOffset, message);
        for (Record record : records) {
          batchMaker.addRecord(record);
        }
        recordCounter += records.size();
        ++nextSourceOffset;
      }
    }
    return lastSourceOffset;
  }
@Override
  public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) throws StageException {
    // Offsets can vary depending on the data source. Here we use an integer as an example only.
    long nextSourceOffset = 0;
    lastSourceOffset = lastSourceOffset == null ? "" : lastSourceOffset;
    if (!lastSourceOffset.equals("")) {
      nextSourceOffset = Long.parseLong(lastSourceOffset);
    }

    int recordCounter = 0;
    long startTime = System.currentTimeMillis();
    int maxRecords = Math.min(maxBatchSize, conf.maxBatchSize);

    while (recordCounter < maxRecords && (startTime + conf.maxWaitTime) > System.currentTimeMillis()) {
      String message = buffer.poll();
      if (null == message) {
        try {
          Thread.sleep(100);
        } catch (Exception e) {
          LOG.debug(e.getMessage(), e);
          break;
        }
      } else {
        List<Record> records = processRedisMessage("id::" + nextSourceOffset, message);
        for (Record record : records) {
          batchMaker.addRecord(record);
        }
        recordCounter += records.size();
        ++nextSourceOffset;
      }
    }
    return lastSourceOffset;
  }
{@inheritDoc}