java.lang.IllegalStateException: more items arrived than were requested #3892

Activity

1. akarnokd commented on Apr 28, 2016

   

   akarnokd

   on Apr 28, 2016

   Member

   More actions

   Looks like OnSubscribeToObservableFuture was not refitted with backpressure support and it fires the future's value unconditionally. I'll post a fix for this. Thanks for reporting!
2. akarnokd

   added

   Bug

   on Apr 28, 2016
3. akarnokd commented on Apr 28, 2016

   

   akarnokd

   on Apr 28, 2016

   Member

   More actions

   See #3893.

   Workaround for now: apply onBackpressureBuffer() after the from(Future) you are using.
4. vadims commented on Apr 28, 2016

   

   vadims

   on Apr 28, 2016

   ContributorAuthor

   More actions

   Great, thanks!
5. vadims commented on May 2, 2016

   

   vadims

   on May 2, 2016

   ContributorAuthor

   More actions

   I think there might be other operators that need to be retrofitted?

   Non-fatal Exception: java.lang.IllegalStateException: more items arrived than were requested
          at rx.internal.producers.ProducerArbiter.produced(ProducerArbiter.java:98)
          at rx.internal.operators.OnSubscribeConcatMap$ConcatMapSubscriber.innerCompleted(OnSubscribeConcatMap.java:206)
          at rx.internal.operators.OnSubscribeConcatMap$ConcatMapInnerSubscriber.onCompleted(OnSubscribeConcatMap.java:345)
          at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
          at rx.internal.operators.OperatorSingle$ParentSubscriber.onCompleted(OperatorSingle.java:129)
          at rx.internal.operators.OperatorTakeLastOne$ParentSubscriber.onCompleted(OperatorTakeLastOne.java:107)
          at rx.internal.operators.OperatorMerge$MergeSubscriber.emitLoop(OperatorMerge.java:635)
          at rx.internal.operators.OperatorMerge$MergeSubscriber.emit(OperatorMerge.java:547)
          at rx.internal.operators.OperatorMerge$MergeSubscriber.onCompleted(OperatorMerge.java:268)
          at rx.internal.operators.OnSubscribeFromIterable.call(OnSubscribeFromIterable.java:47)
          at rx.internal.operators.OnSubscribeFromIterable.call(OnSubscribeFromIterable.java:32)
          at rx.Observable$2.call(Observable.java:233)
          at rx.Observable$2.call(Observable.java:225)
          at rx.Observable$2.call(Observable.java:233)
          at rx.Observable$2.call(Observable.java:225)
          at rx.Observable$2.call(Observable.java:233)
          at rx.Observable$2.call(Observable.java:225)
          at rx.Observable$2.call(Observable.java:233)
          at rx.Observable$2.call(Observable.java:225)
          at rx.Observable.unsafeSubscribe(Observable.java:8741)
          at rx.internal.operators.OnSubscribeConcatMap$ConcatMapSubscriber.drain(OnSubscribeConcatMap.java:286)
          at rx.internal.operators.OnSubscribeConcatMap$ConcatMapSubscriber.innerCompleted(OnSubscribeConcatMap.java:209)
          at rx.internal.operators.OnSubscribeConcatMap$ConcatMapInnerSubscriber.onCompleted(OnSubscribeConcatMap.java:345)
          at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
          at rx.internal.operators.OperatorSingle$ParentSubscriber.onCompleted(OperatorSingle.java:125)
          at rx.internal.operators.OperatorTakeLastOne$ParentSubscriber.emit(OperatorTakeLastOne.java:159)
          at rx.internal.operators.OperatorTakeLastOne$ParentSubscriber.onCompleted(OperatorTakeLastOne.java:125)
          at rx.internal.operators.OperatorMerge$MergeSubscriber.emitLoop(OperatorMerge.java:635)
          at rx.internal.operators.OperatorMerge$MergeSubscriber.emit(OperatorMerge.java:547)
          at rx.internal.operators.OperatorMerge$InnerSubscriber.onCompleted(OperatorMerge.java:836)
          at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
          at rx.internal.operators.OperatorOnBackpressureBuffer$BufferSubscriber.complete(OperatorOnBackpressureBuffer.java:169)
          at rx.internal.util.BackpressureDrainManager.drain(BackpressureDrainManager.java:187)
          at rx.internal.util.BackpressureDrainManager.terminateAndDrain(BackpressureDrainManager.java:114)
          at rx.internal.operators.OperatorOnBackpressureBuffer$BufferSubscriber.onCompleted(OperatorOnBackpressureBuffer.java:140)
          at rx.internal.operators.OperatorSubscribeOn$1$1.onCompleted(OperatorSubscribeOn.java:68)
          at rx.internal.operators.OnSubscribeToObservableFuture$ToObservableFuture.call(OnSubscribeToObservableFuture.java:76)
          at rx.internal.operators.OnSubscribeToObservableFuture$ToObservableFuture.call(OnSubscribeToObservableFuture.java:43)
          at rx.Observable.unsafeSubscribe(Observable.java:8741)
          at rx.internal.operators.OperatorSubscribeOn$1.call(OperatorSubscribeOn.java:94)
          at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55)
          at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:422)
          at java.util.concurrent.FutureTask.run(FutureTask.java:237)
          at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:152)
          at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:265)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587)
          at java.lang.Thread.run(Thread.java:818)
6. akarnokd commented on May 2, 2016

   

   akarnokd

   on May 2, 2016

   Member

   More actions

   Do you have a default value on your single() operator?
7. vadims commented on May 2, 2016

   

   vadims

   on May 2, 2016

   ContributorAuthor

   More actions

   We call lastOrDefault which calls new OperatorSingle<T>(defaultValue).
8. akarnokd commented on May 2, 2016

   

   akarnokd

   on May 2, 2016

   Member

   More actions

   Thanks, found the issue in OperatorSingle, when the default path is taken, the value is emitted without checking for backpressure. I'll post a fix soon.
9. akarnokd commented on May 2, 2016

   

   akarnokd

   on May 2, 2016

   Member

   More actions

   See #3905.
10. vadims commented on May 2, 2016

    

    vadims

    on May 2, 2016

    ContributorAuthor

    More actions

    Thanks!
11. zsxwing

    added a commit that references this issue on May 2, 2016

    1.x: fix singleOrDefault() backpressure if source is empty (#3905)

    ...

    04ef4e6
12. zsxwing commented on May 2, 2016

    

    zsxwing

    on May 2, 2016

    Member

    More actions

    Close via #3905
13. zsxwing

    closed this as completedon May 2, 2016