jdbc-protolib/src/main/java/com/streamsets/pipeline/stage/processor/jdbclookup/JdbcLookupProcessor.java
@Override
  protected void process(Record record, SingleLaneBatchMaker batchMaker) throws StageException {
    try {
      ELVars elVars = getContext().createELVars();
      RecordEL.setRecordInContext(elVars, record);
      String preparedQuery = queryEval.eval(elVars, query, String.class);
      Optional<List<Map<String, Field>>> entry = cache.get(preparedQuery);

      if (!entry.isPresent()) {
        // No results
        switch (missingValuesBehavior) {
          case SEND_TO_ERROR:
            LOG.error(JdbcErrors.JDBC_04.getMessage(), preparedQuery);
            errorRecordHandler.onError(new OnRecordErrorException(record, JdbcErrors.JDBC_04, preparedQuery));
            break;
          case PASS_RECORD_ON:
            batchMaker.addRecord(record);
            break;
          default:
            throw new IllegalStateException("Unknown missing value behavior: " + missingValuesBehavior);
        }
      } else {
        List<Map<String, Field>> values = entry.get();
        switch (multipleValuesBehavior) {
          case FIRST_ONLY:
            setFieldsInRecord(record, values.get(0));
            batchMaker.addRecord(record);
            break;
          case SPLIT_INTO_MULTIPLE_RECORDS:
            for(Map<String, Field> lookupItem : values) {
              Record newRecord = getContext().cloneRecord(record);
              setFieldsInRecord(newRecord, lookupItem);
              batchMaker.addRecord(newRecord);
            }
            break;
          case ALL_AS_LIST:
            Map<String, List<Field>> valuesMap = new HashMap<>();
            for (Map<String, Field> lookupItem : values) {
              lookupItem.forEach((k, v) -> {
                if (valuesMap.get(k) == null) {
                  List<Field> lookupValue = new ArrayList<>();
                  valuesMap.put(k, lookupValue);
                }
                valuesMap.get(k).add(v);
              });
            }
            Map<String, Field> valueMap = new HashMap<>();
            valuesMap.forEach( (k,v) -> valueMap.put(k, Field.create(v)));
            setFieldsInRecord(record, valueMap);
            batchMaker.addRecord(record);
            break;
          default:
            throw new IllegalStateException("Unknown multiple value behavior: " + multipleValuesBehavior);
        }

      }

    } catch (ELEvalException e) {
      LOG.error(JdbcErrors.JDBC_01.getMessage(), query, e);
      throw new OnRecordErrorException(record, JdbcErrors.JDBC_01, query);
    } catch (ExecutionException e) {
      Throwables.propagateIfPossible(e.getCause(), StageException.class);
      throw new IllegalStateException(e); // The cache loader shouldn't throw anything that isn't a StageException.
    } catch (OnRecordErrorException error) { // NOSONAR
      errorRecordHandler.onError(new OnRecordErrorException(record, error.getErrorCode(), error.getParams()));
    }
  }
@Override
  protected void process(Record record, SingleLaneBatchMaker batchMaker) throws StageException {
    try {
      ELVars elVars = getContext().createELVars();
      RecordEL.setRecordInContext(elVars, record);
      String preparedQuery = queryEval.eval(elVars, query, String.class);
      Optional<List<Map<String, Field>>> entry = cache.get(preparedQuery);

      if (!entry.isPresent()) {
        // No results
        switch (missingValuesBehavior) {
          case SEND_TO_ERROR:
            LOG.error(JdbcErrors.JDBC_04.getMessage(), preparedQuery);
            errorRecordHandler.onError(new OnRecordErrorException(record, JdbcErrors.JDBC_04, preparedQuery));
            break;
          case PASS_RECORD_ON:
            batchMaker.addRecord(record);
            break;
          default:
            throw new IllegalStateException("Unknown missing value behavior: " + missingValuesBehavior);
        }
      } else {
        List<Map<String, Field>> values = entry.get();
        switch (multipleValuesBehavior) {
          case FIRST_ONLY:
            setFieldsInRecord(record, values.get(0));
            batchMaker.addRecord(record);
            break;
          case SPLIT_INTO_MULTIPLE_RECORDS:
            for(Map<String, Field> lookupItem : values) {
              Record newRecord = getContext().cloneRecord(record);
              setFieldsInRecord(newRecord, lookupItem);
              batchMaker.addRecord(newRecord);
            }
            break;
          case ALL_AS_LIST:
            Map<String, List<Field>> valuesMap = new HashMap<>();
            for (Map<String, Field> lookupItem : values) {
              lookupItem.forEach((k, v) -> {
                if (valuesMap.get(k) == null) {
                  List<Field> lookupValue = new ArrayList<>();
                  valuesMap.put(k, lookupValue);
                }
                valuesMap.get(k).add(v);
              });
            }
            Map<String, Field> valueMap = new HashMap<>();
            valuesMap.forEach( (k,v) -> valueMap.put(k, Field.create(v)));
            setFieldsInRecord(record, valueMap);
            batchMaker.addRecord(record);
            break;
          default:
            throw new IllegalStateException("Unknown multiple value behavior: " + multipleValuesBehavior);
        }

      }

    } catch (ELEvalException e) {
      LOG.error(JdbcErrors.JDBC_01.getMessage(), query, e);
      throw new OnRecordErrorException(record, JdbcErrors.JDBC_01, query);
    } catch (ExecutionException e) {
      Throwables.propagateIfPossible(e.getCause(), StageException.class);
      throw new IllegalStateException(e); // The cache loader shouldn't throw anything that isn't a StageException.
    } catch (OnRecordErrorException error) { // NOSONAR
      errorRecordHandler.onError(new OnRecordErrorException(record, error.getErrorCode(), error.getParams()));
    }
  }
{@inheritDoc}