What would cause SingleEmitter.onSuccess() to generate a NoSuchElement exception?

user_1299008 0
Viewed 119

I have a Single flow organized like this:

getSomething() // returns Single<>
.flatMap(something -> {
   // various things
   return Single.defer( () -> {
     // various other things
     return Single.<SomeType>create(emitter -> {
        // some more stuff
        someCallbackApi(result -> {
           if (result.isError()) {
             emitter.onError( result.getCause() );
           } else {
             // guaranteed non-null data
             emitter.onSuccess( result.getData() ); // this generates NoSuchElement
           }
        });
     });
   })
   .retryWhen( ... )
   .flatMap( data -> handle(data) )
   .retryWhen( ... );
})
.retryWhen( ... )
.onErrorResumeNext(error -> process(error))
.subscribe(data -> handleSuccess(data), error -> handleError(error));

In test cases, the callback api Single successfully retries a number of times (determined by the test case), and every time on the last retry, the call to emitter.onSuccess() generates the exception below. What is going on? I haven't been able to restructure or change the downstream operators or subscribers to avoid the problem.

java.util.NoSuchElementException: null
    at io.reactivex.internal.operators.flowable.FlowableSingleSingle$SingleElementSubscriber.onComplete(FlowableSingleSingle.java:116)
    at io.reactivex.subscribers.SerializedSubscriber.onComplete(SerializedSubscriber.java:168)
    at io.reactivex.internal.operators.flowable.FlowableRepeatWhen$WhenReceiver.onComplete(FlowableRepeatWhen.java:118)
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.drainLoop(FlowableFlatMap.java:426)
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.drain(FlowableFlatMap.java:366)
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.onComplete(FlowableFlatMap.java:338)
    at io.reactivex.internal.operators.flowable.FlowableZip$ZipCoordinator.drain(FlowableZip.java:210)
    at io.reactivex.internal.operators.flowable.FlowableZip$ZipSubscriber.onNext(FlowableZip.java:381)
    at io.reactivex.processors.UnicastProcessor.drainFused(UnicastProcessor.java:363)
    at io.reactivex.processors.UnicastProcessor.drain(UnicastProcessor.java:396)
    at io.reactivex.processors.UnicastProcessor.onNext(UnicastProcessor.java:458)
    at io.reactivex.processors.SerializedProcessor.onNext(SerializedProcessor.java:103)
    at io.reactivex.internal.operators.flowable.FlowableRepeatWhen$WhenSourceSubscriber.again(FlowableRepeatWhen.java:171)
    at io.reactivex.internal.operators.flowable.FlowableRetryWhen$RetryWhenSubscriber.onError(FlowableRetryWhen.java:76)
    at io.reactivex.internal.operators.single.SingleToFlowable$SingleToFlowableObserver.onError(SingleToFlowable.java:67)
    at io.reactivex.internal.operators.single.SingleFlatMap$SingleFlatMapCallback$FlatMapSingleObserver.onError(SingleFlatMap.java:116)
    at io.reactivex.internal.operators.flowable.FlowableSingleSingle$SingleElementSubscriber.onError(FlowableSingleSingle.java:97)
    at io.reactivex.subscribers.SerializedSubscriber.onError(SerializedSubscriber.java:142)
    at io.reactivex.internal.operators.flowable.FlowableRepeatWhen$WhenReceiver.onError(FlowableRepeatWhen.java:112)
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.checkTerminate(FlowableFlatMap.java:567)
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.drainLoop(FlowableFlatMap.java:374)
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.drain(FlowableFlatMap.java:366)
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.innerError(FlowableFlatMap.java:606)
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$InnerSubscriber.onError(FlowableFlatMap.java:672)
    at io.reactivex.internal.subscriptions.EmptySubscription.error(EmptySubscription.java:55)
    at io.reactivex.internal.operators.flowable.FlowableError.subscribeActual(FlowableError.java:40)
    at io.reactivex.Flowable.subscribe(Flowable.java:14918)
    at io.reactivex.Flowable.subscribe(Flowable.java:14865)
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.onNext(FlowableFlatMap.java:163)
    at io.reactivex.internal.operators.flowable.FlowableZip$ZipCoordinator.drain(FlowableZip.java:249)
    at io.reactivex.internal.operators.flowable.FlowableZip$ZipSubscriber.onNext(FlowableZip.java:381)
    at io.reactivex.processors.UnicastProcessor.drainFused(UnicastProcessor.java:363)
    at io.reactivex.processors.UnicastProcessor.drain(UnicastProcessor.java:396)
    at io.reactivex.processors.UnicastProcessor.onNext(UnicastProcessor.java:458)
    at io.reactivex.processors.SerializedProcessor.onNext(SerializedProcessor.java:103)
    at io.reactivex.internal.operators.flowable.FlowableRepeatWhen$WhenSourceSubscriber.again(FlowableRepeatWhen.java:171)
    at io.reactivex.internal.operators.flowable.FlowableRetryWhen$RetryWhenSubscriber.onError(FlowableRetryWhen.java:76)
    at io.reactivex.internal.operators.single.SingleToFlowable$SingleToFlowableObserver.onError(SingleToFlowable.java:67)
    at io.reactivex.internal.operators.single.SingleFlatMap$SingleFlatMapCallback$FlatMapSingleObserver.onError(SingleFlatMap.java:116)
    at io.reactivex.internal.disposables.EmptyDisposable.error(EmptyDisposable.java:78)
    at io.reactivex.internal.operators.single.SingleError.subscribeActual(SingleError.java:42)
    at io.reactivex.Single.subscribe(Single.java:3603)
    at io.reactivex.internal.operators.single.SingleFlatMap$SingleFlatMapCallback.onSuccess(SingleFlatMap.java:84)
    at io.reactivex.internal.operators.flowable.FlowableSingleSingle$SingleElementSubscriber.onComplete(FlowableSingleSingle.java:114)
    at io.reactivex.subscribers.SerializedSubscriber.onComplete(SerializedSubscriber.java:168)
    at io.reactivex.internal.operators.flowable.FlowableRetryWhen$RetryWhenSubscriber.onComplete(FlowableRetryWhen.java:82)
    at io.reactivex.internal.subscriptions.DeferredScalarSubscription.complete(DeferredScalarSubscription.java:134)
    at io.reactivex.internal.operators.single.SingleToFlowable$SingleToFlowableObserver.onSuccess(SingleToFlowable.java:62)
    at io.reactivex.internal.operators.single.SingleCreate$Emitter.onSuccess(SingleCreate.java:67)

