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);
  }
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
Writes sub - document lookup values to the record