Improve CancellationException (#3039)

Motivation:

1. `H2ClientParentConnectionContext.StacklessCancellationException` was reused by `SpliceFlatStreamToMetaSingle`, which created a false impression in logs that this exception is related to HTTP/2.
2. We should capture the Subscriber's stack-trace for `CancellationException` in `BeforeFinallyHttpOperator` instead of a netty stack-trace, it will help to debug paths that subscribe to the payload post cancel.

Modifications:

1. Make `H2ClientParentConnectionContext.StacklessCancellationException` a top-level class.
2. Wrap `CancellationException` with `Publisher.defer(...)` in `BeforeFinallyHttpOperator`.
3. Add an exception message in `DefaultBlockingIterableProcessor`.

Result:

Easier for users to reason about received `CancellationException`.

servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/DefaultBlockingIterableProcessor.java

@@ -103,7 +103,7 @@ public T next() {
 
         @Override
         public void close() {
-            terminal = TerminalNotification.error(new CancellationException());
+            terminal = TerminalNotification.error(new CancellationException("BlockingIterator closed"));
         }
 
         @Override

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ClientParentConnectionContext.java

@@ -26,7 +26,6 @@
 import io.servicetalk.concurrent.api.internal.SubscribableSingle;
 import io.servicetalk.concurrent.internal.DelayedCancellable;
 import io.servicetalk.concurrent.internal.SequentialCancellable;
-import io.servicetalk.concurrent.internal.ThrowableUtils;
 import io.servicetalk.http.api.FilterableStreamingHttpConnection;
 import io.servicetalk.http.api.HttpConnectionContext;
 import io.servicetalk.http.api.HttpEventKey;
@@ -69,7 +68,6 @@
 
 import java.net.SocketAddress;
 import java.net.SocketOption;
-import java.util.concurrent.CancellationException;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import javax.annotation.Nullable;
 import javax.net.ssl.SSLSession;
@@ -561,23 +559,4 @@ public String toString() {
                     '}';
         }
     }
-
-    static final class StacklessCancellationException extends CancellationException {
-        private static final long serialVersionUID = 3235852873427231209L;
-
-        private StacklessCancellationException(String message) {
-            super(message);
-        }
-
-        // Override fillInStackTrace() so we not populate the backtrace via a native call and so leak the
-        // Classloader.
-        @Override
-        public Throwable fillInStackTrace() {
-            return this;
-        }
-
-        static StacklessCancellationException newInstance(String message, Class<?> clazz, String method) {
-            return ThrowableUtils.unknownStackTrace(new StacklessCancellationException(message), clazz, method);
-        }
-    }
 }

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/SpliceFlatStreamToMetaSingle.java

@@ -28,7 +28,6 @@
 import io.servicetalk.concurrent.internal.DuplicateSubscribeException;
 import io.servicetalk.http.api.HttpResponseMetaData;
 import io.servicetalk.http.api.StreamingHttpResponse;
-import io.servicetalk.http.netty.H2ClientParentConnectionContext.StacklessCancellationException;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/StacklessCancellationException.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright © 2024 Apple Inc. and the ServiceTalk project authors
+ *
+ * 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.servicetalk.http.netty;
+
+import io.servicetalk.concurrent.internal.ThrowableUtils;
+
+import java.util.concurrent.CancellationException;
+
+final class StacklessCancellationException extends CancellationException {
+    private static final long serialVersionUID = 5434529104526587317L;
+
+    private StacklessCancellationException(String message) {
+        super(message);
+    }
+
+    @Override
+    public Throwable fillInStackTrace() {
+        return this;
+    }
+
+    static StacklessCancellationException newInstance(String message, Class<?> clazz, String method) {
+        return ThrowableUtils.unknownStackTrace(new StacklessCancellationException(message), clazz, method);
+    }
+}

servicetalk-http-utils/src/main/java/io/servicetalk/http/utils/BeforeFinallyHttpOperator.java

@@ -174,8 +174,9 @@ public void onSuccess(@Nullable final StreamingHttpResponse response) {
                 // body or risk leaking hot resources which are commonly attached to a message body.
                 toSource(response.messageBody()).subscribe(CancelImmediatelySubscriber.INSTANCE);
                 if (!discardEventsAfterCancel) {
-                    subscriber.onSuccess(response.transformMessageBody(payload ->
-                        Publisher.failed(new CancellationException("Received response post cancel."))));
+                    // Wrap with defer to capture the Subscriber's stack-trace
+                    subscriber.onSuccess(response.transformMessageBody(payload -> Publisher.defer(() ->
+                            Publisher.failed(new CancellationException("Received response post cancel")))));
                 }
             }
             dereferenceSubscriber();