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