couchbase-protolib/src/main/java/com/streamsets/pipeline/stage/destination/couchbase/CouchbaseTarget.java
private WriteOperationType getOperationFromHeader(Record record, String key) {

    String op = record.getHeader().getAttribute(OperationType.SDC_OPERATION_TYPE);
    if (op == null || op.isEmpty()) {
      return config.defaultWriteOperation;
    }

    int opCode;

    try {
      opCode = Integer.parseInt(op);
    } catch(NumberFormatException e) {
        LOG.debug("Unparsable CDC operation. Sending record to error.");
        handleError(record, Errors.COUCHBASE_08, e);
        return null;
      }

    switch (opCode) {
      case OperationType.INSERT_CODE:
        return WriteOperationType.INSERT;
      case OperationType.UPDATE_CODE:
        return WriteOperationType.REPLACE;
      case OperationType.UPSERT_CODE:
        return WriteOperationType.UPSERT;
      case OperationType.DELETE_CODE:
        return WriteOperationType.DELETE;
      default:
        switch (config.unsupportedOperation) {
          case DISCARD:
            LOG.debug("Unsupported CDC operation for key: {}. Discarding record per configuration.", key);
            return null;
          case TOERROR:
            LOG.debug("Unsupported CDC operation for key: {}. Sending record to error configuration.", key);
            handleError(record, Errors.COUCHBASE_09, new RuntimeException());
            return null;
          default:
            LOG.debug("Unsupported CDC operation for key: {}. Using default write operation per configuration.", key);
            return config.defaultWriteOperation;
        }
    }
  }
private WriteOperationType getOperationFromHeader(Record record, String key) {

    String op = record.getHeader().getAttribute(OperationType.SDC_OPERATION_TYPE);
    if (op == null || op.isEmpty()) {
      return config.defaultWriteOperation;
    }

    int opCode;

    try {
      opCode = Integer.parseInt(op);
    } catch(NumberFormatException e) {
        LOG.debug("Unparsable CDC operation. Sending record to error.");
        handleError(record, Errors.COUCHBASE_08, e);
        return null;
      }

    switch (opCode) {
      case OperationType.INSERT_CODE:
        return WriteOperationType.INSERT;
      case OperationType.UPDATE_CODE:
        return WriteOperationType.REPLACE;
      case OperationType.UPSERT_CODE:
        return WriteOperationType.UPSERT;
      case OperationType.DELETE_CODE:
        return WriteOperationType.DELETE;
      default:
        switch (config.unsupportedOperation) {
          case DISCARD:
            LOG.debug("Unsupported CDC operation for key: {}. Discarding record per configuration.", key);
            return null;
          case TOERROR:
            LOG.debug("Unsupported CDC operation for key: {}. Sending record to error configuration.", key);
            handleError(record, Errors.COUCHBASE_09, new RuntimeException());
            return null;
          default:
            LOG.debug("Unsupported CDC operation for key: {}. Using default write operation per configuration.", key);
            return config.defaultWriteOperation;
        }
    }
  }
private WriteOperationType getOperationFromHeader ( Record record , String key ) { String op = record . getHeader ( ) . getAttribute ( OperationType . SDC_OPERATION_TYPE ) ; if ( op == null || op . isEmpty ( ) ) { return config . defaultWriteOperation ; } int opCode ; try { opCode = Integer . parseInt ( op ) ; } catch ( NumberFormatException e ) { LOG . debug ( "Unparsable CDC operation. Sending record to error." ) ; handleError ( record , Errors . COUCHBASE_08 , e ) ; return null ; } switch ( opCode ) { case OperationType . INSERT_CODE : return WriteOperationType . INSERT ; case OperationType . UPDATE_CODE : return WriteOperationType . REPLACE ; case OperationType . UPSERT_CODE : return WriteOperationType . UPSERT ; case OperationType . DELETE_CODE : return WriteOperationType . DELETE ; default : switch ( config . unsupportedOperation ) { case DISCARD : LOG . debug ( "Unsupported CDC operation for key: {}. Discarding record per configuration." , key ) ; return null ; case TOERROR : LOG . debug ( "Unsupported CDC operation for key: {}. Sending record to error configuration." , key ) ; handleError ( record , Errors . COUCHBASE_09 , new RuntimeException ( ) ) ; return null ; default : LOG . debug ( "Unsupported CDC operation for key: {}. Using default write operation per configuration." , key ) ; return config . defaultWriteOperation ; } } }
Evaluates the sdc.operation.type header for a record and returns the equivalent Couchbase write operation type.

@param record the record being evaluated
@param key the document key, used for logging
@return the equivalent WriteOperationType or the default write operation if sdc.operation.type is not configured.
Evaluates the sdc . operation . type header for a record and returns the equivalent Couchbase write operation type .