jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/JdbcUtil.java
public <T> void write(
      Iterator<Record> recordIterator,
      T key,
      LoadingCache<T, JdbcRecordWriter> recordWriters,
      ErrorRecordHandler errorRecordHandler,
      boolean perRecord
  ) throws StageException {
    final JdbcRecordWriter jdbcRecordWriter;
    try {
      jdbcRecordWriter = recordWriters.getUnchecked(key);
    } catch (UncheckedExecutionException ex) {
      final Throwable throwable = ex.getCause();
      final ErrorCode errorCode;
      final Object[] messageParams;
      if (throwable instanceof StageException) {
        StageException stageEx = (StageException) ex.getCause();
        errorCode = stageEx.getErrorCode();
        messageParams = stageEx.getParams();
      } else {
        errorCode = JdbcErrors.JDBC_301;
        messageParams = new Object[] {ex.getMessage(), ex.getCause()};
      }
      // Failed to create RecordWriter, report all as error records.
      while (recordIterator.hasNext()) {
        Record record = recordIterator.next();
        errorRecordHandler.onError(new OnRecordErrorException(record, errorCode, messageParams));
      }
      return;
    }
    List<OnRecordErrorException> errors = perRecord
        ? jdbcRecordWriter.writePerRecord(recordIterator)
        : jdbcRecordWriter.writeBatch(recordIterator);

    for (OnRecordErrorException error : errors) {
      errorRecordHandler.onError(error);
    }
  }
public <T> void write(
      Iterator<Record> recordIterator,
      T key,
      LoadingCache<T, JdbcRecordWriter> recordWriters,
      ErrorRecordHandler errorRecordHandler,
      boolean perRecord
  ) throws StageException {
    final JdbcRecordWriter jdbcRecordWriter;
    try {
      jdbcRecordWriter = recordWriters.getUnchecked(key);
    } catch (UncheckedExecutionException ex) {
      final Throwable throwable = ex.getCause();
      final ErrorCode errorCode;
      final Object[] messageParams;
      if (throwable instanceof StageException) {
        StageException stageEx = (StageException) ex.getCause();
        errorCode = stageEx.getErrorCode();
        messageParams = stageEx.getParams();
      } else {
        errorCode = JdbcErrors.JDBC_301;
        messageParams = new Object[] {ex.getMessage(), ex.getCause()};
      }
      // Failed to create RecordWriter, report all as error records.
      while (recordIterator.hasNext()) {
        Record record = recordIterator.next();
        errorRecordHandler.onError(new OnRecordErrorException(record, errorCode, messageParams));
      }
      return;
    }
    List<OnRecordErrorException> errors = perRecord
        ? jdbcRecordWriter.writePerRecord(recordIterator)
        : jdbcRecordWriter.writeBatch(recordIterator);

    for (OnRecordErrorException error : errors) {
      errorRecordHandler.onError(error);
    }
  }
public < T > void write ( Iterator < Record > recordIterator , T key , LoadingCache < T , JdbcRecordWriter > recordWriters , ErrorRecordHandler errorRecordHandler , boolean perRecord ) throws StageException { final JdbcRecordWriter jdbcRecordWriter ; try { jdbcRecordWriter = recordWriters . getUnchecked ( key ) ; } catch ( UncheckedExecutionException ex ) { final Throwable throwable = ex . getCause ( ) ; final ErrorCode errorCode ; final Object [ ] messageParams ; if ( throwable instanceof StageException ) { StageException stageEx = ( StageException ) ex . getCause ( ) ; errorCode = stageEx . getErrorCode ( ) ; messageParams = stageEx . getParams ( ) ; } else { errorCode = JdbcErrors . JDBC_301 ; messageParams = new Object [ ] { ex . getMessage ( ) , ex . getCause ( ) } ; } // Failed to create RecordWriter, report all as error records. while ( recordIterator . hasNext ( ) ) { Record record = recordIterator . next ( ) ; errorRecordHandler . onError ( new OnRecordErrorException ( record , errorCode , messageParams ) ) ; } return ; } List < OnRecordErrorException > errors = perRecord ? jdbcRecordWriter . writePerRecord ( recordIterator ) : jdbcRecordWriter . writeBatch ( recordIterator ) ; for ( OnRecordErrorException error : errors ) { errorRecordHandler . onError ( error ) ; } }
Write records to a JDBC destination using the recordWriter specified by key, and handle errors

@param recordIterator iterator of SDC records
@param key key to select the recordWriter
@param recordWriters JDBC record writer cache
@param errorRecordHandler error record handler
@param perRecord indicate record or batch update
@throws StageException
Write records to a JDBC destination using the recordWriter specified by key and handle errors