kudu-protolib/src/main/java/com/streamsets/pipeline/stage/processor/kudulookup/KuduLookupProcessor.java
@Override
  protected void process(Record record, SingleLaneProcessor.SingleLaneBatchMaker batchMaker) throws StageException {
    RecordEL.setRecordInContext(tableNameVars, record);
    String tableName = tableNameEval.eval(tableNameVars, conf.kuduTableTemplate, String.class);
    if (!conf.caseSensitive) {
      tableName = tableName.toLowerCase();
    }
    LOG.trace("Processing record:{}  TableName={}", record.toString(), tableName);

    try {
      try {
        KuduLookupKey key = generateLookupKey(record, tableName);
        List<Map<String, Field>> values = cache.get(key);
        if (values.isEmpty()) {
          // No record found
          if (conf.missingLookupBehavior == MissingValuesBehavior.SEND_TO_ERROR) {
            errorRecordHandler.onError(new OnRecordErrorException(record, Errors.KUDU_31));
          } else {
            // Configured to 'Send to next stage' and 'pass as it is'
            batchMaker.addRecord(record);
          }
        } else {
          switch (conf.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;
            default:
              throw new IllegalStateException("Unknown multiple value behavior: " + conf.multipleValuesBehavior);
          }
        }
      } catch (ExecutionException e) {
        Throwables.propagateIfPossible(e.getCause(), StageException.class);
        Throwables.propagateIfPossible(e.getCause(), OnRecordErrorException.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, SingleLaneProcessor.SingleLaneBatchMaker batchMaker) throws StageException {
    RecordEL.setRecordInContext(tableNameVars, record);
    String tableName = tableNameEval.eval(tableNameVars, conf.kuduTableTemplate, String.class);
    if (!conf.caseSensitive) {
      tableName = tableName.toLowerCase();
    }
    LOG.trace("Processing record:{}  TableName={}", record.toString(), tableName);

    try {
      try {
        KuduLookupKey key = generateLookupKey(record, tableName);
        List<Map<String, Field>> values = cache.get(key);
        if (values.isEmpty()) {
          // No record found
          if (conf.missingLookupBehavior == MissingValuesBehavior.SEND_TO_ERROR) {
            errorRecordHandler.onError(new OnRecordErrorException(record, Errors.KUDU_31));
          } else {
            // Configured to 'Send to next stage' and 'pass as it is'
            batchMaker.addRecord(record);
          }
        } else {
          switch (conf.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;
            default:
              throw new IllegalStateException("Unknown multiple value behavior: " + conf.multipleValuesBehavior);
          }
        }
      } catch (ExecutionException e) {
        Throwables.propagateIfPossible(e.getCause(), StageException.class);
        Throwables.propagateIfPossible(e.getCause(), OnRecordErrorException.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}