couchbase-protolib/src/main/java/com/streamsets/pipeline/stage/destination/couchbase/CouchbaseTarget.java
private Observable<DocumentFragment<Mutation>> writeSubdoc(String key, int ttl, long cas, ByteArrayOutputStream baos,
      String subdocPath, Record record) {

    JsonObject frag;

    try {
      frag = JsonObject.fromJson(baos.toString(config.dataFormatConfig.charset));
    }
    catch (IOException e) {
      return handleError(record, Errors.COUCHBASE_12, e);
    }

    AsyncMutateInBuilder mutation = connector.bucket().mutateIn(key);
    SubdocOptionsBuilder options = new SubdocOptionsBuilder().createPath(true);
    String subdocOpType;

    try {
      subdocOpType = subdocOperationELEval.eval(elVars, config.subdocOperationEL, String.class);
    } catch (ELEvalException e) {
      return handleError(record, Errors.COUCHBASE_13, e);
    }

    if(config.allowSubdoc && !subdocOpType.isEmpty()) {
      switch (subdocOpType.toUpperCase()) {
        case "DELETE":
          LOG.debug("Sub-document DELETE key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.remove(subdocPath, options), ttl, cas, false);

        case "INSERT":
          LOG.debug("Sub-document INSERT key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.insert(subdocPath, frag, options), ttl, cas, true);

        case "REPLACE":
          LOG.debug("Sub-document REPLACE key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.replace(subdocPath, frag), ttl, cas,false);

        case "UPSERT":
          LOG.debug("Sub-document UPSERT key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.upsert(subdocPath, frag, options), ttl, cas,true);

        case "ARRAY_PREPEND":
          LOG.debug("Sub-document ARRAY_PREPEND key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.arrayPrepend(subdocPath, frag, options), ttl, cas,true);

        case "ARRAY_APPEND":
          LOG.debug("Sub-document ARRAY_APPEND key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.arrayAppend(subdocPath, frag, options), ttl, cas, true);

        case "ARRAY_ADD_UNIQUE":
          LOG.debug("Sub-document ARRAY_ADD_UNIQUE key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.arrayAddUnique(subdocPath, frag, options), ttl, cas, true);

        default:
          switch (config.unsupportedOperation) {
            case DISCARD:
              LOG.debug("Unsupported sub-document operation: {} for key: {}. Discarding record per configuration.", subdocOpType, key);
              return Observable.empty();
            case TOERROR:
              LOG.debug("Unsupported sub-document operation: {} for key: {}. Sending record to error per configuration.", subdocOpType, key);
              return handleError(record, Errors.COUCHBASE_14, new RuntimeException());
            default:
              LOG.debug("Unsupported sub-document operation: {} for key: {}. Using default write operation per configuration.", subdocOpType, key);
              // Fall through
              // Inherit the CDC or default operation
          }
      }
    }

    WriteOperationType opType = getOperationFromHeader(record, key);

    if(opType == null) {
      return Observable.empty();
    }

    switch (opType) {
      case DELETE:
        LOG.debug("Sub-document DELETE key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
        return buildSubdocMutation(mutation.remove(subdocPath, options), ttl, cas,false);

      case INSERT:
        LOG.debug("Sub-document INSERT key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
        return buildSubdocMutation(mutation.insert(subdocPath, frag, options), ttl, cas, true);

      case REPLACE:
        LOG.debug("Sub-document REPLACE key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
        return buildSubdocMutation(mutation.replace(subdocPath, frag), ttl, cas, false);

      case UPSERT:
        LOG.debug("Sub-document UPSERT key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
        return buildSubdocMutation(mutation.upsert(subdocPath, frag, options), ttl, cas, true);

      default:
        return Observable.empty();
    }
  }
private Observable<DocumentFragment<Mutation>> writeSubdoc(String key, int ttl, long cas, ByteArrayOutputStream baos,
      String subdocPath, Record record) {

    JsonObject frag;

    try {
      frag = JsonObject.fromJson(baos.toString(config.dataFormatConfig.charset));
    }
    catch (IOException e) {
      return handleError(record, Errors.COUCHBASE_12, e);
    }

    AsyncMutateInBuilder mutation = connector.bucket().mutateIn(key);
    SubdocOptionsBuilder options = new SubdocOptionsBuilder().createPath(true);
    String subdocOpType;

    try {
      subdocOpType = subdocOperationELEval.eval(elVars, config.subdocOperationEL, String.class);
    } catch (ELEvalException e) {
      return handleError(record, Errors.COUCHBASE_13, e);
    }

    if(config.allowSubdoc && !subdocOpType.isEmpty()) {
      switch (subdocOpType.toUpperCase()) {
        case "DELETE":
          LOG.debug("Sub-document DELETE key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.remove(subdocPath, options), ttl, cas, false);

        case "INSERT":
          LOG.debug("Sub-document INSERT key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.insert(subdocPath, frag, options), ttl, cas, true);

        case "REPLACE":
          LOG.debug("Sub-document REPLACE key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.replace(subdocPath, frag), ttl, cas,false);

        case "UPSERT":
          LOG.debug("Sub-document UPSERT key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.upsert(subdocPath, frag, options), ttl, cas,true);

        case "ARRAY_PREPEND":
          LOG.debug("Sub-document ARRAY_PREPEND key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.arrayPrepend(subdocPath, frag, options), ttl, cas,true);

        case "ARRAY_APPEND":
          LOG.debug("Sub-document ARRAY_APPEND key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.arrayAppend(subdocPath, frag, options), ttl, cas, true);

        case "ARRAY_ADD_UNIQUE":
          LOG.debug("Sub-document ARRAY_ADD_UNIQUE key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
          return buildSubdocMutation(mutation.arrayAddUnique(subdocPath, frag, options), ttl, cas, true);

        default:
          switch (config.unsupportedOperation) {
            case DISCARD:
              LOG.debug("Unsupported sub-document operation: {} for key: {}. Discarding record per configuration.", subdocOpType, key);
              return Observable.empty();
            case TOERROR:
              LOG.debug("Unsupported sub-document operation: {} for key: {}. Sending record to error per configuration.", subdocOpType, key);
              return handleError(record, Errors.COUCHBASE_14, new RuntimeException());
            default:
              LOG.debug("Unsupported sub-document operation: {} for key: {}. Using default write operation per configuration.", subdocOpType, key);
              // Fall through
              // Inherit the CDC or default operation
          }
      }
    }

    WriteOperationType opType = getOperationFromHeader(record, key);

    if(opType == null) {
      return Observable.empty();
    }

    switch (opType) {
      case DELETE:
        LOG.debug("Sub-document DELETE key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
        return buildSubdocMutation(mutation.remove(subdocPath, options), ttl, cas,false);

      case INSERT:
        LOG.debug("Sub-document INSERT key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
        return buildSubdocMutation(mutation.insert(subdocPath, frag, options), ttl, cas, true);

      case REPLACE:
        LOG.debug("Sub-document REPLACE key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
        return buildSubdocMutation(mutation.replace(subdocPath, frag), ttl, cas, false);

      case UPSERT:
        LOG.debug("Sub-document UPSERT key: {}, sub-document path: {}, TTL: {}, CAS: {}", key, subdocPath, ttl, cas);
        return buildSubdocMutation(mutation.upsert(subdocPath, frag, options), ttl, cas, true);

      default:
        return Observable.empty();
    }
  }
Executes a sub-document write operation.

@param key the document key
@param ttl the document expiry ttl
@param cas the compare-and-swap (CAS) value to apply to the write
@param baos the raw document content
@param subdocPath the path within the parent document in which to write the subdoc
@param record the record being written
@return an observable for the subdoc mutation or an empty observable on error