3.x: Standardize MissingBackpressureException message, introduce Queu…

…eOverflowException (#7459) * 3.x: Standardize MissingBackpressureException messages * Use the correct message. * Fix tests * Add QueueOverflowException, fix tests * Update CompletableConcatTest.java
ReactiveX · Aug 8, 2022 · 2edba23 · 2edba23
1 parent ef51a90
commit 2edba23
Show file tree

Hide file tree

Showing 61 changed files with 164 additions and 94 deletions.
diff --git a/src/main/java/io/reactivex/rxjava3/exceptions/MissingBackpressureException.java b/src/main/java/io/reactivex/rxjava3/exceptions/MissingBackpressureException.java
@@ -20,6 +20,15 @@ public final class MissingBackpressureException extends RuntimeException {
 
     private static final long serialVersionUID = 8517344746016032542L;
 
+    /**
+     * The default error message.
+     * <p>
+     * This can happen if the downstream doesn't call {@link org.reactivestreams.Subscription#request(long)}
+     * in time or at all.
+     * @since 3.1.6
+     */
+    public static final String DEFAULT_MESSAGE = "Could not emit value due to lack of requests";
+
     /**
      * Constructs a MissingBackpressureException without message or cause.
      */
@@ -35,4 +44,13 @@ public MissingBackpressureException(String message) {
         super(message);
     }
 
+    /**
+     * Constructs a new {@code MissingBackpressureException} with the
+     * default message {@value #DEFAULT_MESSAGE}.
+     * @return the new {@code MissingBackpressureException} instance.
+     * @since 3.1.6
+     */
+    public static MissingBackpressureException createDefault() {
+        return new MissingBackpressureException(DEFAULT_MESSAGE);
+    }
 }
diff --git a/src/main/java/io/reactivex/rxjava3/exceptions/QueueOverflowException.java b/src/main/java/io/reactivex/rxjava3/exceptions/QueueOverflowException.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.rxjava3.exceptions;
+
+/**
+ * Indicates an overflow happened because the upstream disregarded backpressure completely or
+ * {@link org.reactivestreams.Subscriber#onNext(Object)} was called concurrently from multiple threads
+ * without synchronization. Rarely, it is an indication of bugs inside an operator.
+ * @since 3.1.6
+ */
+public final class QueueOverflowException extends RuntimeException {
+
+    private static final long serialVersionUID = 8517344746016032542L;
+
+    /**
+     * The message for queue overflows.
+     * <p>
+     * This can happen if the upstream disregards backpressure completely or calls
+     * {@link org.reactivestreams.Subscriber#onNext(Object)} concurrently from multiple threads
+     * without synchronization. Rarely, it is an indication of bugs inside an operator.
+     */
+    private static final String DEFAULT_MESSAGE = "Queue overflow due to illegal concurrent onNext calls or a bug in an operator";
+
+    /**
+     * Constructs a QueueOverflowException with the default message.
+     */
+    public QueueOverflowException() {
+        this(DEFAULT_MESSAGE);
+    }
+
+    /**
+     * Constructs a QueueOverflowException with the given message but no cause.
+     * @param message the error message
+     */
+    public QueueOverflowException(String message) {
+        super(message);
+    }
+}
diff --git a/src/main/java/io/reactivex/rxjava3/internal/jdk8/FlowableFlatMapStream.java b/src/main/java/io/reactivex/rxjava3/internal/jdk8/FlowableFlatMapStream.java
@@ -174,7 +174,7 @@ public void onNext(T t) {
             if (sourceMode != QueueFuseable.ASYNC) {
                 if (!queue.offer(t)) {
                     upstream.cancel();
-                    onError(new MissingBackpressureException("Queue full?!"));
+                    onError(new QueueOverflowException());
                     return;
                 }
             }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/completable/CompletableConcat.java b/src/main/java/io/reactivex/rxjava3/internal/operators/completable/CompletableConcat.java
@@ -120,7 +120,7 @@ public void onSubscribe(Subscription s) {
         public void onNext(CompletableSource t) {
             if (sourceFused == QueueSubscription.NONE) {
                 if (!queue.offer(t)) {
-                    onError(new MissingBackpressureException());
+                    onError(new QueueOverflowException());
                     return;
                 }
             }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/BlockingFlowableIterable.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/BlockingFlowableIterable.java
@@ -140,7 +140,7 @@ public void onNext(T t) {
             if (!queue.offer(t)) {
                 SubscriptionHelper.cancel(this);
 
-                onError(new MissingBackpressureException("Queue full?!"));
+                onError(new QueueOverflowException());
             } else {
                 signalConsumer();
             }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMap.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMap.java
@@ -19,7 +19,7 @@
 import org.reactivestreams.*;
 
 import io.reactivex.rxjava3.core.*;
-import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.exceptions.*;
 import io.reactivex.rxjava3.functions.*;
 import io.reactivex.rxjava3.internal.subscriptions.*;
 import io.reactivex.rxjava3.internal.util.*;
@@ -152,7 +152,7 @@ public final void onNext(T t) {
             if (sourceMode != QueueSubscription.ASYNC) {
                 if (!queue.offer(t)) {
                     upstream.cancel();
-                    onError(new IllegalStateException("Queue full?!"));
+                    onError(new QueueOverflowException());
                     return;
                 }
             }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapEager.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapEager.java
@@ -200,7 +200,7 @@ public void innerNext(InnerQueuedSubscriber<R> inner, R value) {
                 drain();
             } else {
                 inner.cancel();
-                innerError(inner, new MissingBackpressureException());
+                innerError(inner, MissingBackpressureException.createDefault());
             }
         }
 

diff --git a/...ain/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapScheduler.java b/...ain/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapScheduler.java
@@ -19,7 +19,7 @@
 import org.reactivestreams.*;
 
 import io.reactivex.rxjava3.core.*;
-import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.exceptions.*;
 import io.reactivex.rxjava3.functions.*;
 import io.reactivex.rxjava3.internal.operators.flowable.FlowableConcatMap.*;
 import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
@@ -151,7 +151,7 @@ public final void onNext(T t) {
             if (sourceMode != QueueSubscription.ASYNC) {
                 if (!queue.offer(t)) {
                     upstream.cancel();
-                    onError(new IllegalStateException("Queue full?!"));
+                    onError(new QueueOverflowException());
                     return;
                 }
             }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableCreate.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableCreate.java
@@ -442,7 +442,7 @@ static final class ErrorAsyncEmitter<T> extends NoOverflowBaseAsyncEmitter<T> {
 
         @Override
         void onOverflow() {
-            onError(new MissingBackpressureException("create: could not emit value due to lack of requests"));
+            onError(new MissingBackpressureException("create: " + MissingBackpressureException.DEFAULT_MESSAGE));
         }
 
     }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounce.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounce.java
@@ -148,7 +148,7 @@ void emit(long idx, T value) {
                     BackpressureHelper.produced(this, 1);
                 } else {
                     cancel();
-                    downstream.onError(new MissingBackpressureException("Could not deliver value due to lack of requests"));
+                    downstream.onError(MissingBackpressureException.createDefault());
                 }
             }
         }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounceTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounceTimed.java
@@ -159,7 +159,7 @@ void emit(long idx, T t, DebounceEmitter<T> emitter) {
                     emitter.dispose();
                 } else {
                     cancel();
-                    downstream.onError(new MissingBackpressureException("Could not deliver value due to lack of requests"));
+                    downstream.onError(MissingBackpressureException.createDefault());
                 }
             }
         }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlatMap.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlatMap.java
@@ -243,7 +243,7 @@ void tryEmitScalar(U value) {
                         q = getMainQueue();
                     }
                     if (!q.offer(value)) {
-                        onError(new MissingBackpressureException("Scalar queue full?!"));
+                        onError(new QueueOverflowException());
                     }
                 }
                 if (decrementAndGet() == 0) {
@@ -252,7 +252,7 @@ void tryEmitScalar(U value) {
             } else {
                 SimpleQueue<U> q = getMainQueue();
                 if (!q.offer(value)) {
-                    onError(new MissingBackpressureException("Scalar queue full?!"));
+                    onError(new QueueOverflowException());
                     return;
                 }
                 if (getAndIncrement() != 0) {
@@ -278,7 +278,7 @@ void tryEmit(U value, InnerSubscriber<T, U> inner) {
                         inner.queue = q;
                     }
                     if (!q.offer(value)) {
-                        onError(new MissingBackpressureException("Inner queue full?!"));
+                        onError(new QueueOverflowException());
                     }
                 }
                 if (decrementAndGet() == 0) {
@@ -291,7 +291,7 @@ void tryEmit(U value, InnerSubscriber<T, U> inner) {
                     inner.queue = q;
                 }
                 if (!q.offer(value)) {
-                    onError(new MissingBackpressureException("Inner queue full?!"));
+                    onError(new QueueOverflowException());
                     return;
                 }
                 if (getAndIncrement() != 0) {

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlattenIterable.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlattenIterable.java
@@ -180,7 +180,7 @@ public void onNext(T t) {
                 return;
             }
             if (fusionMode == NONE && !queue.offer(t)) {
-                onError(new MissingBackpressureException("Queue is full?!"));
+                onError(new QueueOverflowException());
                 return;
             }
             drain();

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupBy.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupBy.java
@@ -168,7 +168,7 @@ public void onNext(T t) {
                     if (emittedGroups != get()) {
                         downstream.onNext(group);
                     } else {
-                        MissingBackpressureException mbe = new MissingBackpressureException(groupHangWarning(emittedGroups));
+                        MissingBackpressureException mbe = groupHangWarning(emittedGroups);
                         mbe.initCause(ex);
                         onError(mbe);
                         return;
@@ -194,13 +194,13 @@ public void onNext(T t) {
                     }
                 } else {
                     upstream.cancel();
-                    onError(new MissingBackpressureException(groupHangWarning(emittedGroups)));
+                    onError(groupHangWarning(emittedGroups));
                 }
             }
         }
 
-        static String groupHangWarning(long n) {
-            return "Unable to emit a new group (#" + n + ") due to lack of requests. Please make sure the downstream can always accept a new group as well as each group is consumed in order for the whole operator to be able to proceed.";
+        static MissingBackpressureException groupHangWarning(long n) {
+            return new MissingBackpressureException("Unable to emit a new group (#" + n + ") due to lack of requests. Please make sure the downstream can always accept a new group as well as each group is consumed in order for the whole operator to be able to proceed.");
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupJoin.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupJoin.java
@@ -276,7 +276,7 @@ void drain() {
                             a.onNext(w);
                             BackpressureHelper.produced(requested, 1);
                         } else {
-                            fail(new MissingBackpressureException("Could not emit value due to lack of requests"), a, q);
+                            fail(MissingBackpressureException.createDefault(), a, q);
                             return;
                         }
 

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableInterval.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableInterval.java
@@ -93,7 +93,7 @@ public void run() {
                     downstream.onNext(count++);
                     BackpressureHelper.produced(this, 1);
                 } else {
-                    downstream.onError(new MissingBackpressureException("Can't deliver value " + count + " due to lack of requests"));
+                    downstream.onError(new MissingBackpressureException("Could not emit value " + count + " due to lack of requests"));
                     DisposableHelper.dispose(resource);
                 }
             }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableIntervalRange.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableIntervalRange.java
@@ -114,7 +114,7 @@ public void run() {
                         decrementAndGet();
                     }
                 } else {
-                    downstream.onError(new MissingBackpressureException("Can't deliver value " + count + " due to lack of requests"));
+                    downstream.onError(new MissingBackpressureException("Could not emit value " + count + " due to lack of requests"));
                     DisposableHelper.dispose(resource);
                 }
             }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableJoin.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableJoin.java
@@ -260,7 +260,7 @@ void drain() {
 
                                 e++;
                             } else {
-                                ExceptionHelper.addThrowable(error, new MissingBackpressureException("Could not emit value due to lack of requests"));
+                                ExceptionHelper.addThrowable(error, MissingBackpressureException.createDefault());
                                 q.clear();
                                 cancelAll();
                                 errorAll(a);
@@ -321,7 +321,7 @@ else if (mode == RIGHT_VALUE) {
 
                                 e++;
                             } else {
-                                ExceptionHelper.addThrowable(error, new MissingBackpressureException("Could not emit value due to lack of requests"));
+                                ExceptionHelper.addThrowable(error, MissingBackpressureException.createDefault());
                                 q.clear();
                                 cancelAll();
                                 errorAll(a);

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableObserveOn.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableObserveOn.java
@@ -113,7 +113,7 @@ public final void onNext(T t) {
             if (!queue.offer(t)) {
                 upstream.cancel();
 
-                error = new MissingBackpressureException("Queue is full?!");
+                error = new QueueOverflowException();
                 done = true;
             }
             trySchedule();

diff --git a/...o/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategy.java b/...o/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategy.java
@@ -140,7 +140,7 @@ public void onNext(T t) {
                 }
             } else if (callError) {
                 upstream.cancel();
-                onError(new MissingBackpressureException());
+                onError(MissingBackpressureException.createDefault());
             } else {
                 drain();
             }

diff --git a/...in/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureError.java b/...in/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureError.java
@@ -66,7 +66,7 @@ public void onNext(T t) {
                 BackpressureHelper.produced(this, 1);
             } else {
                 upstream.cancel();
-                onError(new MissingBackpressureException("could not emit value due to lack of requests"));
+                onError(MissingBackpressureException.createDefault());
             }
         }
 

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublish.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublish.java
@@ -226,7 +226,7 @@ public void onSubscribe(Subscription s) {
         public void onNext(T t) {
             // we expect upstream to honor backpressure requests
             if (sourceMode == QueueSubscription.NONE && !queue.offer(t)) {
-                onError(new MissingBackpressureException("Prefetch queue is full?!"));
+                onError(new QueueOverflowException());
                 return;
             }
             // since many things can happen concurrently, we have a common dispatch

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublishMulticast.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublishMulticast.java
@@ -212,7 +212,7 @@ public void onNext(T t) {
             }
             if (sourceMode == QueueSubscription.NONE && !queue.offer(t)) {
                 upstream.get().cancel();
-                onError(new MissingBackpressureException());
+                onError(MissingBackpressureException.createDefault());
                 return;
             }
             drain();

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSamplePublisher.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSamplePublisher.java
@@ -129,7 +129,7 @@ void emit() {
                     BackpressureHelper.produced(requested, 1);
                 } else {
                     cancel();
-                    downstream.onError(new MissingBackpressureException("Couldn't emit value due to lack of requests!"));
+                    downstream.onError(MissingBackpressureException.createDefault());
                 }
             }
         }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSampleTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSampleTimed.java
@@ -125,7 +125,7 @@ void emit() {
                     BackpressureHelper.produced(requested, 1);
                 } else {
                     cancel();
-                    downstream.onError(new MissingBackpressureException("Couldn't emit value due to lack of requests!"));
+                    downstream.onError(MissingBackpressureException.createDefault());
                 }
             }
         }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSequenceEqual.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSequenceEqual.java
@@ -300,7 +300,7 @@ public void onSubscribe(Subscription s) {
         public void onNext(T t) {
             if (sourceMode == QueueSubscription.NONE) {
                 if (!queue.offer(t)) {
-                    onError(new MissingBackpressureException());
+                    onError(MissingBackpressureException.createDefault());
                     return;
                 }
             }

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSwitchMap.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSwitchMap.java
@@ -381,7 +381,7 @@ public void onNext(R t) {
             SwitchMapSubscriber<T, R> p = parent;
             if (index == p.unique) {
                 if (fusionMode == QueueSubscription.NONE && !queue.offer(t)) {
-                    onError(new MissingBackpressureException("Queue full?!"));
+                    onError(new QueueOverflowException());
                     return;
                 }
                 p.drain();

diff --git a/...ain/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleFirstTimed.java b/...ain/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleFirstTimed.java
@@ -96,7 +96,7 @@ public void onNext(T t) {
                 } else {
                     done = true;
                     cancel();
-                    downstream.onError(new MissingBackpressureException("Could not deliver value due to lack of requests"));
+                    downstream.onError(MissingBackpressureException.createDefault());
                     return;
                 }
 

diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableTimer.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableTimer.java
@@ -78,7 +78,7 @@ public void run() {
                     downstream.onComplete();
                 } else {
                     lazySet(EmptyDisposable.INSTANCE);
-                    downstream.onError(new MissingBackpressureException("Can't deliver value due to lack of requests"));
+                    downstream.onError(MissingBackpressureException.createDefault());
                 }
             }
         }