jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/JdbcOracleCDCRecordReader.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 oracle.cdc.operation in record header.
      if (StringUtils.isBlank(op)) {
          op = record.getHeader().getAttribute(OracleCDCOperationCode.OPERATION);
          if (op != null) {
            // Convert the Oracle specific operation code to SDC standard operation code
            opCode = OracleCDCOperationCode.convertFromOracleToSDCCode(op);
          }
      } else {
        opCode = JDBCOperationType.convertToIntCode(op);
      }
      if (opCode == -1){
        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 oracle.cdc.operation in record header.
      if (StringUtils.isBlank(op)) {
          op = record.getHeader().getAttribute(OracleCDCOperationCode.OPERATION);
          if (op != null) {
            // Convert the Oracle specific operation code to SDC standard operation code
            opCode = OracleCDCOperationCode.convertFromOracleToSDCCode(op);
          }
      } else {
        opCode = JDBCOperationType.convertToIntCode(op);
      }
      if (opCode == -1){
        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 oracle.cdc.operation in record header. if ( StringUtils . isBlank ( op ) ) { op = record . getHeader ( ) . getAttribute ( OracleCDCOperationCode . OPERATION ) ; if ( op != null ) { // Convert the Oracle specific operation code to SDC standard operation code opCode = OracleCDCOperationCode . convertFromOracleToSDCCode ( op ) ; } } else { opCode = JDBCOperationType . convertToIntCode ( op ) ; } if ( opCode == - 1 ) { 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
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 .