3.x: Standardize MissingBackpressureException message, introduce QueueOverflowException (#7459)

* 3.x: Standardize MissingBackpressureException messages

* Use the correct message.

* Fix tests

* Add QueueOverflowException, fix tests

* Update CompletableConcatTest.java

Author: akarnokd

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);
+    }
 }

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);
+    }
+}

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;
                 }
             }

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;
                 }
             }

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();
             }

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;
                 }
             }

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());
             }
         }
 

src/main/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;
                 }
             }

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));
         }
 
     }

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());
                 }
             }
         }