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);
  }
Create record and add it to {@link com.streamsets.pipeline.api.BatchMaker}