Solved: Many thanks to @dano for pointing out the retryWhen behavior when used with Single. In this case, the outermost retryWhen operator had a bad terminating condition, roughly like:

.retryWhen(errors -> errors.zipWith( Flowable.range(1, maxRetries), ...)
                           .flatMap( zipped -> {
  if (zipped.retryCount() <= maxRetries) {
    return Flowable.just(0L);
  }
  return Flowable.error( new Exception() );
})

...Flowable.range() will complete when it has generated the last number, which will cause the Single to emit NoSuchElement. Just bumping the count argument to Flowable.range() by one is enough to fix the problem:

.retryWhen(errors -> errors.zipWith( Flowable.range(1, maxRetries + 1), ...)
                           .flatMap( zipped -> {
  if (zipped.retryCount() <= maxRetries) {
    return Flowable.just(0L);
  }
  return Flowable.error( new Exception() );
})
rx-java2
1 Answers

This is happening because of the way you implemented the callback you passed to retryWhen. The retryWhen docuementation states (emphasis mine):

Re-subscribes to the current Single if and when the Publisher returned by the handler function signals a value.

If the Publisher signals an onComplete, the resulting Single will signal a NoSuchElementException.

One of the Flowable instances you're returning inside of the calls to retryWhen is emitting onComplete, which leads to the NoSuchElementException.

Here's a very simple example that produces the same error:

Single.error(new Exception("hey"))
    .retryWhen(e -> Flowable.just(1))
    .subscribe(System.out::println, e -> e.printStackTrace());

The stacktrace this produces starts with this, same as yours:

java.util.NoSuchElementException
    at io.reactivex.internal.operators.flowable.FlowableSingleSingle$SingleElementSubscriber.onComplete(FlowableSingleSingle.java:116)
    at io.reactivex.subscribers.SerializedSubscriber.onComplete(SerializedSubscriber.java:168)
    at io.reactivex.internal.operators.flowable.FlowableRepeatWhen$WhenReceiver.onComplete(FlowableRepeatWhen.java:118)

You don't include any of your code from inside the retryWhen calls, so I can't say exactly what you did wrong, but generally you want to chain whatever you do to the Flowable that is passed in. So my example above would look like this, if we really wanted to retry forever:

Single.error(new Exception("hey"))
    .retryWhen(e -> e.flatMap(ign -> Flowable.just(1)))
    .subscribe(System.out::println, e -> e.printStackTrace());
user_20735950
Related