Repository: kroxylicious/kroxylicious (java)

Issue:
Async chaining methods #xxxxAsync of InternalCompletionStage run on JDK thread rather than Netty thread
The filter documentation includes the following words:

https://github.com/kroxylicious/kroxylicious/blob/b255d2e6cd0400ed973c3bb60ff1173cdff00a6d/docs/custom-filters.adoc?plain=1#L441C110-L443C15

```
Note that this stage is completed on the same thread as the rest of the Filter work, so we can safely mutate Filter members without synchronising.
```

Whilst this is true in the case of *non-async* chaining (`thenApply` etc) to`context.sendRequest`, in the case on asynchronous chaining (`thenApplyAsync`), this is not the case.  If the user were to try to update filter members from with such an action, that update will not be published safely between threads.

###  *non-async* chaining (safe publication)
```java
        return context.sendRequest(version, ....)
                      .thenCompose(asyncResponse -> {
                           // This is safe - we are on netty thread
                           this.myFilterList.add(..)
                           return context.forwardRequest(header, request));
                      }:
    }
```

###  async chaining (unsafe publication)
```java
        return context.sendRequest(version, ....)
                      .thenComposeAsync(resp -> {
                           // Unsafe - we are on a ForkJoinPool thread
                           this.myFilterList.add(..)
                           return context.forwardRequest(header, request));
                      }:
    }
```







Pull request description:
Fix #537: Ensure actions chained to the #sendRequest run on netty loop executor
### Type of change

_Select the type of your PR_

- Bugfix

### Description

Actions chained to the completion stage returned by sendRequest using the `*Async()` methods should be run on Netty event loop, rather than the JDK's default Fork-Join pool.  This will allow the user to write code that updates the state of the filter with thread safety.

This PR replaces #550. I was going the wrong way.

### Additional Context


### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [x] Write tests
- [x] Make sure all tests pass
- [ ] Review performance test results. Ensure that any degradations to performance numbers are understood and justified.
- [ ] Make sure all Sonar-Lint warnings are addressed or are justifiably ignored.
- [x] Update documentation
- [x] Reference relevant issue(s) and close them after merging
- [x] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index fa2c0b70..98fef49e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,8 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 ## 1.0.0
 
 ## 0.3.0
