The following code works OK with couchbase-core-io-1.1.6.jar, couchbase-java-client-2.1.6.jar, rxjava-1.0.4.jar.
But if I run it with couchbase-core-io-1.7.2.jar, couchbase-java-client-2.7.2.jar, rxjava-1.0.4.jar it fails with:
com.couchbase.client.core.RequestCancelledException: Could not dispatch request, cancelling instead of retrying.
at com.couchbase.client.core.retry.RetryHelper.retryOrCancel(RetryHelper.java:51)
at com.couchbase.client.core.service.PooledService.send(PooledService.java:334)
at com.couchbase.client.core.node.CouchbaseNode.send(CouchbaseNode.java:249)
at com.couchbase.client.core.node.locate.KeyValueLocator.locateForCouchbaseBucket(KeyValueLocator.java:140)
at com.couchbase.client.core.node.locate.KeyValueLocator.locateAndDispatch(KeyValueLocator.java:90)
at com.couchbase.client.core.RequestHandler.dispatchRequest(RequestHandler.java:259)
at com.couchbase.client.core.RequestHandler.onEvent(RequestHandler.java:208)
at com.couchbase.client.core.RequestHandler.onEvent(RequestHandler.java:79)
at com.couchbase.client.deps.com.lmax.disruptor.BatchEventProcessor.run(BatchEventProcessor.java:150)
at com.couchbase.client.deps.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:748)
cbEnv = DefaultCouchbaseEnvironment.builder().connectTimeout(20000).build();
cluster = CouchbaseCluster.create(cbEnv, cbUrl);
//insert
Observable.from(list).flatMap(new Func1<JsonDocument, rx.Observable>()
{
@Override
public Observable call(final JsonDocument docToInsert)
{
return bucket
.insert(docToInsert)
.retryWhen(
RetryBuilder.anyOf(TemporaryFailureException.class)
.delay(Delay.exponential(TimeUnit.MILLISECONDS, 100)).max(30).build())
.doOnError(
new Action1()
{
@Override
public void call(Throwable e)
{
System.err.println(e);
}
}
);
}
}).last().toBlocking().single();