Unable to retrieve View using couchbase-java-client-2.1.2-dp, Observer issue

Hi,

I was upgrading from couchbase-client-1.x.jar to 2.1.2-dp after upgrading the CouchBase Server to 3.0.2.

java-client-2.1.2-dp.jar + core-io-1.1.2-dp.jar + rxjava-1.0.4.jar

I was able to perform simple add / retrieve document action from the Java library (bucket.upsert, bucket.get) but when I was trying to query from a published view I start to get the following error:

java.lang.IllegalStateException: This Observable can only have one subscription. Use Observable.publish() if you want to multicast.
at com.couchbase.client.core.utils.UnicastAutoReleaseSubject$OnSubscribeAction.call(UnicastAutoReleaseSubject.java:227) ~[couchbase-core-io-1.1.2-dp.jar:1.1.1-17-gf6d5dd3-dirty]
at com.couchbase.client.core.utils.UnicastAutoReleaseSubject$OnSubscribeAction.call(UnicastAutoReleaseSubject.java:202) ~[couchbase-core-io-1.1.2-dp.jar:1.1.1-17-gf6d5dd3-dirty]
at rx.Observable$1.call(Observable.java:145) ~[rxjava-1.0.4.jar:1.0.4]
at rx.Observable$1.call(Observable.java:137) ~[rxjava-1.0.4.jar:1.0.4]
at rx.Observable$1.call(Observable.java:145) ~[rxjava-1.0.4.jar:1.0.4]
at rx.Observable$1.call(Observable.java:137) ~[rxjava-1.0.4.jar:1.0.4]
at rx.Observable$1.call(Observable.java:145) ~[rxjava-1.0.4.jar:1.0.4]
at rx.Observable$1.call(Observable.java:137) ~[rxjava-1.0.4.jar:1.0.4]
at rx.Observable$1.call(Observable.java:145) ~[rxjava-1.0.4.jar:1.0.4]
at rx.Observable$1.call(Observable.java:137) ~[rxjava-1.0.4.jar:1.0.4]
at rx.Observable$1.call(Observable.java:145) ~[rxjava-1.0.4.jar:1.0.4]
at rx.Observable$1.call(Observable.java:137) ~[rxjava-1.0.4.jar:1.0.4]
at rx.Observable$1.call(Observable.java:145) ~[rxjava-1.0.4.jar:1.0.4]
at rx.Observable$1.call(Observable.java:137) ~[rxjava-1.0.4.jar:1.0.4]
at rx.Observable.subscribe(Observable.java:7393) ~[rxjava-1.0.4.jar:1.0.4]
at com.couchbase.client.java.util.Blocking.blockForSingle(Blocking.java:73) ~[couchbase-java-client-2.1.2-dp.jar:2.1.1-7-g63629c6-dirty]
at com.couchbase.client.java.view.DefaultViewResult.allRows(DefaultViewResult.java:36) ~[couchbase-java-client-2.1.2-dp.jar:2.1.1-7-g63629c6-dirty]
at com.couchbase.client.java.view.DefaultViewResult.allRows(DefaultViewResult.java:31) ~[couchbase-java-client-2.1.2-dp.jar:2.1.1-7-g63629c6-dirty]

The code is quite simple which I extract the import parts of it:

        CouchbaseEnvironment couchbaseEnvironment = DefaultCouchbaseEnvironment.builder()
                .viewTimeout(5000)
                .queryTimeout(5000)
                .computationPoolSize(4)
                .queryEnabled(false).build();

        cluster = CouchbaseCluster.create(couchbaseEnvironment, serverNodes);
        bucket = cluster.openBucket(bucket_name, password, bucketConnectionTimeout, TimeUnit.SECONDS);            
        ViewQuery query = ViewQuery.from(designDoc, viewName).stale(Stale.FALSE);
        
              // add key for view
              if (key != null)
              {query.key(key);}
              
              // Fetch the View
              ViewResult result = bucket.query(query);
        return result.allRows();  <<=== this throws exception

My goal is to return the List of the document contents from the view result and I also tried “Iterator” or “for (ViewRow row : result)” but they also get me same error.

I was wondering if there is something I missed or I implemented in the wrong way…

Hope there is someone can enlighten me, thanks.

Actually it looks like you are accessing result.allRows() more than once. Can you try to assign it to a variable (the List) and return this?

Hi Daschl,

Nice one, haven’t thought about the iterator being called twice and introduced the error.

Case closed, thanks a lot~

@daschl I’m not seeing where he is accessing result.allRows() more than once. I only see the one call there at the end of his code block.

I am having the same issue with the following java implementation:

...
JsonArray key = JsonArray.from(stringA, stringB);
ViewQuery vq = ViewQuery.from(designDocument, viewName).stale(Stale.TRUE).key(key);
ViewResult viewResults = bucket.query(vq);

if (!viewResults.success()) {
        return new LinkedList<>();
}

List<ViewRow> rows = viewResults.allRows();

for (ViewRow row : rows) {
    ... do work ...
}
...

I appear to be having the same issue. Seemingly a single access call to the rows() and the same IllegalStateException is thrown on the “List rows = viewResults.allRows();” line.

Potentially when you access viewResults which might have already traversed the iterator.
Therefore you can try to replace

 if (!viewResults.success()) {
        return new LinkedList<>();
}

to

 if (!result == null) {
        return new LinkedList<>();
}

A good idea, but that block is not affecting the success of the allRows() call. I can remove the

if (!viewResults.success()) {
    return new LinkedList<>();
}

completely and the

List<ViewRow> rows = viewResults.allRows();

will still throw the IllegalStateException.

edit: It just appears to be a timing issue. Increasing the autoReleaseAfter value in the CouchbaseEnvironment object did the trick. It’s not clear what sort of performance implications this will have on our system without some thorough L&P testing.

The impact is bufs will live longer potentially, moving into the old generation and eventually causing more CG overhead (since collecting old is more expensive than young generations).