+
+* [#537](https://github.com/kroxylicious/kroxylicious/issues/537): Computation stages chained to the CompletionStage return by #sendRequest using the default executor async methods now run on the Netty Event Loop.
 * [#612](https://github.com/kroxylicious/kroxylicious/pull/612): [Breaking] Allow filter authors to declare when their filter requires configuration. Note this includes a backwards incompatible change to the contract of the `Contributor`. `getInstance` will now throw exceptions rather than returning `null` to mean there was a problem or this contributor does not know about the requested type.
 * [#608](https://github.com/kroxylicious/kroxylicious/pull/608): Improve the contributor API to allow it to express more properties about the configuration. This release deprecates `Contributor.getConfigType` in favour of `Contributor.getConfigDefinition`. It also removes the proliferation of ContributorManager classes by providing a single type which can handle all Contributors.
 * [#538](https://github.com/kroxylicious/kroxylicious/pull/538): Refactor FilterHandler and fix several bugs that would leave messages unflushed to client/broker.
@@ -51,4 +53,4 @@ In the kroxylicious config, the brokerAddressPattern parameter for the PortPerBr
 present on the brokerAddressPattern parameter. Previously, it was accepted but would lead to a failure later.
 
 Kroxylicious configuration no longer requires a non empty `filters` list, users can leave it unset or configure in an empty
-list of filters and Kroxylicious will proxy to the cluster successfully.
\ No newline at end of file
+list of filters and Kroxylicious will proxy to the cluster successfully.
diff --git a/docs/custom-filters.adoc b/docs/custom-filters.adoc
index 83da777b..fd69f14c 100644
--- a/docs/custom-filters.adoc
+++ b/docs/custom-filters.adoc
@@ -292,6 +292,9 @@ The proxy guarantees that all methods of a given filter instance will always be
 the CompletionStage completion in the case of <<Sending asynchronous requests to the Cluster>>).
 Therefore, there is no need to use synchronization when accessing such fields.
 
+See the {api-javadoc}/io/kroxylicious/proxy/filter/package-summary.html#implementing.threadSafety[`io.kroxylicious.proxy.filter`]
+package javadoc for more information on thread-safety.
+
 === Filter Patterns
 
 Kroxylicious Protocol Filters support several patterns:
@@ -453,11 +456,13 @@ public class FetchFilter implements FetchRequestFilter {
 <1> We construct a new request object.  This will be forwarded asynchronously.
 <2> We send the request towards the Cluster, specifying the api version to use and obtaining a CompletionStage for the
 response.
-<3> We use CompletionStage chaining to mutate the filtered fetch request using the response from the request sent at <1>.
-<4> We use CompletionStage chaining to forward the mutated request.
+<3> We use a computation stage to mutate the filtered fetch request using the response from the request sent at <1>.
+<4> We use another computation stage to forward the mutated request.
 
-NOTE: Kroxylicious provides the guarantee that the chained stage is completed on the same thread as the rest of the
-Filter work, so we can safely mutate Filter members without synchronising.
+NOTE: Kroxylicious provides the guarantee that computation stages chained using the _default execution methods_ are
+executed on the same thread as the rest of the Filter work, so we can safely mutate Filter members without synchronising.
+See the {api-javadoc}/io/kroxylicious/proxy/filter/package-summary.html#implementing.threadSafety[`io.kroxylicious.proxy.filter`]
+package javadoc for more information on thread-safety.
 
 As you can see, we need to know the API version we want our message to be encoded at. This must be done carefully as we
 currently do not have a mechanism for your Filter to know the supported API versions of the Cluster. If
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContext.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContext.java
index 918113b5..1702d74c 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContext.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContext.java
@@ -83,11 +83,15 @@ public interface FilterContext {
      * of the filter that invoked the operation, invoking them, but not itself. The response does not
      * pass through filter downstream.
      * <br/>
+     * Default and asynchronous default computation stages chained to the returned
+     * {@link java.util.concurrent.CompletionStage} are guaranteed to be executed by the thread associated with the
+     * connection. See {@link io.kroxylicious.proxy.filter} for more details.
      *
      * @param apiVersion The version of the request to use
      * @param request The request to send.
      * @param <T> The type of the response
      * @return CompletionStage that will yield the response.
+     * @see io.kroxylicious.proxy.filter Thread Safety
      */
     @NonNull
     <T extends ApiMessage> CompletionStage<T> sendRequest(short apiVersion, @NonNull ApiMessage request);
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilter.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilter.java
index 1e906f83..c56cb994 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilter.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilter.java
@@ -49,6 +49,7 @@ public interface RequestFilter extends Filter {
      * @return a non-null CompletionStage that, when complete, will yield a RequestFilterResult containing the
      *         request to be forwarded.
      * @see io.kroxylicious.proxy.filter Creating Filter Result objects
+     * @see io.kroxylicious.proxy.filter Thread Safety
      */
     CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey,
                                                    RequestHeaderData header,
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilter.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilter.java
index 2017f2d6..f7725e22 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilter.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilter.java
@@ -49,6 +49,7 @@ public interface ResponseFilter extends Filter {
      * @return a non-null CompletionStage that, when complete, will yield a ResponseFilterResult containing the
      *         response to be forwarded.
      * @see io.kroxylicious.proxy.filter Creating Filter Result objects
+     * @see io.kroxylicious.proxy.filter Thread Safety
      */
     CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey,
                                                      ResponseHeaderData header,
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/package-info.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/package-info.java
index fe38eedc..e9c7013f 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/package-info.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/package-info.java
@@ -38,5 +38,23 @@
  *     <li>{@link io.kroxylicious.proxy.filter.FilterContext#requestFilterResultBuilder()} and</li>
  *     <li>{@link io.kroxylicious.proxy.filter.FilterContext#responseFilterResultBuilder()}.</li>
  * </ul>
+ * <h3 id='implementing.threadSafety'>Thread Safety</h3>
+ * <p>
+ * The Filter API provides the following thread-safety guarantees:
+ * </p>
+ * <ol>
+ *   <li>There is a single thread associated with each connection and this association lasts for the lifetime of connection..</li>
+ *   <li>Each filter instance is associated with exactly one connection.</li>
+ *   <li>Construction of the filter instance and dispatch of the filter methods {@code onXxxRequest} and
+ *       {@code onXxxResponse} takes place on that same thread.</li>
+ *   <li>Any computation stages chained to the {@link java.util.concurrent.CompletionStage} returned by
+ *       {@link io.kroxylicious.proxy.filter.FilterContext#sendRequest(short, org.apache.kafka.common.protocol.ApiMessage)}
+ *       using the default execution methods (using methods without the suffix async) or default asynchronous execution
+ *       (using methods with suffix async that employ the stage's default asynchronous execution facility)
+ *       are guaranteed to be performed by that same thread.  Computation stages chained using custom asynchronous
+ *       execution (using methods with suffix async that take an Executor argument) do not get this guarantee.</li>
+ *  </ol>
+ *  <p>Filter implementations are free to rely on these guarantees to safely maintain state within fields
+ *     of the Filter without employing additional synchronization.</p>
  */
 package io.kroxylicious.proxy.filter;
diff --git a/kroxylicious-api/src/main/templates/Kproxy/Filter.ftl b/kroxylicious-api/src/main/templates/Kproxy/Filter.ftl
index 39b8110b..243d5b03 100644
--- a/kroxylicious-api/src/main/templates/Kproxy/Filter.ftl
+++ b/kroxylicious-api/src/main/templates/Kproxy/Filter.ftl
@@ -66,6 +66,7 @@ public interface ${filterClass} extends Filter {
      * @return a non-null CompletionStage that, when complete, will yield a ${filterResultClass} containing the
      *         ${msgType} to be forwarded.
      * @see io.kroxylicious.proxy.filter Creating Filter Result objects
+     * @see io.kroxylicious.proxy.filter  Thread Safety
      */
      CompletionStage<${filterResultClass}> on${messageSpec.name}(short apiVersion, ${headerClass} header, ${dataClass} ${msgType}, FilterContext context);
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
index 95319bae..dc8642cb 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
@@ -46,7 +46,6 @@ import io.kroxylicious.proxy.frame.DecodedResponseFrame;
 import io.kroxylicious.proxy.frame.OpaqueRequestFrame;
 import io.kroxylicious.proxy.frame.OpaqueResponseFrame;
 import io.kroxylicious.proxy.frame.RequestFrame;
-import io.kroxylicious.proxy.future.InternalCompletionStage;
 import io.kroxylicious.proxy.internal.filter.RequestFilterResultBuilderImpl;
 import io.kroxylicious.proxy.internal.filter.ResponseFilterResultBuilderImpl;
 import io.kroxylicious.proxy.internal.util.Assertions;
@@ -196,7 +195,8 @@ public class FilterHandler extends ChannelDuplexHandler {
             closeConnection();
             return CompletableFuture.completedFuture(null);
         }
-        return stage.toCompletableFuture();
+        return stage instanceof InternalCompletionStage ? ((InternalCompletionStage<ResponseFilterResult>) stage).getUnderlyingCompletableFuture()
+                : stage.toCompletableFuture();
     }
 
     private CompletableFuture<ResponseFilterResult> configureResponseFilterChain(DecodedResponseFrame<?> decodedFrame, CompletableFuture<ResponseFilterResult> future) {
@@ -236,7 +236,8 @@ public class FilterHandler extends ChannelDuplexHandler {
             closeConnection();
             return CompletableFuture.completedFuture(null);
         }
-        return stage.toCompletableFuture();
+        return stage instanceof InternalCompletionStage ? ((InternalCompletionStage<RequestFilterResult>) stage).getUnderlyingCompletableFuture()
+                : stage.toCompletableFuture();
     }
 
     private CompletableFuture<RequestFilterResult> configureRequestFilterChain(DecodedRequestFrame<?> decodedFrame, ChannelPromise promise,
@@ -503,8 +504,7 @@ public class FilterHandler extends ChannelDuplexHandler {
             }
             boolean hasResponse = apiKey != ApiKeys.PRODUCE
                     || ((ProduceRequestData) request).acks() != 0;
-            var filterPromise = new CompletableFuture<T>();
-            var filterStage = new InternalCompletionStage<>(filterPromise);
+            var filterPromise = new InternalCompletableFuture<T>(ctx.executor());
             var frame = new InternalRequestFrame<>(
                     apiVersion, -1, hasResponse,
                     filter, filterPromise, header, request);
@@ -535,7 +535,7 @@ public class FilterHandler extends ChannelDuplexHandler {
                 filterPromise
                         .completeExceptionally(new TimeoutException("Asynchronous %s request made by filter %s was timed-out.".formatted(apiKey, filterDescriptor())));
             }, timeoutMs, TimeUnit.MILLISECONDS);
-            return filterStage;
+            return filterPromise.minimalCompletionStage();
         }
 
         @Override
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/InternalCompletableFuture.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/InternalCompletableFuture.java
new file mode 100644
index 00000000..6e29367f
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/InternalCompletableFuture.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.internal;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executor;
+
+/**
+ * Implementation of CompletableFuture that uses the given {@link Executor}
+ * for async method invocations that do not specify one.
+ * <br/>
+ * @param <T> The result type returned by this future's {@code join}
+ * and {@code get} methods.
+ * @see InternalCompletionStage
+ */
+class InternalCompletableFuture<T> extends CompletableFuture<T> {
+
+    private final Executor executor;
+
+    InternalCompletableFuture(Executor executor) {
+        this.executor = Objects.requireNonNull(executor);
+    }
+
+    /**
+     * Returns a new incomplete InternalCompletableFuture of the type to be
+     * returned by a CompletionStage method.
+     *
+     * @param <U> the type of the value
+     * @return a new CompletableFuture
+     */
+    @Override
+    public <U> CompletableFuture<U> newIncompleteFuture() {
+        return new InternalCompletableFuture<>(executor);
+    }
+
+    /**
+     * Returns the default Executor used for async methods that do not specify an Executor.
+     *
+     * @return the default executor
+     */
+    @Override
+    public Executor defaultExecutor() {
+        return executor;
+    }
+
+    /**
+     * Returns a new CompletionStage that is completed normally with
+     * the same value as this CompletableFuture when it completes
+     * normally, and cannot be independently completed or otherwise
+     * used in ways not defined by the methods of interface {@link
+     * CompletionStage}.  If this CompletableFuture completes
+     * exceptionally, then the returned CompletionStage completes
+     * exceptionally with a CompletionException with this exception as
+     * cause.
+     * <br/>
+     * The return CompletionStage implementation disallows the use
+     * of {@link #toCompletableFuture()} ()}
+     */
+    @Override
+    public CompletionStage<T> minimalCompletionStage() {
+        return new InternalCompletionStage<>(this);
+    }
+
+    /**
+     * Returns a new CompletableFuture that is already completed with the given value.
+     * @param value – the value
+     * @param <U> the type of the value
+     * @return the completed CompletableFuture
+     */
+    public static <U> CompletableFuture<U> completedFuture(Executor executor, U value) {
+        var f = new InternalCompletableFuture<U>(executor);
+        f.complete(value);
+        return f;
+    }
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/future/InternalCompletionStage.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/InternalCompletionStage.java
similarity index 66%
rename from kroxylicious/src/main/java/io/kroxylicious/proxy/future/InternalCompletionStage.java
rename to kroxylicious/src/main/java/io/kroxylicious/proxy/internal/InternalCompletionStage.java
index e6ee5ed5..3bfea6c5 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/future/InternalCompletionStage.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/InternalCompletionStage.java
@@ -3,7 +3,7 @@
  *
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
-package io.kroxylicious.proxy.future;
+package io.kroxylicious.proxy.internal;
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
@@ -18,226 +18,234 @@ import java.util.function.Function;
  * to block the proxy thread loop using the CompletionStage we offer though the kroxylicious
  * filter api, or complete the underlying future unexpectedly.
  */
-public class InternalCompletionStage<T> implements CompletionStage<T> {
+class InternalCompletionStage<T> implements CompletionStage<T> {
 
     private final CompletionStage<T> completionStage;
 
-    public InternalCompletionStage(CompletionStage<T> completionStage) {
+    InternalCompletionStage(CompletionStage<T> completionStage) {
         this.completionStage = completionStage;
     }
 
+    private <U> CompletionStage<U> wrap(CompletionStage<U> completionStage) {
+        return completionStage instanceof InternalCompletionStage ? completionStage : new InternalCompletionStage<>(completionStage);
+    }
+
     @Override
     public <U> CompletionStage<U> thenApply(Function<? super T, ? extends U> fn) {
-        return completionStage.thenApply(fn);
+        return wrap(completionStage.thenApply(fn));
     }
 
     @Override
     public <U> CompletionStage<U> thenApplyAsync(Function<? super T, ? extends U> fn) {
-        return completionStage.thenApplyAsync(fn);
+        return wrap(completionStage.thenApplyAsync(fn));
     }
 
     @Override
     public <U> CompletionStage<U> thenApplyAsync(Function<? super T, ? extends U> fn, Executor executor) {
-        return completionStage.thenApplyAsync(fn, executor);
+        return wrap(completionStage.thenApplyAsync(fn, executor));
     }
 
     @Override
     public CompletionStage<Void> thenAccept(Consumer<? super T> action) {
-        return completionStage.thenAccept(action);
+        return wrap(completionStage.thenAccept(action));
     }
 
     @Override
     public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action) {
-        return completionStage.thenAcceptAsync(action);
+        return wrap(completionStage.thenAcceptAsync(action));
     }
 
     @Override
     public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor) {
-        return completionStage.thenAcceptAsync(action, executor);
+        return wrap(completionStage.thenAcceptAsync(action, executor));
     }
 
     @Override
     public CompletionStage<Void> thenRun(Runnable action) {
-        return completionStage.thenRun(action);
+        return wrap(completionStage.thenRun(action));
     }
 
     @Override
     public CompletionStage<Void> thenRunAsync(Runnable action) {
-        return completionStage.thenRunAsync(action);
+        return wrap(completionStage.thenRunAsync(action));
     }
 
     @Override
     public CompletionStage<Void> thenRunAsync(Runnable action, Executor executor) {
-        return completionStage.thenRunAsync(action, executor);
+        return wrap(completionStage.thenRunAsync(action, executor));
     }
 
     @Override
     public <U, V> CompletionStage<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn) {
-        return completionStage.thenCombine(other, fn);
+        return wrap(completionStage.thenCombine(other, fn));
     }
 
     @Override
     public <U, V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn) {
-        return completionStage.thenCombineAsync(other, fn);
+        return wrap(completionStage.thenCombineAsync(other, fn));
     }
 
     @Override
     public <U, V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn, Executor executor) {
-        return completionStage.thenCombineAsync(other, fn, executor);
+        return wrap(completionStage.thenCombineAsync(other, fn, executor));
     }
 
     @Override
     public <U> CompletionStage<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) {
-        return completionStage.thenAcceptBoth(other, action);
+        return wrap(completionStage.thenAcceptBoth(other, action));
     }
 
     @Override
     public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) {
-        return completionStage.thenAcceptBothAsync(other, action);
+        return wrap(completionStage.thenAcceptBothAsync(other, action));
     }
 
     @Override
     public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor) {
-        return completionStage.thenAcceptBothAsync(other, action, executor);
+        return wrap(completionStage.thenAcceptBothAsync(other, action, executor));
     }
 
     @Override
     public CompletionStage<Void> runAfterBoth(CompletionStage<?> other, Runnable action) {
-        return completionStage.runAfterBoth(other, action);
+        return wrap(completionStage.runAfterBoth(other, action));
     }
 
     @Override
     public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action) {
-        return completionStage.runAfterBothAsync(other, action);
+        return wrap(completionStage.runAfterBothAsync(other, action));
     }
 
     @Override
     public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor) {
-        return completionStage.runAfterBothAsync(other, action, executor);
+        return wrap(completionStage.runAfterBothAsync(other, action, executor));
     }
 
     @Override
     public <U> CompletionStage<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn) {
-        return completionStage.applyToEither(other, fn);
+        return wrap(completionStage.applyToEither(other, fn));
     }
 
     @Override
     public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn) {
-        return completionStage.applyToEitherAsync(other, fn);
+        return wrap(completionStage.applyToEitherAsync(other, fn));
     }
 
     @Override
     public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor) {
-        return completionStage.applyToEitherAsync(other, fn, executor);
+        return wrap(completionStage.applyToEitherAsync(other, fn, executor));
     }
 
     @Override
     public CompletionStage<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) {
-        return completionStage.acceptEither(other, action);
+        return wrap(completionStage.acceptEither(other, action));
     }
 
     @Override
     public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) {
-        return completionStage.acceptEitherAsync(other, action);
+        return wrap(completionStage.acceptEitherAsync(other, action));
     }
 
     @Override
     public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor) {
-        return completionStage.acceptEitherAsync(other, action, executor);
+        return wrap(completionStage.acceptEitherAsync(other, action, executor));
     }
 
     @Override
     public CompletionStage<Void> runAfterEither(CompletionStage<?> other, Runnable action) {
-        return completionStage.runAfterEither(other, action);
+        return wrap(completionStage.runAfterEither(other, action));
     }
 
     @Override
     public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action) {
-        return completionStage.runAfterEitherAsync(other, action);
+        return wrap(completionStage.runAfterEitherAsync(other, action));
     }
 
     @Override
     public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor) {
-        return completionStage.runAfterEitherAsync(other, action, executor);
+        return wrap(completionStage.runAfterEitherAsync(other, action, executor));
     }
 
     @Override
     public <U> CompletionStage<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) {
-        return completionStage.thenCompose(fn);
+        return wrap(completionStage.thenCompose(fn));
     }
 
     @Override
     public <U> CompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) {
-        return completionStage.thenComposeAsync(fn);
+        return wrap(completionStage.thenComposeAsync(fn));
     }
 
     @Override
     public <U> CompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor) {
-        return completionStage.thenComposeAsync(fn, executor);
+        return wrap(completionStage.thenComposeAsync(fn, executor));
     }
 
     @Override
     public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) {
-        return completionStage.handle(fn);
+        return wrap(completionStage.handle(fn));
     }
 
     @Override
     public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) {
-        return completionStage.handleAsync(fn);
+        return wrap(completionStage.handleAsync(fn));
     }
 
     @Override
     public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
