[2.1.11] NPE in SpscLinkedArrayQueue.lvElement

[boris-petrov]

I've seen already a couple of those after upgrading. Unfortunately I cannot give you any reproduction as I have zero idea where it might come from but here are a couple of stacktraces:

java.lang.NullPointerException
 at io.reactivex.internal.queue.SpscLinkedArrayQueue.lvElement(SpscLinkedArrayQueue.java:249)
 at io.reactivex.internal.queue.SpscLinkedArrayQueue.newBufferPoll(SpscLinkedArrayQueue.java:154)
 at io.reactivex.internal.queue.SpscLinkedArrayQueue.poll(SpscLinkedArrayQueue.java:144)
 at io.reactivex.internal.queue.SpscLinkedArrayQueue.clear(SpscLinkedArrayQueue.java:184)
 at io.reactivex.internal.operators.mixed.ObservableConcatMapMaybe$ConcatMapMaybeMainObserver.dispose(ObservableConcatMapMaybe.java:152)
 at io.reactivex.internal.observers.BasicFuseableObserver.dispose(BasicFuseableObserver.java:152)
 at io.reactivex.internal.observers.BasicFuseableObserver.dispose(BasicFuseableObserver.java:152)
 at io.reactivex.internal.operators.observable.ObservableTake$TakeObserver.dispose(ObservableTake.java:90)
 at io.reactivex.internal.disposables.DisposableHelper.dispose(DisposableHelper.java:125)
 at io.reactivex.internal.observers.LambdaObserver.dispose(LambdaObserver.java:100)
 at java.util.HashMap$Values.forEach(HashMap.java:981)

java.lang.NullPointerException
 at io.reactivex.internal.queue.SpscLinkedArrayQueue.lvElement(SpscLinkedArrayQueue.java:249)
 at io.reactivex.internal.queue.SpscLinkedArrayQueue.newBufferPoll(SpscLinkedArrayQueue.java:154)
 at io.reactivex.internal.queue.SpscLinkedArrayQueue.poll(SpscLinkedArrayQueue.java:144)
 at io.reactivex.internal.queue.SpscLinkedArrayQueue.clear(SpscLinkedArrayQueue.java:184)
 at io.reactivex.internal.operators.mixed.ObservableConcatMapMaybe$ConcatMapMaybeMainObserver.drain(ObservableConcatMapMaybe.java:200)
 at io.reactivex.internal.operators.mixed.ObservableConcatMapMaybe$ConcatMapMaybeMainObserver.innerSuccess(ObservableConcatMapMaybe.java:165)
 at io.reactivex.internal.operators.mixed.ObservableConcatMapMaybe$ConcatMapMaybeMainObserver$ConcatMapMaybeObserver.onSuccess(ObservableConcatMapMaybe.java:290)
 at io.reactivex.internal.operators.single.SingleFlatMapMaybe$FlatMapMaybeObserver.onSuccess(SingleFlatMapMaybe.java:117)
 at io.reactivex.internal.operators.maybe.MaybeJust.subscribeActual(MaybeJust.java:36)
 at io.reactivex.Maybe.subscribe(Maybe.java:4065)

This didn't happen on 2.1.10 or any of the previous versions. Please tell me if I can help more.

[akarnokd]

Hi and thanks for reporting. This is a bug in the newly introduced operators of 2.1.11. As a temporary workaround, use the traditional concatMap with a type conversion operator.

[boris-petrov]

Thank you for the immediate support! Thanks for the great work!

[akarnokd]

You are welcome. Still, please keep using 2.1.11 as long as 2.1.12 comes out in case there are more mistakes such as this - you seem to be exceptionally lucky to stumble upon them as it takes about 1 minute on my desktop machine to trigger the bug by only using standard methods.

[DavidDTA]

Hi, we're seeing a similar crash in 2.2.12:

