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()));
    }
  }
@ 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}
{