-        return completionStage.handleAsync(fn, executor);
+        return wrap(completionStage.handleAsync(fn, executor));
     }
 
     @Override
     public CompletionStage<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) {
-        return completionStage.whenComplete(action);
+        return wrap(completionStage.whenComplete(action));
     }
 
     @Override
     public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) {
-        return completionStage.whenCompleteAsync(action);
+        return wrap(completionStage.whenCompleteAsync(action));
     }
 
     @Override
     public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) {
-        return completionStage.whenCompleteAsync(action, executor);
+        return wrap(completionStage.whenCompleteAsync(action, executor));
     }
 
     @Override
     public CompletionStage<T> exceptionally(Function<Throwable, ? extends T> fn) {
-        return completionStage.exceptionally(fn);
+        return wrap(completionStage.exceptionally(fn));
     }
 
     @Override
     public CompletionStage<T> exceptionallyAsync(Function<Throwable, ? extends T> fn) {
-        return completionStage.exceptionallyAsync(fn);
+        return wrap(completionStage.exceptionallyAsync(fn));
     }
 
     @Override
     public CompletionStage<T> exceptionallyAsync(Function<Throwable, ? extends T> fn, Executor executor) {
-        return completionStage.exceptionallyAsync(fn, executor);
+        return wrap(completionStage.exceptionallyAsync(fn, executor));
     }
 
     @Override
     public CompletionStage<T> exceptionallyCompose(Function<Throwable, ? extends CompletionStage<T>> fn) {
-        return completionStage.exceptionallyCompose(fn);
+        return wrap(completionStage.exceptionallyCompose(fn));
     }
 
     @Override
     public CompletionStage<T> exceptionallyComposeAsync(Function<Throwable, ? extends CompletionStage<T>> fn) {
-        return completionStage.exceptionallyComposeAsync(fn);
+        return wrap(completionStage.exceptionallyComposeAsync(fn));
     }
 
     @Override
     public CompletionStage<T> exceptionallyComposeAsync(Function<Throwable, ? extends CompletionStage<T>> fn, Executor executor) {
-        return completionStage.exceptionallyComposeAsync(fn, executor);
+        return wrap(completionStage.exceptionallyComposeAsync(fn, executor));
     }
 
     @Override
     public CompletableFuture<T> toCompletableFuture() {
         throw new UnsupportedOperationException("CompletableFuture usage disallowed, we don't want to block the event loop or allow unexpected completion");
     }
