Avoid self-suppression for exceptions (#2211)

Motivation: In case a processing function re-throws the original exception, we risk to face `IllegalArgumentException("Self-suppression not permitted)` when operator adds a re-thrown exception as suppressed. Modifications: - Add utility that verifies the suppressed exception is not the same as the original one; Result: No `IllegalArgumentException("Self-suppression not permitted)`.
apple · May 11, 2022 · 3d26046 · 3d26046
1 parent 4ce81d5
commit 3d26046
Show file tree

Hide file tree

Showing 38 changed files with 97 additions and 61 deletions.
diff --git a/...etalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/AbstractPubToSingle.java b/...etalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/AbstractPubToSingle.java
@@ -27,6 +27,7 @@
 import static io.servicetalk.concurrent.Cancellable.IGNORE_CANCEL;
 import static io.servicetalk.concurrent.api.SubscriberApiUtils.unwrapNullUnchecked;
 import static io.servicetalk.concurrent.internal.SubscriberUtils.checkDuplicateSubscription;
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 
 abstract class AbstractPubToSingle<T> extends AbstractNoHandleSubscribeSingle<T> {
     private final Publisher<T> source;
@@ -110,7 +111,7 @@ void terminate(Object terminal) {
                     subscriber.onSubscribe(IGNORE_CANCEL);
                 } catch (Throwable t) {
                     if (terminal instanceof Throwable) {
-                        ((Throwable) terminal).addSuppressed(t);
+                        addSuppressed((Throwable) terminal, t);
                     } else {
                         LOGGER.warn("Unexpected exception from onSubscribe from subscriber {}. Discarding result {}.",
                                 subscriber, terminal, t);

diff --git a/...-concurrent-api/src/main/java/io/servicetalk/concurrent/api/BeforeFinallyCompletable.java b/...-concurrent-api/src/main/java/io/servicetalk/concurrent/api/BeforeFinallyCompletable.java
@@ -19,6 +19,7 @@
 
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 final class BeforeFinallyCompletable extends AbstractSynchronousCompletableOperator {
@@ -78,8 +79,7 @@ public void onError(Throwable cause) {
                     doFinally.onError(cause);
                 }
             } catch (Throwable error) {
-                error.addSuppressed(cause);
-                original.onError(error);
+                original.onError(addSuppressed(error, cause));
                 return;
             }
             original.onError(cause);

diff --git a/...lk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/BeforeFinallyPublisher.java b/...lk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/BeforeFinallyPublisher.java
@@ -17,6 +17,7 @@
 
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 final class BeforeFinallyPublisher<T> extends AbstractSynchronousPublisherOperator<T, T> {
@@ -93,8 +94,7 @@ public void onError(Throwable cause) {
                     doFinally.onError(cause);
                 }
             } catch (Throwable err) {
-                err.addSuppressed(cause);
-                original.onError(err);
+                original.onError(addSuppressed(err, cause));
                 return;
             }
             original.onError(cause);

diff --git a/...etalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/BeforeFinallySingle.java b/...etalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/BeforeFinallySingle.java
@@ -19,6 +19,7 @@
 
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 final class BeforeFinallySingle<T> extends AbstractSynchronousSingleOperator<T, T> {
@@ -79,8 +80,7 @@ public void onError(Throwable cause) {
                     doFinally.onError(cause);
                 }
             } catch (Throwable err) {
-                err.addSuppressed(cause);
-                original.onError(err);
+                original.onError(addSuppressed(err, cause));
                 return;
             }
             original.onError(cause);

diff --git a/...ncurrent-api/src/main/java/io/servicetalk/concurrent/api/BeforeSubscriberCompletable.java b/...ncurrent-api/src/main/java/io/servicetalk/concurrent/api/BeforeSubscriberCompletable.java
@@ -19,6 +19,7 @@
 
 import java.util.function.Supplier;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 final class BeforeSubscriberCompletable extends AbstractSynchronousCompletableOperator {
@@ -67,8 +68,7 @@ public void onError(Throwable t) {
             try {
                 subscriber.onError(t);
             } catch (Throwable cause) {
-                t.addSuppressed(cause);
-                original.onError(t);
+                original.onError(addSuppressed(t, cause));
                 return;
             }
             original.onError(t);

diff --git a/...concurrent-api/src/main/java/io/servicetalk/concurrent/api/BeforeSubscriberPublisher.java b/...concurrent-api/src/main/java/io/servicetalk/concurrent/api/BeforeSubscriberPublisher.java
@@ -17,6 +17,7 @@
 
 import java.util.function.Supplier;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 final class BeforeSubscriberPublisher<T> extends AbstractSynchronousPublisherOperator<T, T> {
@@ -70,8 +71,7 @@ public void onError(Throwable t) {
             try {
                 subscriber.onError(t);
             } catch (Throwable cause) {
-                t.addSuppressed(cause);
-                original.onError(t);
+                original.onError(addSuppressed(t, cause));
                 return;
             }
             original.onError(t);

diff --git a/...lk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/BeforeSubscriberSingle.java b/...lk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/BeforeSubscriberSingle.java
@@ -19,6 +19,7 @@
 
 import java.util.function.Supplier;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 final class BeforeSubscriberSingle<T> extends AbstractSynchronousSingleOperator<T, T> {
@@ -66,8 +67,7 @@ public void onError(Throwable t) {
             try {
                 subscriber.onError(t);
             } catch (Throwable cause) {
-                t.addSuppressed(cause);
-                original.onError(t);
+                original.onError(addSuppressed(t, cause));
                 return;
             }
             original.onError(t);

diff --git a/...oncurrent-api/src/main/java/io/servicetalk/concurrent/api/CompletableMergeSubscriber.java b/...oncurrent-api/src/main/java/io/servicetalk/concurrent/api/CompletableMergeSubscriber.java
@@ -21,6 +21,7 @@
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import javax.annotation.Nullable;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.concurrent.atomic.AtomicReferenceFieldUpdater.newUpdater;
 
 abstract class CompletableMergeSubscriber implements Subscriber {
@@ -68,9 +69,7 @@ public final void onError(Throwable t) {
                     return;
                 }
             } else {
-                Throwable tmpT = (Throwable) terminalNotification;
-                tmpT.addSuppressed(t);
-                t = tmpT;
+                t = addSuppressed((Throwable) terminalNotification, t);
                 break;
             }
         }

diff --git a/...talk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/CompositeCancellable.java b/...talk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/CompositeCancellable.java
@@ -20,6 +20,7 @@
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
 import static io.servicetalk.utils.internal.PlatformDependent.throwException;
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -56,7 +57,7 @@ public void cancel() {
                     if (t == null) {
                         t = tt;
                     } else {
-                        t.addSuppressed(tt);
+                        addSuppressed(t, tt);
                     }
                 }
             }

diff --git a/...k-concurrent-api/src/main/java/io/servicetalk/concurrent/api/CompositeExceptionUtils.java b/...k-concurrent-api/src/main/java/io/servicetalk/concurrent/api/CompositeExceptionUtils.java
@@ -17,6 +17,8 @@
 
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
+
 final class CompositeExceptionUtils {
     /**
      * Default to {@code 1} so {@link Throwable#addSuppressed(Throwable)} will not be used by default.
@@ -33,7 +35,7 @@ static <T> void addPendingError(AtomicIntegerFieldUpdater<T> updater, T owner, i
         if (newSize < 0) {
             updater.set(owner, Integer.MAX_VALUE);
         } else if (newSize < maxDelayedErrors && original != causeToAdd) {
-            original.addSuppressed(causeToAdd);
+            addSuppressed(original, causeToAdd);
         } else {
             updater.decrementAndGet(owner);
         }

diff --git a/...-concurrent-api/src/main/java/io/servicetalk/concurrent/api/OnErrorResumeCompletable.java b/...-concurrent-api/src/main/java/io/servicetalk/concurrent/api/OnErrorResumeCompletable.java
@@ -23,6 +23,7 @@
 import java.util.function.Predicate;
 import javax.annotation.Nullable;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 final class OnErrorResumeCompletable extends AbstractNoHandleSubscribeCompletable {
@@ -86,8 +87,7 @@ public void onError(Throwable throwable) {
                         requireNonNull(parent.nextFactory.apply(throwable)) :
                         null;
             } catch (Throwable t) {
-                t.addSuppressed(throwable);
-                subscriber.onError(t);
+                subscriber.onError(addSuppressed(t, throwable));
                 return;
             }
 

diff --git a/...lk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/OnErrorResumePublisher.java b/...lk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/OnErrorResumePublisher.java
@@ -21,6 +21,7 @@
 import java.util.function.Predicate;
 import javax.annotation.Nullable;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 final class OnErrorResumePublisher<T> extends AbstractNoHandleSubscribePublisher<T> {
@@ -76,18 +77,17 @@ public void onNext(T t) {
         }
 
         @Override
-        public void onError(Throwable t) {
+        public void onError(Throwable throwable) {
             final Publisher<? extends T> next;
             try {
-                next = !resubscribed && predicate.test(t) ? requireNonNull(nextFactory.apply(t)) : null;
-            } catch (Throwable throwable) {
-                throwable.addSuppressed(t);
-                subscriber.onError(throwable);
+                next = !resubscribed && predicate.test(throwable) ? requireNonNull(nextFactory.apply(throwable)) : null;
+            } catch (Throwable t) {
+                subscriber.onError(addSuppressed(t, throwable));
                 return;
             }
 
             if (next == null) {
-                subscriber.onError(t);
+                subscriber.onError(throwable);
             } else {
                 final Subscriber<? super T> offloadedSubscriber =
                         contextProvider.wrapPublisherSubscriber(this, contextMap);

diff --git a/...etalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/OnErrorResumeSingle.java b/...etalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/OnErrorResumeSingle.java
@@ -23,6 +23,7 @@
 import java.util.function.Predicate;
 import javax.annotation.Nullable;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 final class OnErrorResumeSingle<T> extends AbstractNoHandleSubscribeSingle<T> {
@@ -81,8 +82,7 @@ public void onError(Throwable throwable) {
             try {
                 next = !resubscribed && predicate.test(throwable) ? requireNonNull(nextFactory.apply(throwable)) : null;
             } catch (Throwable t) {
-                t.addSuppressed(throwable);
-                subscriber.onError(t);
+                subscriber.onError(addSuppressed(t, throwable));
                 return;
             }
 

diff --git a/servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/RedoPublisher.java b/servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/RedoPublisher.java
@@ -22,6 +22,7 @@
 import java.util.function.IntPredicate;
 
 import static io.servicetalk.concurrent.internal.TerminalNotification.complete;
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 
 /**
  * {@link Publisher} to do {@link Publisher#repeat(IntPredicate)} and {@link Publisher#retry(BiIntPredicate)}
@@ -116,7 +117,7 @@ private void tryRedo(TerminalNotification notification) {
             } catch (Throwable cause) {
                 Throwable originalCause = notification.cause();
                 if (originalCause != null) {
-                    cause.addSuppressed(originalCause);
+                    addSuppressed(cause, originalCause);
                 }
                 subscriber.onError(cause);
                 return;

diff --git a/...icetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/RedoWhenPublisher.java b/...icetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/RedoWhenPublisher.java
@@ -25,6 +25,7 @@
 import java.util.function.IntFunction;
 
 import static io.servicetalk.concurrent.internal.TerminalNotification.complete;
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -143,7 +144,7 @@ private void redoIfRequired(TerminalNotification terminalNotification) {
             } catch (Throwable cause) {
                 Throwable originalCause = terminalNotification.cause();
                 if (originalCause != null) {
-                    cause.addSuppressed(originalCause);
+                    addSuppressed(cause, originalCause);
                 }
                 subscriber.onError(cause);
                 return;

diff --git a/servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/RetrySingle.java b/servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/RetrySingle.java
@@ -21,6 +21,8 @@
 
 import javax.annotation.Nullable;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
+
 /**
  * A {@link Single} implementation as returned by {@link Single#retry(BiIntPredicate)}.
  *
@@ -97,8 +99,7 @@ public void onError(Throwable t) {
             try {
                 shouldRetry = retrySingle.shouldRetry.test(++retryCount, t);
             } catch (Throwable cause) {
-                cause.addSuppressed(t);
-                target.onError(cause);
+                target.onError(addSuppressed(cause, t));
                 return;
             }
             if (shouldRetry) {

diff --git a/servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/RetryWhenSingle.java b/servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/RetryWhenSingle.java
@@ -22,6 +22,7 @@
 
 import javax.annotation.Nullable;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -104,8 +105,7 @@ public void onError(Throwable t) {
             try {
                 retryDecider = requireNonNull(retrySingle.shouldRetry.apply(++retryCount, t));
             } catch (Throwable cause) {
-                cause.addSuppressed(t);
-                target.onError(cause);
+                target.onError(addSuppressed(cause, t));
                 return;
             }
 

diff --git a/...k-concurrent-api/src/testFixtures/java/io/servicetalk/concurrent/api/TestCompletable.java b/...k-concurrent-api/src/testFixtures/java/io/servicetalk/concurrent/api/TestCompletable.java
@@ -31,6 +31,7 @@
 import javax.annotation.Nullable;
 
 import static io.servicetalk.utils.internal.PlatformDependent.throwException;
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -150,7 +151,7 @@ private Subscriber checkSubscriberAndExceptions() {
             final RuntimeException exception = new RuntimeException("Unexpected exception(s) encountered",
                     exceptions.get(0));
             for (int i = 1; i < exceptions.size(); i++) {
-                exception.addSuppressed(exceptions.get(i));
+                addSuppressed(exception, exceptions.get(i));
             }
             throw exception;
         }

diff --git a/...alk-concurrent-api/src/testFixtures/java/io/servicetalk/concurrent/api/TestPublisher.java b/...alk-concurrent-api/src/testFixtures/java/io/servicetalk/concurrent/api/TestPublisher.java
@@ -30,6 +30,7 @@
 import javax.annotation.Nullable;
 
 import static io.servicetalk.utils.internal.PlatformDependent.throwException;
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -176,7 +177,7 @@ private Subscriber<? super T> checkSubscriberAndExceptions() {
             final RuntimeException exception = new RuntimeException("Unexpected exception(s) encountered",
                     exceptions.get(0));
             for (int i = 1; i < exceptions.size(); i++) {
-                exception.addSuppressed(exceptions.get(i));
+                addSuppressed(exception, exceptions.get(i));
             }
             throw exception;
         }

diff --git a/...cetalk-concurrent-api/src/testFixtures/java/io/servicetalk/concurrent/api/TestSingle.java b/...cetalk-concurrent-api/src/testFixtures/java/io/servicetalk/concurrent/api/TestSingle.java
@@ -31,6 +31,7 @@
 
 import static io.servicetalk.concurrent.test.internal.AwaitUtils.await;
 import static io.servicetalk.utils.internal.PlatformDependent.throwException;
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -153,7 +154,7 @@ private Subscriber<? super T> checkSubscriberAndExceptions() {
             final RuntimeException exception = new RuntimeException("Unexpected exception(s) encountered",
                     exceptions.get(0));
             for (int i = 1; i < exceptions.size(); i++) {
-                exception.addSuppressed(exceptions.get(i));
+                addSuppressed(exception, exceptions.get(i));
             }
             throw exception;
         }

diff --git a/...rrent-internal/src/main/java/io/servicetalk/concurrent/internal/TerminalNotification.java b/...rrent-internal/src/main/java/io/servicetalk/concurrent/internal/TerminalNotification.java
@@ -21,6 +21,7 @@
 import java.util.Objects;
 import javax.annotation.Nullable;
 
+import static io.servicetalk.utils.internal.ThrowableUtils.addSuppressed;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -80,8 +81,7 @@ public void terminate(PublisherSource.Subscriber<?> subscriber, Throwable additi
             subscriber.onError(additionalCause);
         } else {
             assert cause != null;
-            cause.addSuppressed(additionalCause);
-            subscriber.onError(cause);
+            subscriber.onError(addSuppressed(cause, additionalCause));
         }
     }
 
@@ -102,8 +102,7 @@ public void terminate(Subscriber subscriber, Throwable additionalCause) {
             subscriber.onError(additionalCause);
         } else {
             assert cause != null;
-            cause.addSuppressed(additionalCause);
-            subscriber.onError(cause);
+            subscriber.onError(addSuppressed(cause, additionalCause));
         }
     }