Fatal Exception: java.lang.NullPointerException: Attempt to invoke virtual method 'java.lang.Object java.util.concurrent.atomic.AtomicReferenceArray.get(int)' on a null object reference
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.lvElement + 251(SpscLinkedArrayQueue.java:251)
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.poll + 138(SpscLinkedArrayQueue.java:138)
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.clear + 186(SpscLinkedArrayQueue.java:186)
       at io.reactivex.internal.operators.flowable.FlowableOnBackpressureBuffer$BackpressureBufferSubscriber.cancel + 154(FlowableOnBackpressureBuffer.java:154)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.cancel + 154(FlowableObserveOn.java:154)
       at io.reactivex.internal.subscribers.BasicFuseableSubscriber.cancel + 158(BasicFuseableSubscriber.java:158)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.cancel + 154(FlowableObserveOn.java:154)
       at io.reactivex.internal.subscribers.BasicFuseableConditionalSubscriber.cancel + 157(BasicFuseableConditionalSubscriber.java:157)
       at io.reactivex.internal.subscribers.BasicFuseableSubscriber.cancel + 158(BasicFuseableSubscriber.java:158)
       at io.reactivex.internal.operators.flowable.FlowableElementAtSingle$ElementAtSubscriber.onNext + 85(FlowableElementAtSingle.java:85)
       at io.reactivex.internal.operators.flowable.FlowableDistinctUntilChanged$DistinctUntilChangedSubscriber.tryOnNext + 101(FlowableDistinctUntilChanged.java:101)
       at io.reactivex.internal.operators.flowable.FlowableMap$MapConditionalSubscriber.tryOnNext + 128(FlowableMap.java:128)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$ObserveOnConditionalSubscriber.runAsync + 649(FlowableObserveOn.java:649)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.run + 176(FlowableObserveOn.java:176)

[akarnokd]

@DavidDTA could you provide a code sample that reproduces this issue?

[DavidDTA]

We just upgraded our project from 1.x, and did not see these errors in testing, but are seeing them in production. We're working to figure out the exact cause, and will provide more information as we have it.

Here are some additional stack traces, in case you notice any pattern:

Fatal Exception: java.lang.NullPointerException: Attempt to invoke virtual method 'java.lang.Object java.util.concurrent.atomic.AtomicReferenceArray.get(int)' on a null object reference
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.lvElement + 251(SpscLinkedArrayQueue.java:251)
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.newBufferPoll + 155(SpscLinkedArrayQueue.java:155)
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.poll + 145(SpscLinkedArrayQueue.java:145)
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.clear + 186(SpscLinkedArrayQueue.java:186)
       at io.reactivex.internal.operators.flowable.FlowableOnBackpressureBuffer$BackpressureBufferSubscriber.cancel + 154(FlowableOnBackpressureBuffer.java:154)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.cancel + 154(FlowableObserveOn.java:154)
       at io.reactivex.internal.subscribers.BasicFuseableSubscriber.cancel + 158(BasicFuseableSubscriber.java:158)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.cancel + 154(FlowableObserveOn.java:154)
       at io.reactivex.internal.subscribers.BasicFuseableConditionalSubscriber.cancel + 157(BasicFuseableConditionalSubscriber.java:157)
       at io.reactivex.internal.subscribers.BasicFuseableSubscriber.cancel + 158(BasicFuseableSubscriber.java:158)
       at io.reactivex.internal.operators.mixed.FlowableConcatMapMaybe$ConcatMapMaybeSubscriber.cancel + 168(FlowableConcatMapMaybe.java:168)
       at io.reactivex.internal.operators.flowable.FlowableElementAtSingle$ElementAtSubscriber.onNext + 85(FlowableElementAtSingle.java:85)
       at io.reactivex.internal.operators.mixed.FlowableConcatMapMaybe$ConcatMapMaybeSubscriber.drain + 284(FlowableConcatMapMaybe.java:284)
       at io.reactivex.internal.operators.mixed.FlowableConcatMapMaybe$ConcatMapMaybeSubscriber.onNext + 137(FlowableConcatMapMaybe.java:137)
       at io.reactivex.internal.operators.flowable.FlowableDistinctUntilChanged$DistinctUntilChangedSubscriber.tryOnNext + 101(FlowableDistinctUntilChanged.java:101)
       at io.reactivex.internal.operators.flowable.FlowableMap$MapConditionalSubscriber.tryOnNext + 128(FlowableMap.java:128)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$ObserveOnConditionalSubscriber.runAsync + 649(FlowableObserveOn.java:649)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.run + 176(FlowableObserveOn.java:176)