+
+    CompletableFuture<T> getUnderlyingCompletableFuture() {
+        return completionStage.toCompletableFuture();
+    }
 }


Test patch:
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
index ca2b3fef..4d32e3dc 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
@@ -12,6 +12,7 @@ import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -43,7 +44,6 @@ import io.kroxylicious.proxy.frame.DecodedRequestFrame;
 import io.kroxylicious.proxy.frame.DecodedResponseFrame;
 import io.kroxylicious.proxy.frame.OpaqueRequestFrame;
 import io.kroxylicious.proxy.frame.OpaqueResponseFrame;
-import io.kroxylicious.proxy.future.InternalCompletionStage;
 import io.kroxylicious.proxy.internal.filter.RequestFilterResultBuilderImpl;
 import io.kroxylicious.proxy.internal.filter.ResponseFilterResultBuilderImpl;
 
@@ -778,6 +778,52 @@ public class FilterHandlerTest extends FilterHarness {
         assertThrows(ExecutionException.class, q::get);
     }
 
+    @Test
+    void sendRequestChainedActionsRunOnNettyEventLoop() {
+        var eventLoopThreadFuture = new CompletableFuture<Thread>();
+
+        var asyncRequestBody = new FetchRequestData();
+        var applyActionThread = new AtomicReference<Thread>();
+        var applyAsyncActionThread = new AtomicReference<Thread>();
+        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> context.sendRequest((short) 3, asyncRequestBody)
+                .thenApply(u1 -> {
+                    applyActionThread.set(Thread.currentThread());
+                    return null;
+                }).thenApplyAsync(u2 -> {
+                    applyAsyncActionThread.set(Thread.currentThread());
+                    return null;
+                }).thenCompose(u3 -> context.forwardRequest(header, request));
+
+        buildChannel(filter);
+
+        // capture the thread used by the embedded channel
+        channel.eventLoop().submit(() -> eventLoopThreadFuture.complete(Thread.currentThread()));
+        channel.runPendingTasks();
+        assertThat(eventLoopThreadFuture).isCompleted();
+
+        var frame = writeRequest(new ApiVersionsRequestData());
+
+        // Process the async request and write response
+        var propagated = channel.readOutbound();
+        writeInternalResponse(new FetchResponseData(), ((InternalRequestFrame<?>) propagated).promise());
+
+        // Running the tasks will run the actions chained to the async response
+        channel.runPendingTasks();
+
+        // Verify actions ran on the expected thread.
+        assertThat(applyActionThread)
+                .describedAs("first chained action (apply) must run on event loop")
+                .hasValue(eventLoopThreadFuture.getNow(null));
+
+        assertThat(applyAsyncActionThread)
+                .describedAs("second chained action (applySync) must run on event loop")
+                .hasValue(eventLoopThreadFuture.getNow(null));
+
+        // Verify the filtered request arrived at outcome.
+        propagated = channel.readOutbound();
+        assertThat(propagated).isEqualTo(frame);
+    }
+
     private static void assertResponseMessageTaggedWith(String filterName, DecodedResponseFrame<?> propagated) {
         String tag = collectTagsToStrings(propagated.body(), ARBITRARY_TAG);
         assertEquals(tag, filterName);
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/InternalCompletableFutureTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/InternalCompletableFutureTest.java
new file mode 100644
index 00000000..10c7bf69
--- /dev/null
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/InternalCompletableFutureTest.java
@@ -0,0 +1,164 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.internal;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiFunction;
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class InternalCompletableFutureTest {
+
+    private ExecutorService executor;
+
+    @BeforeEach
+    void beforeEach() {
+        executor = Executors.newSingleThreadExecutor();
+    }
+
+    @AfterEach
+    void afterEach() {
+        executor.shutdown();
+    }
+
+    @Test
+    void asyncChainingMethodExecutesOnThreadOfExecutor() throws Exception {
+        var threadOfExecutor = executor.submit(Thread::currentThread).get();
+        var future = InternalCompletableFuture.completedFuture(executor, null);
+
+        var actualThread = new AtomicReference<Thread>();
+        var result = future.thenAcceptAsync((u) -> {
+            assertThat(actualThread).hasValue(null);
+            actualThread.set(Thread.currentThread());
+        });
+        result.join();
+
+        assertThat(result).isCompleted();
+        assertThat(actualThread).hasValue(threadOfExecutor);
+
+    }
+
+    @Test
+    void minimalStageDisallowsToCompletableFuture() {
+        var future = new InternalCompletableFuture<Void>(executor);
+        var stage = future.minimalCompletionStage();
+        assertThatThrownBy(stage::toCompletableFuture).isInstanceOf(UnsupportedOperationException.class);
+    }
+
+    static Stream<Arguments> chainingMethodsReturnValueDisallowToCompletableFuture() {
+        var other = CompletableFuture.<Void> completedFuture(null);
+        var completed = CompletableFuture.<Void> completedFuture(null);
+        return Stream.of(
+                Arguments.of("thenAccept", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenAccept(u -> {
+                })),
+                Arguments.of("thenAcceptAsync", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenAcceptAsync(u -> {
+                })),
+                Arguments.of("thenAcceptAsync(E)", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenAcceptAsync(u -> {
+                }, e)),
+
+                Arguments.of("thenApply", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenApply(u -> u)),
+                Arguments.of("thenApplyAsync", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenApplyAsync(u -> u)),
+                Arguments.of("thenApplyAsync(E)", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenApplyAsync(u -> u, e)),
+
+                Arguments.of("thenCombine", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenCombine(other, (u1, u2) -> u1)),
+                Arguments.of("thenCombineAsync",
+                        (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenCombineAsync(other, (u1, u2) -> u1)),
+                Arguments.of("thenCombineAsync(E)",
+                        (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenCombineAsync(other, (u1, u2) -> u1, e)),
+
+                Arguments.of("thenCompose", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenCompose(u -> completed)),
+                Arguments.of("thenComposeAsync", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenComposeAsync(u -> completed)),
+                Arguments.of("thenComposeAsync(E)", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenComposeAsync(u -> completed, e)),
+
+                Arguments.of("thenRun", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenRun(() -> {
+                })),
+                Arguments.of("thenRunAsync", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenRunAsync(() -> {
+                })),
+                Arguments.of("thenRunAsync(E)", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenRunAsync(() -> {
+                }, e)),
+
+                Arguments.of("handle", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.handle((u, t) -> u)),
+                Arguments.of("handleAsync", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.handleAsync((u, t) -> u)),
+                Arguments.of("handleAsync(E)", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.handleAsync((u, t) -> u, e)),
+
+                Arguments.of("whenComplete", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.whenComplete((u, t) -> {
+                })),
+                Arguments.of("whenCompleteAsync", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.whenCompleteAsync((u, t) -> {
+                })),
+                Arguments.of("whenCompleteAsync(E)", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.whenCompleteAsync((u, t) -> {
+                }, e)),
+
+                Arguments.of("exceptionally", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.exceptionally(t -> null)),
+                Arguments.of("exceptionallyAsync", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.exceptionallyAsync(t -> null)),
+                Arguments.of("exceptionallyAsync(E)", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.exceptionallyAsync(t -> null, e)),
+
+                Arguments.of("exceptionallyCompose",
+                        (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.exceptionallyCompose(t -> completed)),
+                Arguments.of("exceptionallyComposeAsync",
+                        (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.exceptionallyComposeAsync(t -> completed)),
+                Arguments.of("exceptionallyComposeAsync(E)",
+                        (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.exceptionallyComposeAsync(t -> completed, e)),
+
+                Arguments.of("acceptEither", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.acceptEither(other, u -> {
+                })),
+                Arguments.of("acceptEitherAsync", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.acceptEitherAsync(other, u -> {
+                })),
+                Arguments.of("acceptEitherAsync(E)", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.acceptEitherAsync(other, u -> {
+                }, e)),
+
+                Arguments.of("applyToEither", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.applyToEither(other, u -> u)),
+                Arguments.of("applyToEitherAsync", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.applyToEitherAsync(other, u -> u)),
+                Arguments.of("applyToEitherAsync(E)",
+                        (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.applyToEitherAsync(other, u -> u, e)),
+
+                Arguments.of("runAfterEither", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.runAfterEither(other, () -> {
+                })),
+                Arguments.of("runAfterEitherAsync", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.runAfterEitherAsync(other, () -> {
+                })),
+                Arguments.of("runAfterEitherAsync(E)", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.runAfterEitherAsync(other, () -> {
+                }, e)),
+
+                Arguments.of("theAcceptBoth", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenAcceptBoth(other, (u1, u2) -> {
+                })),
+                Arguments.of("thenAcceptBothAsync",
+                        (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenAcceptBothAsync(other, (u1, u2) -> {
+                        })),
+                Arguments.of("thenAcceptBothAsync(E)",
+                        (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.thenAcceptBothAsync(other, (u1, u2) -> {
+                        }, e)),
+
+                Arguments.of("runAfterBoth", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.runAfterBoth(other, () -> {
+                })),
+                Arguments.of("runAfterBothAsync", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.runAfterBothAsync(other, () -> {
+                })),
+                Arguments.of("runAfterBothAsync(E)", (BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>>) (s, e) -> s.runAfterBothAsync(other, () -> {
+                }, e)));
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource
+    void chainingMethodsReturnValueDisallowToCompletableFuture(String name, BiFunction<CompletionStage<Void>, Executor, CompletionStage<Void>> func) {
+        var future = new InternalCompletableFuture<Void>(executor);
+        var stage = future.minimalCompletionStage();
+        var result = func.apply(stage, executor);
+        assertThatThrownBy(result::toCompletableFuture).isInstanceOf(UnsupportedOperationException.class);
+    }
+}