couchbase-protolib/src/main/java/com/streamsets/pipeline/stage/processor/couchbase/CouchbaseProcessor.java
private Observable<Record> setFragmentInRecord(Record record, DocumentFragment<Lookup> frag) {
    if(frag.content(0) == null) {
      LOG.debug("Sub-document path not found");
      return handleError(record, Errors.COUCHBASE_25, true);
    }

    for(SubdocMappingConfig subdocMapping : config.subdocMappingConfigs) {
     Object fragJson = frag.content(subdocMapping.subdocPath);

      if(fragJson == null) {
        return handleError(record, Errors.COUCHBASE_25, true);
      }

      try {
        record.set(subdocMapping.sdcField, jsonToField(fragJson));
        record.getHeader().setAttribute(config.CAS_HEADER_ATTRIBUTE, String.valueOf(frag.cas()));
      } catch (IOException e) {
        try {
          record.set(subdocMapping.sdcField, jsonToField(JsonObject.fromJson(fragJson.toString()).toMap()));
          record.getHeader().setAttribute(config.CAS_HEADER_ATTRIBUTE, String.valueOf(frag.cas()));
        } catch (IOException ex) {
          return handleError(record, Errors.COUCHBASE_19, ex, false);
        }
      }
    }

    return Observable.just(record);
  }
private Observable<Record> setFragmentInRecord(Record record, DocumentFragment<Lookup> frag) {
    if(frag.content(0) == null) {
      LOG.debug("Sub-document path not found");
      return handleError(record, Errors.COUCHBASE_25, true);
    }

    for(SubdocMappingConfig subdocMapping : config.subdocMappingConfigs) {
     Object fragJson = frag.content(subdocMapping.subdocPath);

      if(fragJson == null) {
        return handleError(record, Errors.COUCHBASE_25, true);
      }

      try {
        record.set(subdocMapping.sdcField, jsonToField(fragJson));
        record.getHeader().setAttribute(config.CAS_HEADER_ATTRIBUTE, String.valueOf(frag.cas()));
      } catch (IOException e) {
        try {
          record.set(subdocMapping.sdcField, jsonToField(JsonObject.fromJson(fragJson.toString()).toMap()));
          record.getHeader().setAttribute(config.CAS_HEADER_ATTRIBUTE, String.valueOf(frag.cas()));
        } catch (IOException ex) {
          return handleError(record, Errors.COUCHBASE_19, ex, false);
        }
      }
    }

    return Observable.just(record);
  }
Writes sub-document lookup values to the record

@param record the record being processed
@param frag the sub-document fragment from Couchbase
@return an Observable of the updated record