OperatorMerge - IllegalStateException from RxRingBuffer

[davidmoten]

Using 1.0.4, on completion of a merge of some busy async sources (11 million items getting round robined over Schedulers.computation in batches of 20000) I am getting the stack trace below. Is this is a bug?

java.lang.IllegalStateException: This instance has been unsubscribed and the queue is no longer usable.
    at rx.internal.util.RxRingBuffer.onNext(RxRingBuffer.java:335)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.handleScalarSynchronousObservableWithRequestLimits(OperatorMerge.java:277)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.handleScalarSynchronousObservable(OperatorMerge.java:216)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:149)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:93)
    at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:55)
    at rx.internal.operators.NotificationLite.accept(NotificationLite.java:150)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainScalarValueQueue(OperatorMerge.java:368)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainQueuesIfNeeded(OperatorMerge.java:316)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.access$400(OperatorMerge.java:93)
    at rx.internal.operators.OperatorMerge$MergeProducer.request(OperatorMerge.java:508)
    at rx.Subscriber.request(Subscriber.java:107)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainQueuesIfNeeded(OperatorMerge.java:323)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.access$400(OperatorMerge.java:93)
    at rx.internal.operators.OperatorMerge$MergeProducer.request(OperatorMerge.java:508)
    at rx.internal.operators.OperatorSubscribeOn$1$1$1$1$1.call(OperatorSubscribeOn.java:94)
    at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:47)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
    at java.util.concurrent.FutureTask.run(FutureTask.java:262)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:178)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:292)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)
Caused by: rx.exceptions.OnErrorThrowable$OnNextValue: OnError while emitting onNext value: au.gov.amsa.util.nmea.NmeaMessage.class
    at rx.exceptions.OnErrorThrowable.addValueAsLastCause(OnErrorThrowable.java:98)
    at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:58)
    ... 18 more

[davidmoten]

same error with latest 1.x branch

[davidmoten]

correction, does NOT happen with latest 1.x branch! Thank you, closing!

[akarnokd]

This is a disagreeable feature I've already asked in #1963. There is still a small window where this could happen.