jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/multithread/CTJdbcRunnable.java
@Override
  public void createAndAddRecord(
      ResultSet rs,
      TableRuntimeContext tableRuntimeContext,
      BatchContext batchContext
  ) throws SQLException, StageException {
    ResultSetMetaData md = rs.getMetaData();

    LinkedHashMap<String, Field> fields = jdbcUtil.resultSetToFields(
        rs,
        commonSourceConfigBean,
        errorRecordHandler,
        tableJdbcConfigBean.unknownTypeAction,
        recordHeader,
        DatabaseVendor.SQL_SERVER
    );

    Map<String, String> columnOffsets = new HashMap<>();


    // Generate Offset includes primary keys, sys_change_version, and sys_change_operation
    for (String key : tableRuntimeContext.getSourceTableContext().getOffsetColumns()) {
      String value = rs.getString(key);
      if (Strings.isNullOrEmpty(value)) {
        value = fields.get(key) != null ? fields.get(key).getValueAsString() : "";
      }
      columnOffsets.put(key, value);
    }


    columnOffsets.put(SYS_CHANGE_OPERATION, rs.getString(SYS_CHANGE_OPERATION));

    String offsetFormat = OffsetQueryUtil.getOffsetFormat(columnOffsets);

    Record record = context.createRecord(tableRuntimeContext.getQualifiedName() + "::" + offsetFormat);
    record.set(Field.createListMap(fields));

    //Set Column Headers
    jdbcUtil.setColumnSpecificHeaders(
        record,
        Collections.singleton(tableRuntimeContext.getSourceTableContext().getTableName()),
        md,
        JDBC_NAMESPACE_HEADER
    );

    //Set Operation Headers
    int op = MSOperationCode.convertToJDBCCode(rs.getString(SYS_CHANGE_OPERATION));
    record.getHeader().setAttribute(OperationType.SDC_OPERATION_TYPE, String.valueOf(op));

    for (String fieldName : recordHeader) {
      record.getHeader().setAttribute(JDBC_NAMESPACE_HEADER + fieldName, rs.getString(fieldName) != null ? rs.getString(fieldName) : "NULL" );
    }

    int columns = rs.getMetaData().getColumnCount();
    if (fields.size() != columns) {
      errorRecordHandler.onError(JdbcErrors.JDBC_35, fields.size(), columns);
      return; // Don't output this record.
    } else {
      batchContext.getBatchMaker().addRecord(record);
    }

    offsets.put(tableRuntimeContext.getOffsetKey(), offsetFormat);
  }
@Override
  public void createAndAddRecord(
      ResultSet rs,
      TableRuntimeContext tableRuntimeContext,
      BatchContext batchContext
  ) throws SQLException, StageException {
    ResultSetMetaData md = rs.getMetaData();

    LinkedHashMap<String, Field> fields = jdbcUtil.resultSetToFields(
        rs,
        commonSourceConfigBean,
        errorRecordHandler,
        tableJdbcConfigBean.unknownTypeAction,
        recordHeader,
        DatabaseVendor.SQL_SERVER
    );

    Map<String, String> columnOffsets = new HashMap<>();


    // Generate Offset includes primary keys, sys_change_version, and sys_change_operation
    for (String key : tableRuntimeContext.getSourceTableContext().getOffsetColumns()) {
      String value = rs.getString(key);
      if (Strings.isNullOrEmpty(value)) {
        value = fields.get(key) != null ? fields.get(key).getValueAsString() : "";
      }
      columnOffsets.put(key, value);
    }


    columnOffsets.put(SYS_CHANGE_OPERATION, rs.getString(SYS_CHANGE_OPERATION));

    String offsetFormat = OffsetQueryUtil.getOffsetFormat(columnOffsets);

    Record record = context.createRecord(tableRuntimeContext.getQualifiedName() + "::" + offsetFormat);
    record.set(Field.createListMap(fields));

    //Set Column Headers
    jdbcUtil.setColumnSpecificHeaders(
        record,
        Collections.singleton(tableRuntimeContext.getSourceTableContext().getTableName()),
        md,
        JDBC_NAMESPACE_HEADER
    );

    //Set Operation Headers
    int op = MSOperationCode.convertToJDBCCode(rs.getString(SYS_CHANGE_OPERATION));
    record.getHeader().setAttribute(OperationType.SDC_OPERATION_TYPE, String.valueOf(op));

    for (String fieldName : recordHeader) {
      record.getHeader().setAttribute(JDBC_NAMESPACE_HEADER + fieldName, rs.getString(fieldName) != null ? rs.getString(fieldName) : "NULL" );
    }

    int columns = rs.getMetaData().getColumnCount();
    if (fields.size() != columns) {
      errorRecordHandler.onError(JdbcErrors.JDBC_35, fields.size(), columns);
      return; // Don't output this record.
    } else {
      batchContext.getBatchMaker().addRecord(record);
    }

    offsets.put(tableRuntimeContext.getOffsetKey(), offsetFormat);
  }
@ Override public void createAndAddRecord ( ResultSet rs , TableRuntimeContext tableRuntimeContext , BatchContext batchContext ) throws SQLException , StageException { ResultSetMetaData md = rs . getMetaData ( ) ; LinkedHashMap < String , Field > fields = jdbcUtil . resultSetToFields ( rs , commonSourceConfigBean , errorRecordHandler , tableJdbcConfigBean . unknownTypeAction , recordHeader , DatabaseVendor . SQL_SERVER ) ; Map < String , String > columnOffsets = new HashMap <> ( ) ; // Generate Offset includes primary keys, sys_change_version, and sys_change_operation for ( String key : tableRuntimeContext . getSourceTableContext ( ) . getOffsetColumns ( ) ) { String value = rs . getString ( key ) ; if ( Strings . isNullOrEmpty ( value ) ) { value = fields . get ( key ) != null ? fields . get ( key ) . getValueAsString ( ) : "" ; } columnOffsets . put ( key , value ) ; } columnOffsets . put ( SYS_CHANGE_OPERATION , rs . getString ( SYS_CHANGE_OPERATION ) ) ; String offsetFormat = OffsetQueryUtil . getOffsetFormat ( columnOffsets ) ; Record record = context . createRecord ( tableRuntimeContext . getQualifiedName ( ) + "::" + offsetFormat ) ; record . set ( Field . createListMap ( fields ) ) ; //Set Column Headers jdbcUtil . setColumnSpecificHeaders ( record , Collections . singleton ( tableRuntimeContext . getSourceTableContext ( ) . getTableName ( ) ) , md , JDBC_NAMESPACE_HEADER ) ; //Set Operation Headers int op = MSOperationCode . convertToJDBCCode ( rs . getString ( SYS_CHANGE_OPERATION ) ) ; record . getHeader ( ) . setAttribute ( OperationType . SDC_OPERATION_TYPE , String . valueOf ( op ) ) ; for ( String fieldName : recordHeader ) { record . getHeader ( ) . setAttribute ( JDBC_NAMESPACE_HEADER + fieldName , rs . getString ( fieldName ) != null ? rs . getString ( fieldName ) : "NULL" ) ; } int columns = rs . getMetaData ( ) . getColumnCount ( ) ; if ( fields . size ( ) != columns ) { errorRecordHandler . onError ( JdbcErrors . JDBC_35 , fields . size ( ) , columns ) ; return ; // Don't output this record. } else { batchContext . getBatchMaker ( ) . addRecord ( record ) ; } offsets . put ( tableRuntimeContext . getOffsetKey ( ) , offsetFormat ) ; }
Create record and add it to {@link com.streamsets.pipeline.api.BatchMaker}
Create record and add it to {