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;
        }
    }
  }
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.