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