Fatal Exception: java.lang.NullPointerException: Attempt to invoke virtual method 'java.lang.Object java.util.concurrent.atomic.AtomicReferenceArray.get(int)' on a null object reference
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.lvElement + 251(SpscLinkedArrayQueue.java:251)
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.newBufferPoll + 155(SpscLinkedArrayQueue.java:155)
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.poll + 145(SpscLinkedArrayQueue.java:145)
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.clear + 186(SpscLinkedArrayQueue.java:186)
       at io.reactivex.internal.operators.flowable.FlowableOnBackpressureBuffer$BackpressureBufferSubscriber.clear + 264(FlowableOnBackpressureBuffer.java:264)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.clear + 236(FlowableObserveOn.java:236)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.checkTerminated + 188(FlowableObserveOn.java:188)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$ObserveOnSubscriber.runAsync + 399(FlowableObserveOn.java:399)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.run + 176(FlowableObserveOn.java:176)
       at io.reactivex.internal.schedulers.ScheduledRunnable.run + 66(ScheduledRunnable.java:66)
       at io.reactivex.internal.schedulers.ScheduledRunnable.call + 57(ScheduledRunnable.java:57)

Fatal Exception: java.lang.NullPointerException: Attempt to invoke virtual method 'java.lang.Object java.util.concurrent.atomic.AtomicReferenceArray.get(int)' on a null object reference
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.lvElement + 251(SpscLinkedArrayQueue.java:251)
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.poll + 138(SpscLinkedArrayQueue.java:138)
       at io.reactivex.internal.queue.SpscLinkedArrayQueue.clear + 186(SpscLinkedArrayQueue.java:186)
       at io.reactivex.internal.operators.flowable.FlowableOnBackpressureBuffer$BackpressureBufferSubscriber.cancel + 154(FlowableOnBackpressureBuffer.java:154)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.cancel + 154(FlowableObserveOn.java:154)
       at io.reactivex.internal.subscribers.BasicFuseableSubscriber.cancel + 158(BasicFuseableSubscriber.java:158)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.cancel + 154(FlowableObserveOn.java:154)
       at io.reactivex.internal.subscribers.BasicFuseableSubscriber.cancel + 158(BasicFuseableSubscriber.java:158)
       at io.reactivex.internal.operators.flowable.FlowableSkipWhile$SkipWhileSubscriber.cancel + 93(FlowableSkipWhile.java:93)
       at io.reactivex.internal.subscribers.BasicFuseableSubscriber.cancel + 158(BasicFuseableSubscriber.java:158)
       at io.reactivex.internal.operators.flowable.FlowableElementAtSingle$ElementAtSubscriber.onNext + 85(FlowableElementAtSingle.java:85)
       at io.reactivex.internal.operators.flowable.FlowableFilter$FilterSubscriber.tryOnNext + 74(FlowableFilter.java:74)
       at io.reactivex.internal.operators.flowable.FlowableFilter$FilterSubscriber.onNext + 52(FlowableFilter.java:52)
       at io.reactivex.internal.operators.flowable.FlowableSkipWhile$SkipWhileSubscriber.onNext + 56(FlowableSkipWhile.java:56)
       at io.reactivex.internal.operators.flowable.FlowableMap$MapSubscriber.onNext + 68(FlowableMap.java:68)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$ObserveOnSubscriber.runAsync + 407(FlowableObserveOn.java:407)
       at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.run + 176(FlowableObserveOn.java:176)

[akarnokd]

It crashes when it tries to clear the buffer of onBackpressureBuffer upon receiving cancellation. However, this particular crash would only happen when clear is invoked concurrently, which shouldn't be possible. Any more indications on the operator chain would be helpful because the stacktrace does not indicate the full chain.

[DavidDTA]

The full chain for the last stack trace is:

Single.just(...)
  .delaySubscription(
    Completable.amb(
      listOf(
        Completable.timer(...),
        FlowableProcessor
          .map { ... }
          .onBackpressureBuffer()
          .compose { it.observeOn(...) }
          .map { ... }
          .observeOn(...)
          .map { ... }
          .skipWhile { ... }
          .filter { ... }
          .firstOrErrror()
          .ignoreElement()
      )
    )
      .toObservable<Any>()
  )
  .subscribeOn(...)

[akarnokd]

Thanks. I have some suspicion I have to verify. In the meantime, could you see if you put hide() after onBackpressureBuffer() the crashes stop happening?

[DavidDTA]

I can attempt this, but it will take time to collect that data.