jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/JdbcMicrosoftRecordReader.java
@Override
  @VisibleForTesting
  int getOperationFromRecord(
      Record record,
      int defaultOpCode,
      UnsupportedOperationAction unsupportedAction,
      List<OnRecordErrorException> errorRecords ) {

    int opCode = -1; // -1 is invalid and not used in OperationType.
    String op = null;
    try {
      // Try sdc.operation.type first
      op = record.getHeader().getAttribute(OperationType.SDC_OPERATION_TYPE);
      // If not set, look for "__$operation" in record.
      if (StringUtils.isBlank(op)) {
        if (record.has(MSOperationCode.getOpField())) {
          int intOp = record.get(MSOperationCode.getOpField()).getValueAsInteger();
          // Convert the MS specific operation code to SDC standard operation code
          opCode = MSOperationCode.convertToJDBCCode(intOp);
        }
      } else {
        opCode = JDBCOperationType.convertToIntCode(op);
      }
      if (opCode == -1) { // Both MS code and sdc code are not set. Use default.
        opCode = defaultOpCode;
      }
    } catch (NumberFormatException | UnsupportedOperationException ex) {
      LOG.debug(
          "Operation obtained from record is not supported: {}. Handle by UnsupportedOpertaionAction {}. {}",
          ex.getMessage(),
          unsupportedAction.getLabel(),
          ex
      );
      switch (unsupportedAction) {
        case DISCARD:
          LOG.debug("Discarding record with unsupported operation {}", op);
          break;
        case SEND_TO_ERROR:
          LOG.debug("Sending record to error due to unsupported operation {}", op);
          errorRecords.add(new OnRecordErrorException(record, JdbcErrors.JDBC_70, op));
          break;
        case USE_DEFAULT:
          opCode = defaultOpCode;
          break;
        default: //unknown action
          LOG.debug("Sending record to error due to unknown operation: {}", op);
      }
    }
    return opCode;
  }
@Override
  @VisibleForTesting
  int getOperationFromRecord(
      Record record,
      int defaultOpCode,
      UnsupportedOperationAction unsupportedAction,
      List<OnRecordErrorException> errorRecords ) {

    int opCode = -1; // -1 is invalid and not used in OperationType.
    String op = null;
    try {
      // Try sdc.operation.type first
      op = record.getHeader().getAttribute(OperationType.SDC_OPERATION_TYPE);
      // If not set, look for "__$operation" in record.
      if (StringUtils.isBlank(op)) {
        if (record.has(MSOperationCode.getOpField())) {
          int intOp = record.get(MSOperationCode.getOpField()).getValueAsInteger();
          // Convert the MS specific operation code to SDC standard operation code
          opCode = MSOperationCode.convertToJDBCCode(intOp);
        }
      } else {
        opCode = JDBCOperationType.convertToIntCode(op);
      }
      if (opCode == -1) { // Both MS code and sdc code are not set. Use default.
        opCode = defaultOpCode;
      }
    } catch (NumberFormatException | UnsupportedOperationException ex) {
      LOG.debug(
          "Operation obtained from record is not supported: {}. Handle by UnsupportedOpertaionAction {}. {}",
          ex.getMessage(),
          unsupportedAction.getLabel(),
          ex
      );
      switch (unsupportedAction) {
        case DISCARD:
          LOG.debug("Discarding record with unsupported operation {}", op);
          break;
        case SEND_TO_ERROR:
          LOG.debug("Sending record to error due to unsupported operation {}", op);
          errorRecords.add(new OnRecordErrorException(record, JdbcErrors.JDBC_70, op));
          break;
        case USE_DEFAULT:
          opCode = defaultOpCode;
          break;
        default: //unknown action
          LOG.debug("Sending record to error due to unknown operation: {}", op);
      }
    }
    return opCode;
  }
Get an operation code from record.
First, look for sdc.operation.code from record header.
If not set, look for "__$operation" in record. It is a specific field that MS SQL CDC origin set.

If either of them is set, check if the value is supported operation.
If not supported, let it handle by UnsupportedAction
check if the value is supported.
@param record
@return
@throws StageException