question: private void processUpdateCounts ( int [ ] updateCounts , int commandsInBatch ) { if ( updateCounts . length < commandsInBatch ) { error . emit ( tuples . get ( updateCounts . length + batchStartIdx ) ) ; errorTuples ++ ; tuplesWrittenSuccessfully += updateCounts . length ; batchStartIdx += updateCounts . length + 1 ; if ( ( tuples . size ( ) - batchStartIdx ) > 0 ) { processBatch ( ) ; } } else { tuplesWrittenSuccessfully = commandsInBatch ; for ( int i = 0 ; i < commandsInBatch ; i ++ ) { if ( updateCounts [ i ] == Statement . EXECUTE_FAILED ) { error . emit ( tuples . get ( i + batchStartIdx ) ) ; errorTuples ++ ; tuplesWrittenSuccessfully -- ; } } } }
target:  Identify which commands in the batch failed and redirect these on the error port. See https://docs.oracle.com/javase/7/docs/api/java/sql/BatchUpdateException.html for more details