couchbase-protolib/src/main/java/com/streamsets/pipeline/stage/processor/couchbase/CouchbaseProcessor.java
private Observable<Record> setN1QLRowInRecord(Record record, AsyncN1qlQueryRow row) {

    for(N1QLMappingConfig n1qlMapping : config.n1qlMappingConfigs) {
      if (config.multipleValueOperation == MultipleValueType.FIRST && record.get(n1qlMapping.sdcField) != null) {
        LOG.debug("Only populating output field with first record. Skipping additional result.");
        return Observable.empty();
      }

      Object property = row.value().get(n1qlMapping.property);

      if (property == null) {
        LOG.debug("Requested property not returned: {}", n1qlMapping.property);
        return handleError(record, Errors.COUCHBASE_27, true);
      }

      try {
        record.set(n1qlMapping.sdcField, jsonToField(property));
      } catch (IOException e) {
        try {
          record.set(n1qlMapping.sdcField, jsonToField(JsonObject.fromJson(property.toString()).toMap()));
        } catch (IOException ex) {
          LOG.debug("Unable to set N1QL property in record");
          return handleError(record, Errors.COUCHBASE_19, ex, false);
        }
      }
    }
    return Observable.just(record);
  }
private Observable<Record> setN1QLRowInRecord(Record record, AsyncN1qlQueryRow row) {

    for(N1QLMappingConfig n1qlMapping : config.n1qlMappingConfigs) {
      if (config.multipleValueOperation == MultipleValueType.FIRST && record.get(n1qlMapping.sdcField) != null) {
        LOG.debug("Only populating output field with first record. Skipping additional result.");
        return Observable.empty();
      }

      Object property = row.value().get(n1qlMapping.property);

      if (property == null) {
        LOG.debug("Requested property not returned: {}", n1qlMapping.property);
        return handleError(record, Errors.COUCHBASE_27, true);
      }

      try {
        record.set(n1qlMapping.sdcField, jsonToField(property));
      } catch (IOException e) {
        try {
          record.set(n1qlMapping.sdcField, jsonToField(JsonObject.fromJson(property.toString()).toMap()));
        } catch (IOException ex) {
          LOG.debug("Unable to set N1QL property in record");
          return handleError(record, Errors.COUCHBASE_19, ex, false);
        }
      }
    }
    return Observable.just(record);
  }
private Observable < Record > setN1QLRowInRecord ( Record record , AsyncN1qlQueryRow row ) { for ( N1QLMappingConfig n1qlMapping : config . n1qlMappingConfigs ) { if ( config . multipleValueOperation == MultipleValueType . FIRST && record . get ( n1qlMapping . sdcField ) != null ) { LOG . debug ( "Only populating output field with first record. Skipping additional result." ) ; return Observable . empty ( ) ; } Object property = row . value ( ) . get ( n1qlMapping . property ) ; if ( property == null ) { LOG . debug ( "Requested property not returned: {}" , n1qlMapping . property ) ; return handleError ( record , Errors . COUCHBASE_27 , true ) ; } try { record . set ( n1qlMapping . sdcField , jsonToField ( property ) ) ; } catch ( IOException e ) { try { record . set ( n1qlMapping . sdcField , jsonToField ( JsonObject . fromJson ( property . toString ( ) ) . toMap ( ) ) ) ; } catch ( IOException ex ) { LOG . debug ( "Unable to set N1QL property in record" ) ; return handleError ( record , Errors . COUCHBASE_19 , ex , false ) ; } } } return Observable . just ( record ) ; }
Writes N1QL query result rows to the record

@param record the record being processed
@param row the current row from the N1QL query result
@return an Observable of the updated record
Writes N1QL query result rows to the record