Interrupted while waiting for subscription to complete after upsert

This is my upser method,

public List upsertRawJsonDoc1(String generatorId, String idPrefix, List contents)
{
if(contents == null)
{
return null;
}
List rjd = new ArrayList(contents.size());
Observable.from(contents).flatMap(new Func1<String,Observable>(){
@Override
public Observable call(String t)
{
return bucket.async().counter(generatorId, 1)
.map(jsonLongDocument -> {
String idStr = idPrefix + generatorId + jsonLongDocument.content();
LOGGER.debug("\n method3 generatorId:" + idStr + “\n content:” + t);
return RawJsonDocument.create(idStr,t);
});
}}).subscribe(new Action1() {
@Override
public void call(RawJsonDocument t)
{
rjd.add(bucket.async().upsert(t).toBlocking().single());
}});
return rjd;
}

this is my Junit Tester:
@Test
public void testIncrementIds3()
{
assertThat(generatorId.upsertRawJsonDoc1(“counter”, “idprefix”, Arrays.asList(“aabbccdd”,“ffddeeaa”)).size(),is(2));
}

After upsert It throws exception, I don’t know why it throws exception,how to resolve it ? Thanks
rx.exceptions.OnErrorNotImplementedException: Interrupted while waiting for subscription to complete.
at rx.internal.util.InternalObservableUtils$ErrorNotImplementedAction.call(InternalObservableUtils.java:386) ~[rxjava-1.3.0.jar:1.3.0]
at rx.internal.util.InternalObservableUtils$ErrorNotImplementedAction.call(InternalObservableUtils.java:383) ~[rxjava-1.3.0.jar:1.3.0]
at rx.internal.util.ActionSubscriber.onError(ActionSubscriber.java:44) ~[rxjava-1.3.0.jar:1.3.0]
at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:153) ~[rxjava-1.3.0.jar:1.3.0]
at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:115) ~[rxjava-1.3.0.jar:1.3.0]
at rx.exceptions.Exceptions.throwOrReport(Exceptions.java:212) ~[rxjava-1.3.0.jar:1.3.0]
at rx.observers.SafeSubscriber.onNext(SafeSubscriber.java:139) ~[rxjava-1.3.0.jar:1.3.0]
at rx.internal.operators.OperatorMerge$MergeSubscriber.emitLoop(OperatorMerge.java:726) ~[rxjava-1.3.0.jar:1.3.0]
at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:433) ~[rxjava-1.3.0.jar:1.3.0]
at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:355) ~[rxjava-1.3.0.jar:1.3.0]
at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:846) ~[rxjava-1.3.0.jar:1.3.0]
at rx.internal.operators.OnSubscribeMap$MapSubscriber.onNext(OnSubscribeMap.java:77) ~[rxjava-1.3.0.jar:1.3.0]
at rx.internal.operators.OnSubscribeMap$MapSubscriber.onNext(OnSubscribeMap.java:77) ~[rxjava-1.3.0.jar:1.3.0]
at rx.observers.Subscribers$5.onNext(Subscribers.java:235) ~[rxjava-1.3.0.jar:1.3.0]
at rx.internal.operators.OnSubscribeDoOnEach$DoOnEachSubscriber.onNext(OnSubscribeDoOnEach.java:101) ~[rxjava-1.3.0.jar:1.3.0]
at rx.internal.producers.SingleProducer.request(SingleProducer.java:65) ~[rxjava-1.3.0.jar:1.3.0]
at rx.Subscriber.setProducer(Subscriber.java:211) ~[rxjava-1.3.0.jar:1.3.0]
at rx.internal.operators.OnSubscribeMap$MapSubscriber.setProducer(OnSubscribeMap.java:102) ~[rxjava-1.3.0.jar:1.3.0]
at rx.internal.operators.OnSubscribeMap$MapSubscriber.setProducer(OnSubscribeMap.java:102) ~[rxjava-1.3.0.jar:1.3.0]
at rx.Subscriber.setProducer(Subscriber.java:205) ~[rxjava-1.3.0.jar:1.3.0]
at rx.Subscriber.setProducer(Subscriber.java:205) ~[rxjava-1.3.0.jar:1.3.0]
at rx.subjects.AsyncSubject.onCompleted(AsyncSubject.java:103) ~[rxjava-1.3.0.jar:1.3.0]
at com.couchbase.client.core.endpoint.AbstractGenericHandler.completeResponse(AbstractGenericHandler.java:384) [core-io-1.4.6.jar:na]
at com.couchbase.client.core.endpoint.AbstractGenericHandler.access$000(AbstractGenericHandler.java:69) [core-io-1.4.6.jar:na]
at com.couchbase.client.core.endpoint.AbstractGenericHandler$1.call(AbstractGenericHandler.java:402) [core-io-1.4.6.jar:na]
at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) [rxjava-1.3.0.jar:1.3.0]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_121]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_121]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [na:1.8.0_121]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [na:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [na:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [na:1.8.0_121]
at java.lang.Thread.run(Thread.java:745) [na:1.8.0_121]
Caused by: java.lang.IllegalStateException: Interrupted while waiting for subscription to complete.
at rx.internal.util.BlockingUtils.awaitForComplete(BlockingUtils.java:54) ~[rxjava-1.3.0.jar:1.3.0]
at rx.observables.BlockingObservable.blockForSingle(BlockingObservable.java:460) ~[rxjava-1.3.0.jar:1.3.0]
at rx.observables.BlockingObservable.single(BlockingObservable.java:340) ~[rxjava-1.3.0.jar:1.3.0]
at com.abel.config.DBOperator$6.call(DBOperator.java:176) ~[classes/:na]
at com.abel.config.DBOperator$6.call(DBOperator.java:1) ~[classes/:na]
at rx.internal.util.ActionSubscriber.onNext(ActionSubscriber.java:39) ~[rxjava-1.3.0.jar:1.3.0]
at rx.observers.SafeSubscriber.onNext(SafeSubscriber.java:134) ~[rxjava-1.3.0.jar:1.3.0]
… 26 common frames omitted
Caused by: java.lang.InterruptedException: null
at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998) ~[na:1.8.0_121]
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) ~[na:1.8.0_121]
at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) ~[na:1.8.0_121]
at rx.internal.util.BlockingUtils.awaitForComplete(BlockingUtils.java:47) ~[rxjava-1.3.0.jar:1.3.0]
… 32 common frames omitted

no suggestion?no suggestion?no suggestion?