jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/JdbcMultiRowRecordWriter.java
private void processQueue(
      LinkedList<Record> queue,
      List<OnRecordErrorException> errorRecords,
      Connection connection,
      int maxRowsPerBatch,
      int opCode
  ) throws StageException {
    if (queue.isEmpty()) {
      return;
    }

    int rowCount = 0;
    // Assume that columns are all same for the same operation to the same table
    // If some columns are missing in record, the record goes to error.
    final Record first = queue.getFirst();
    SortedMap<String, String> columnsToParameters = recordReader.getColumnsToParameters(
        first,
        opCode,
        getColumnsToParameters(),
        opCode == OperationType.UPDATE_CODE ? getColumnsToFieldNoPK() : getColumnsToFields()
    );

    if (columnsToParameters.isEmpty()) {
      // no parameters found for configured columns
      if (LOG.isWarnEnabled()) {
        LOG.warn("No parameters found for record with ID {}; skipping", first.getHeader().getSourceId());
      }
      return;
    }

    String query = generateQueryForMultiRow(
        opCode,
        columnsToParameters,
        getPrimaryKeyColumns(),
        // the next batch will have either the max number of records, or however many are left.
        Math.min(maxRowsPerBatch, queue.size())
    );

    // Need to store removed records from queue, because we might need to add newly generated columns
    // to records for Jdbc Tee Processor.
    LinkedList<Record> removed = new LinkedList<>();

    try (PreparedStatement statement = jdbcUtil.getPreparedStatement(getGeneratedColumnMappings(), query, connection)) {
      int paramIdx = 1;
      // Start processing records in queue. All records have the same operation to the same table.
      while (!queue.isEmpty()) {
        Record r = queue.removeFirst();
        if (opCode != DELETE_CODE) {
          paramIdx = setParamsToStatement(paramIdx, statement, columnsToParameters, r, connection, opCode);
        }
        if (opCode != OperationType.INSERT_CODE) {
          paramIdx = setPrimaryKeys(paramIdx, r, statement, opCode);
        }
        removed.add(r);
        ++rowCount;
        if (rowCount == maxRowsPerBatch) {
          // time to execute the current batch
          processBatch(removed, errorRecords, statement, connection);
          // reset our counters
          rowCount = 0;
          paramIdx = 1;
          removed.clear();
        }
      }
    } catch (SQLException e) {
      handleSqlException(e, removed, errorRecords);
    }

    // Process the rest of the records that are removed from queue but haven't processed yet
    // this happens when rowCount is still less than maxRowsPerBatch.
    // This is a bit of an ugly fix as its not very DRY but sufficient until there's a larger
    // refactoring of this code.
    if (!removed.isEmpty()) {
      query = generateQueryForMultiRow(
          opCode,
          columnsToParameters,
          getPrimaryKeyColumns(),
          removed.size() // always the remainder
      );

      try (PreparedStatement statement = jdbcUtil.getPreparedStatement(
          getGeneratedColumnMappings(),
          query,
          connection
      )) {
        int paramIdx = 1;
        for (Record r : removed) {
          if (opCode != DELETE_CODE) {
            paramIdx = setParamsToStatement(paramIdx, statement, columnsToParameters, r, connection, opCode);
          }
          if (opCode != OperationType.INSERT_CODE) {
            paramIdx = setPrimaryKeys(paramIdx, r, statement, opCode);
          }
        }
        processBatch(removed, errorRecords, statement, connection);
      } catch (SQLException e) {
        handleSqlException(e, removed, errorRecords);
      }
    }
  }
private void processQueue(
      LinkedList<Record> queue,
      List<OnRecordErrorException> errorRecords,
      Connection connection,
      int maxRowsPerBatch,
      int opCode
  ) throws StageException {
    if (queue.isEmpty()) {
      return;
    }

    int rowCount = 0;
    // Assume that columns are all same for the same operation to the same table
    // If some columns are missing in record, the record goes to error.
    final Record first = queue.getFirst();
    SortedMap<String, String> columnsToParameters = recordReader.getColumnsToParameters(
        first,
        opCode,
        getColumnsToParameters(),
        opCode == OperationType.UPDATE_CODE ? getColumnsToFieldNoPK() : getColumnsToFields()
    );

    if (columnsToParameters.isEmpty()) {
      // no parameters found for configured columns
      if (LOG.isWarnEnabled()) {
        LOG.warn("No parameters found for record with ID {}; skipping", first.getHeader().getSourceId());
      }
      return;
    }

    String query = generateQueryForMultiRow(
        opCode,
        columnsToParameters,
        getPrimaryKeyColumns(),
        // the next batch will have either the max number of records, or however many are left.
        Math.min(maxRowsPerBatch, queue.size())
    );

    // Need to store removed records from queue, because we might need to add newly generated columns
    // to records for Jdbc Tee Processor.
    LinkedList<Record> removed = new LinkedList<>();

    try (PreparedStatement statement = jdbcUtil.getPreparedStatement(getGeneratedColumnMappings(), query, connection)) {
      int paramIdx = 1;
      // Start processing records in queue. All records have the same operation to the same table.
      while (!queue.isEmpty()) {
        Record r = queue.removeFirst();
        if (opCode != DELETE_CODE) {
          paramIdx = setParamsToStatement(paramIdx, statement, columnsToParameters, r, connection, opCode);
        }
        if (opCode != OperationType.INSERT_CODE) {
          paramIdx = setPrimaryKeys(paramIdx, r, statement, opCode);
        }
        removed.add(r);
        ++rowCount;
        if (rowCount == maxRowsPerBatch) {
          // time to execute the current batch
          processBatch(removed, errorRecords, statement, connection);
          // reset our counters
          rowCount = 0;
          paramIdx = 1;
          removed.clear();
        }
      }
    } catch (SQLException e) {
      handleSqlException(e, removed, errorRecords);
    }

    // Process the rest of the records that are removed from queue but haven't processed yet
    // this happens when rowCount is still less than maxRowsPerBatch.
    // This is a bit of an ugly fix as its not very DRY but sufficient until there's a larger
    // refactoring of this code.
    if (!removed.isEmpty()) {
      query = generateQueryForMultiRow(
          opCode,
          columnsToParameters,
          getPrimaryKeyColumns(),
          removed.size() // always the remainder
      );

      try (PreparedStatement statement = jdbcUtil.getPreparedStatement(
          getGeneratedColumnMappings(),
          query,
          connection
      )) {
        int paramIdx = 1;
        for (Record r : removed) {
          if (opCode != DELETE_CODE) {
            paramIdx = setParamsToStatement(paramIdx, statement, columnsToParameters, r, connection, opCode);
          }
          if (opCode != OperationType.INSERT_CODE) {
            paramIdx = setPrimaryKeys(paramIdx, r, statement, opCode);
          }
        }
        processBatch(removed, errorRecords, statement, connection);
      } catch (SQLException e) {
        handleSqlException(e, removed, errorRecords);
      }
    }
  }
Process all records in queue. All records have same operation to same table.
Generate a query and set parameters from each record. INSERT and DELETE can be multi-row operation
but UPDATE is single-row operation.
If maxStatement
@param errorRecords
@param connection
@param maxRowsPerBatch
@param opCode
@param queue
@return
@throws OnRecordErrorException