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();
    }
  }
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
Executes a sub - document write operation .