basic-lib/src/main/java/com/streamsets/pipeline/stage/origin/http/HttpClientSource.java
private int parsePaginatedResult(BatchMaker batchMaker, String sourceOffset, Record record) throws
      StageException {
    int numSubRecords = 0;

    if (!record.has(conf.pagination.resultFieldPath)) {
      final StageException stageException = new StageException(Errors.HTTP_12, conf.pagination.resultFieldPath);
      LOG.error(stageException.getMessage());
      throw stageException;
    }
    Field resultField = record.get(conf.pagination.resultFieldPath);

    if (resultField.getType() != Field.Type.LIST) {
      final StageException stageException = new StageException(Errors.HTTP_08, resultField.getType());
      LOG.error(stageException.getMessage());
      throw stageException;
    }

    List<Field> results = resultField.getValueAsList();
    int subRecordIdx = 0;
    for (Field result : results) {
      Record r = getContext().createRecord(sourceOffset + "::" + subRecordIdx++);
      if (conf.pagination.keepAllFields) {
        r.set(record.get().clone());
        r.set(conf.pagination.resultFieldPath, result);
      } else {
        r.set(result);
      }
      addResponseHeaders(r.getHeader());
      batchMaker.addRecord(r);
      ++numSubRecords;
    }
    if (conf.pagination.mode != PaginationMode.LINK_FIELD) {
      haveMorePages = numSubRecords > 0;
    }
    return numSubRecords;
  }
private int parsePaginatedResult(BatchMaker batchMaker, String sourceOffset, Record record) throws
      StageException {
    int numSubRecords = 0;

    if (!record.has(conf.pagination.resultFieldPath)) {
      final StageException stageException = new StageException(Errors.HTTP_12, conf.pagination.resultFieldPath);
      LOG.error(stageException.getMessage());
      throw stageException;
    }
    Field resultField = record.get(conf.pagination.resultFieldPath);

    if (resultField.getType() != Field.Type.LIST) {
      final StageException stageException = new StageException(Errors.HTTP_08, resultField.getType());
      LOG.error(stageException.getMessage());
      throw stageException;
    }

    List<Field> results = resultField.getValueAsList();
    int subRecordIdx = 0;
    for (Field result : results) {
      Record r = getContext().createRecord(sourceOffset + "::" + subRecordIdx++);
      if (conf.pagination.keepAllFields) {
        r.set(record.get().clone());
        r.set(conf.pagination.resultFieldPath, result);
      } else {
        r.set(result);
      }
      addResponseHeaders(r.getHeader());
      batchMaker.addRecord(r);
      ++numSubRecords;
    }
    if (conf.pagination.mode != PaginationMode.LINK_FIELD) {
      haveMorePages = numSubRecords > 0;
    }
    return numSubRecords;
  }
Parses a paginated result from the configured field.

@param batchMaker batch to add records to
@param sourceOffset to use as a base when creating records
@param record the source record containing an array to be converted to records
@return number of records parsed
@throws StageException if an unhandled error is encountered