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

    WriteOperationType opType = getOperationFromHeader(record, key);

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

    AbstractDocument doc;

    if(config.dataFormat == DataFormat.JSON) {
      try {
         doc = JsonDocument.create(key, ttl, JsonObject.fromJson(baos.toString(config.dataFormatConfig.charset)), cas);
      } catch(Exception e) {
        return handleError(record, Errors.COUCHBASE_10, e);
      }
    } else {
       doc = ByteArrayDocument.create(key, ttl, baos.toByteArray(), cas);
    }

    switch (opType) {
      case DELETE: {
          LOG.debug("DELETE key: {}, TTL: {}, CAS: {}", key, ttl, cas);
          return connector.bucket().remove(doc, config.persistTo, config.replicateTo)
              .timeout(config.couchbase.kvTimeout, TimeUnit.MILLISECONDS);
      }
      case INSERT: {
          LOG.debug("INSERT key: {}, TTL: {}, CAS: {}", key, ttl, cas);
          return connector.bucket().insert(doc, config.persistTo, config.replicateTo)
              .timeout(config.couchbase.kvTimeout, TimeUnit.MILLISECONDS);
      }
      case REPLACE: {
          LOG.debug("REPLACE key: {}, TTL: {}, CAS: {}", key, ttl, cas);
          return connector.bucket().replace(doc, config.persistTo, config.replicateTo)
              .timeout(config.couchbase.kvTimeout, TimeUnit.MILLISECONDS);
      }
      case UPSERT: {
          LOG.debug("UPSERT key: {}, TTL: {}, CAS: {}", key, ttl, cas);
          return connector.bucket().upsert(doc, config.persistTo, config.replicateTo)
              .timeout(config.couchbase.kvTimeout, TimeUnit.MILLISECONDS);
      }
      default:
        return Observable.empty();
    }
  }
private Observable<AbstractDocument> writeDoc(String key, int ttl, long cas, ByteArrayOutputStream baos, Record record) {

    WriteOperationType opType = getOperationFromHeader(record, key);

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

    AbstractDocument doc;

    if(config.dataFormat == DataFormat.JSON) {
      try {
         doc = JsonDocument.create(key, ttl, JsonObject.fromJson(baos.toString(config.dataFormatConfig.charset)), cas);
      } catch(Exception e) {
        return handleError(record, Errors.COUCHBASE_10, e);
      }
    } else {
       doc = ByteArrayDocument.create(key, ttl, baos.toByteArray(), cas);
    }

    switch (opType) {
      case DELETE: {
          LOG.debug("DELETE key: {}, TTL: {}, CAS: {}", key, ttl, cas);
          return connector.bucket().remove(doc, config.persistTo, config.replicateTo)
              .timeout(config.couchbase.kvTimeout, TimeUnit.MILLISECONDS);
      }
      case INSERT: {
          LOG.debug("INSERT key: {}, TTL: {}, CAS: {}", key, ttl, cas);
          return connector.bucket().insert(doc, config.persistTo, config.replicateTo)
              .timeout(config.couchbase.kvTimeout, TimeUnit.MILLISECONDS);
      }
      case REPLACE: {
          LOG.debug("REPLACE key: {}, TTL: {}, CAS: {}", key, ttl, cas);
          return connector.bucket().replace(doc, config.persistTo, config.replicateTo)
              .timeout(config.couchbase.kvTimeout, TimeUnit.MILLISECONDS);
      }
      case UPSERT: {
          LOG.debug("UPSERT key: {}, TTL: {}, CAS: {}", key, ttl, cas);
          return connector.bucket().upsert(doc, config.persistTo, config.replicateTo)
              .timeout(config.couchbase.kvTimeout, TimeUnit.MILLISECONDS);
      }
      default:
        return Observable.empty();
    }
  }
Executes a 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 record the record being written
@return an observable for the document write or an empty observable on error