2.x: fix flatMapX calling SpscLinkedArrayQueue.offer concurrently (#4678)

Author: akarnokd

src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapMaybe.java

@@ -191,14 +191,18 @@ void innerSuccess(InnerObserver inner, R value) {
                     }
                 } else {
                     SpscLinkedArrayQueue<R> q = getOrCreateQueue();
-                    q.offer(value);
+                    synchronized (q) {
+                        q.offer(value);
+                    }
                 }
                 if (decrementAndGet() == 0) {
                     return;
                 }
             } else {
                 SpscLinkedArrayQueue<R> q = getOrCreateQueue();
-                q.offer(value);
+                synchronized (q) {
+                    q.offer(value);
+                }
                 active.decrementAndGet();
                 if (getAndIncrement() != 0) {
                     return;

src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMapSingle.java

@@ -191,14 +191,18 @@ void innerSuccess(InnerObserver inner, R value) {
                     }
                 } else {
                     SpscLinkedArrayQueue<R> q = getOrCreateQueue();
-                    q.offer(value);
+                    synchronized (q) {
+                        q.offer(value);
+                    }
                 }
                 if (decrementAndGet() == 0) {
                     return;
                 }
             } else {
                 SpscLinkedArrayQueue<R> q = getOrCreateQueue();
-                q.offer(value);
+                synchronized (q) {
+                    q.offer(value);
+                }
                 active.decrementAndGet();
                 if (getAndIncrement() != 0) {
                     return;

src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapMaybe.java

@@ -167,7 +167,9 @@ void innerSuccess(InnerObserver inner, R value) {
                 }
             } else {
                 SpscLinkedArrayQueue<R> q = getOrCreateQueue();
-                q.offer(value);
+                synchronized (q) {
+                    q.offer(value);
+                }
                 active.decrementAndGet();
                 if (getAndIncrement() != 0) {
                     return;

src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMapSingle.java

@@ -167,7 +167,9 @@ void innerSuccess(InnerObserver inner, R value) {
                 }
             } else {
                 SpscLinkedArrayQueue<R> q = getOrCreateQueue();
-                q.offer(value);
+                synchronized (q) {
+                    q.offer(value);
+                }
                 active.decrementAndGet();
                 if (getAndIncrement() != 0) {
                     return;