Repository: kroxylicious/kroxylicious
Pull request #895 (2024-01-16 22:54:50)

Ensure we execute deferred Filter methods on the eventloop
In our thread-safety guarantees javadoc we state:

```
 *   <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>
```

Currently if a message is enqueued in an instance of `FilterHandler` because there is deferred work in progress, the  subsequent dispatch will be driven by the completion of the previous read or write. This could be driven by an uncontrolled thread (this is in the hands of the Filter author). So we break our promise of dispatching `onXxxRequest` and `onXxxResponse` on the channel's thread.

sonarcloud[bot]: ## [![Quality Gate Failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-failed-20px.png 'Quality Gate Failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=895) **Quality Gate failed**  
Failed conditions

[47.4% Coverage on New Code](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=895&metric=new_coverage&view=list) (required ≥ 76%)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=895)


sonarcloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=895) **Quality Gate passed**  
The SonarCloud Quality Gate passed, but some issues were introduced.

[1 New issue](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=895&resolved=false&inNewCodePeriod=true)  
[0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=895&resolved=false&inNewCodePeriod=true)  
[100.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=895&metric=new_coverage&view=list)  
[0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=895&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=895)


sonarcloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=895) **Quality Gate passed**  
The SonarCloud Quality Gate passed, but some issues were introduced.

[1 New issue](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=895&resolved=false&inNewCodePeriod=true)  
[0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=895&resolved=false&inNewCodePeriod=true)  
[100.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=895&metric=new_coverage&view=list)  
[0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=895&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=895)


sonarcloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=895) **Quality Gate passed**  
The SonarCloud Quality Gate passed, but some issues were introduced.

[1 New issue](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=895&resolved=false&inNewCodePeriod=true)  
[0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=895&resolved=false&inNewCodePeriod=true)  
[100.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=895&metric=new_coverage&view=list)  
[0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=895&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=895)


sonarcloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=895) **Quality Gate passed**  
The SonarCloud Quality Gate passed, but some issues were introduced.

[1 New issue](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=895&resolved=false&inNewCodePeriod=true)  
[0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=895&resolved=false&inNewCodePeriod=true)  
[100.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=895&metric=new_coverage&view=list)  
[0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=895&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=895)



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2b0305dd..cf6eaed5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,7 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 
 ## 0.5.0
 
+* [#895](https://github.com/kroxylicious/kroxylicious/pull/895): Ensure we execute deferred Filter methods on the eventloop
 * [#896](https://github.com/kroxylicious/kroxylicious/issues/896): In TLS config, use passwordFile as property to accept password material from a file rather than filePath.
 * [#844](https://github.com/kroxylicious/kroxylicious/pull/844): Fix connect to upstream using TLS client authentication
 * [#885](https://github.com/kroxylicious/kroxylicious/pull/885): Bump kroxy.extension.version from 0.8.0 to 0.8.1
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/SafeInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/SafeInvoker.java
index 910a52be..db2a7d99 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/SafeInvoker.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/SafeInvoker.java
@@ -29,37 +29,47 @@ record SafeInvoker(FilterInvoker invoker) implements FilterInvoker {
 
     @Override
     public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, FilterContext filterContext) {
-        if (invoker.shouldHandleRequest(apiKey, apiVersion)) {
-            CompletionStage<RequestFilterResult> stage = invoker.onRequest(apiKey, apiVersion, header, body, filterContext);
-            if (stage == null) {
-                if (logger.isWarnEnabled()) {
-                    logger.warn("invoker onRequest returned null for apiKey {}, apiVersion {}, channel: {}," +
-                            " Filters should always return a CompletionStage", apiKey, apiVersion, filterContext.channelDescriptor());
+        try {
+            if (invoker.shouldHandleRequest(apiKey, apiVersion)) {
+                CompletionStage<RequestFilterResult> stage = invoker.onRequest(apiKey, apiVersion, header, body, filterContext);
+                if (stage == null) {
+                    if (logger.isWarnEnabled()) {
+                        logger.warn("invoker onRequest returned null for apiKey {}, apiVersion {}, channel: {}," +
+                                " Filters should always return a CompletionStage", apiKey, apiVersion, filterContext.channelDescriptor());
+                    }
+                    return CompletableFuture.failedFuture(new IllegalStateException("invoker onRequest returned null for apiKey " + apiKey));
                 }
-                return CompletableFuture.failedFuture(new IllegalStateException("invoker onRequest returned null for apiKey " + apiKey));
+                return stage;
+            }
+            else {
+                return filterContext.forwardRequest(header, body);
             }
-            return stage;
         }
-        else {
-            return filterContext.forwardRequest(header, body);
+        catch (Exception e) {
+            return CompletableFuture.failedFuture(e);
         }
     }
 
     @Override
     public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, FilterContext filterContext) {
-        if (invoker.shouldHandleResponse(apiKey, apiVersion)) {
-            CompletionStage<ResponseFilterResult> stage = invoker.onResponse(apiKey, apiVersion, header, body, filterContext);
-            if (stage == null) {
-                if (logger.isWarnEnabled()) {
-                    logger.warn("invoker onResponse returned null for apiKey {}, apiVersion {}, channel: {}," +
-                            " Filters should always return a CompletionStage", apiKey, apiVersion, filterContext.channelDescriptor());
+        try {
+            if (invoker.shouldHandleResponse(apiKey, apiVersion)) {
+                CompletionStage<ResponseFilterResult> stage = invoker.onResponse(apiKey, apiVersion, header, body, filterContext);
+                if (stage == null) {
+                    if (logger.isWarnEnabled()) {
+                        logger.warn("invoker onResponse returned null for apiKey {}, apiVersion {}, channel: {}," +
+                                " Filters should always return a CompletionStage", apiKey, apiVersion, filterContext.channelDescriptor());
+                    }
+                    return CompletableFuture.failedFuture(new IllegalStateException("invoker onResponse returned null for apiKey " + apiKey));
                 }
-                return CompletableFuture.failedFuture(new IllegalStateException("invoker onResponse returned null for apiKey " + apiKey));
+                return stage;
+            }
+            else {
+                return filterContext.forwardResponse(header, body);
             }
-            return stage;
         }
-        else {
-            return filterContext.forwardResponse(header, body);
+        catch (Exception e) {
+            return CompletableFuture.failedFuture(e);
         }
     }
 
diff --git a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
index 4e968901..5027ce5d 100644
--- a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
+++ b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
@@ -299,11 +299,11 @@ public class FilterHandler extends ChannelDuplexHandler {
             }
             future.completeExceptionally(new TimeoutException("Filter %s was timed-out.".formatted(filterDescriptor())));
         }, timeoutMs, TimeUnit.MILLISECONDS);
-        return future.thenApply(filterResult -> {
+        return future.thenApplyAsync(filterResult -> {
             timeoutFuture.cancel(false);
 
             return filterResult;
-        });
+        }, ctx.executor());
     }
 
     private void deferredResponseCompleted(ResponseFilterResult ignored, Throwable throwable) {


Test patch:
diff --git a/kroxylicious-api/src/test/java/io/kroxylicious/proxy/filter/SafeInvokerTest.java b/kroxylicious-api/src/test/java/io/kroxylicious/proxy/filter/SafeInvokerTest.java
index a6a0a52b..2d90b15d 100644
--- a/kroxylicious-api/src/test/java/io/kroxylicious/proxy/filter/SafeInvokerTest.java
+++ b/kroxylicious-api/src/test/java/io/kroxylicious/proxy/filter/SafeInvokerTest.java
@@ -43,6 +43,46 @@ class SafeInvokerTest {
         verify(invoker).shouldHandleRequest(API_VERSIONS, (short) 1);
     }
 
+    @Test
+    void testOnRequestExceptionsInShouldHandleIsConvertedToFuture() {
+        // given
+        RuntimeException exception = new RuntimeException("boom!");
+        when(invoker.shouldHandleRequest(API_VERSIONS, (short) 1)).thenThrow(exception);
+        ApiVersionsRequestData message = new ApiVersionsRequestData();
+        RequestHeaderData header = new RequestHeaderData();
+        FilterContext filterContext = Mockito.mock(FilterContext.class);
+        CompletableFuture<RequestFilterResult> forwardStage = new CompletableFuture<>();
+        when(filterContext.forwardRequest(header, message)).thenReturn(forwardStage);
+
+        // when
+        CompletionStage<RequestFilterResult> stage = safeInvoker.onRequest(API_VERSIONS, (short) 1, header, message, filterContext);
+
+        // then
+        assertThat(stage).failsWithin(Duration.ZERO).withThrowableThat()
+                .isInstanceOf(ExecutionException.class).withCause(exception);
+        verify(invoker).shouldHandleRequest(API_VERSIONS, (short) 1);
+    }
+
+    @Test
+    void testOnRequestExceptionsInOnRequestIsConvertedToFuture() {
+        // given
+        RuntimeException exception = new RuntimeException("boom!");
+        when(invoker.shouldHandleRequest(API_VERSIONS, (short) 1)).thenReturn(true);
+        ApiVersionsRequestData message = new ApiVersionsRequestData();
+        RequestHeaderData header = new RequestHeaderData();
+        FilterContext filterContext = Mockito.mock(FilterContext.class);
+        CompletableFuture<RequestFilterResult> forwardStage = new CompletableFuture<>();
+        when(filterContext.forwardRequest(header, message)).thenReturn(forwardStage);
+        when(invoker.onRequest(API_VERSIONS, (short) 1, header, message, filterContext)).thenThrow(exception);
+
+        // when
+        CompletionStage<RequestFilterResult> stage = safeInvoker.onRequest(API_VERSIONS, (short) 1, header, message, filterContext);
+
+        // then
+        assertThat(stage).failsWithin(Duration.ZERO).withThrowableThat()
+                .isInstanceOf(ExecutionException.class).withCause(exception);
+    }
+
     @Test
     void testShouldHandleResponseDelegated() {
         // given
@@ -162,4 +202,35 @@ class SafeInvokerTest {
                 .withMessageContaining("invoker onResponse returned null for apiKey API_VERSIONS");
     }
 
+    @Test
+    void testOnResponseIfDelegateShouldHandleThrowsFailedFutureReturned() {
+        // given
+        RuntimeException exception = new RuntimeException("boom!");
+        when(invoker.shouldHandleResponse(API_VERSIONS, (short) 1)).thenThrow(exception);
+
+        // when
+        CompletionStage<ResponseFilterResult> stage = safeInvoker.onResponse(API_VERSIONS, (short) 1, new ResponseHeaderData(), new ApiVersionsResponseData(),
+                Mockito.mock(FilterContext.class));
+
+        // then
+        assertThat(stage).isCompletedExceptionally().failsWithin(Duration.ZERO).withThrowableOfType(ExecutionException.class)
+                .withCause(exception);
+    }
+
+    @Test
+    void testOnResponseIfDelegateOnResponseThrowsFailedFutureReturned() {
+        // given
+        RuntimeException exception = new RuntimeException("boom!");
+        when(invoker.shouldHandleResponse(API_VERSIONS, (short) 1)).thenReturn(true);
+        when(invoker.onResponse(any(), anyShort(), any(), any(), any())).thenThrow(exception);
+
+        // when
+        CompletionStage<ResponseFilterResult> stage = safeInvoker.onResponse(API_VERSIONS, (short) 1, new ResponseHeaderData(), new ApiVersionsResponseData(),
+                Mockito.mock(FilterContext.class));
+
+        // then
+        assertThat(stage).isCompletedExceptionally().failsWithin(Duration.ZERO).withThrowableOfType(ExecutionException.class)
+                .withCause(exception);
+    }
+
 }
\ No newline at end of file
diff --git a/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/client/CorrelationManager.java b/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/client/CorrelationManager.java
index 93ce5603..556ae0fc 100644
--- a/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/client/CorrelationManager.java
+++ b/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/client/CorrelationManager.java
@@ -6,6 +6,7 @@
 package io.kroxylicious.test.client;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
@@ -55,6 +56,16 @@ public class CorrelationManager {
         return brokerRequests.remove(correlationId);
     }
 
+    public void onChannelClose() {
+        List<Integer> pending = brokerRequests.keySet().stream().toList();
+        for (Integer correlation : pending) {
+            Correlation corr = brokerRequests.get(correlation);
+            if (corr.responseFuture != null) {
+                corr.responseFuture.completeExceptionally(new RuntimeException("channel closed before response received!"));
+            }
+        }
+    }
+
     /**
      * A record for which responses should be decoded, together with their
      * API key and version.
diff --git a/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/client/KafkaClient.java b/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/client/KafkaClient.java
index 25331a4c..4f38c825 100644
--- a/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/client/KafkaClient.java
+++ b/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/client/KafkaClient.java
@@ -131,6 +131,9 @@ public final class KafkaClient implements AutoCloseable {
 
             ChannelFuture connect = b.connect(host, port);
             connect.addListeners((ChannelFutureListener) channelFuture -> candidate.complete(channelFuture.channel()));
+            connect.channel().closeFuture().addListener(future -> {
+                correlationManager.onChannelClose();
+            });
             return candidate;
         }
         else {
diff --git a/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/tester/MockServerKroxyliciousTester.java b/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/tester/MockServerKroxyliciousTester.java
index c6b55103..f36322b9 100644
--- a/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/tester/MockServerKroxyliciousTester.java
+++ b/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/tester/MockServerKroxyliciousTester.java
@@ -109,6 +109,14 @@ public class MockServerKroxyliciousTester extends DefaultKroxyliciousTester {
         return requests.get(0);
     }
 
+    /**
+     * Get all requests recorded by the MockServer for a given ApiKey
+     * @return requests
+     */
+    public List<Request> getRequestsForApiKey(ApiKeys apiKeys) {
+        return mockServer.getReceivedRequests().stream().filter(request -> request.apiKeys() == apiKeys).toList();
+    }
+
     /**
      * Close the mocks erver and kroxylicious proxy
      */
diff --git a/kroxylicious-integration-test-support/src/test/java/io/kroxylicious/test/client/CorrelationManagerTest.java b/kroxylicious-integration-test-support/src/test/java/io/kroxylicious/test/client/CorrelationManagerTest.java
new file mode 100644
index 00000000..3072077e
--- /dev/null
+++ b/kroxylicious-integration-test-support/src/test/java/io/kroxylicious/test/client/CorrelationManagerTest.java
@@ -0,0 +1,67 @@
+/*
+ * 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.test.client;
+
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class CorrelationManagerTest {
+
+    @Test
+    void testPendingFuturesCompletedExceptionallyOnChannelClose() {
+        // given
+        CorrelationManager correlationManager = new CorrelationManager();
+        CompletableFuture<SequencedResponse> responseFuture = new CompletableFuture<>();
+        CompletableFuture<SequencedResponse> responseFuture2 = new CompletableFuture<>();
+        correlationManager.putBrokerRequest((short) 1, (short) 1, 1, responseFuture);
+        correlationManager.putBrokerRequest((short) 1, (short) 1, 2, responseFuture2);
+
+        // when
+        correlationManager.onChannelClose();
+
+        // then
+        assertThat(responseFuture).failsWithin(Duration.ZERO);
+        assertThat(responseFuture2).failsWithin(Duration.ZERO);
+    }
+
+    @Test
+    void testNullFutureToleratedOnChannelClose() {
+        // given
+        CorrelationManager correlationManager = new CorrelationManager();
+        CompletableFuture<SequencedResponse> responseFuture = new CompletableFuture<>();
+        correlationManager.putBrokerRequest((short) 1, (short) 1, 1, responseFuture);
+        correlationManager.putBrokerRequest((short) 1, (short) 1, 2, null);
+
+        // when
+        correlationManager.onChannelClose();
+
+        // then
+        assertThat(responseFuture).failsWithin(Duration.ZERO);
+    }
+
+    @Test
+    void testCorrelationRetrievableOnceOnly() {
+        // given
+        int correlationId = 1;
+        CorrelationManager correlationManager = new CorrelationManager();
+        CompletableFuture<SequencedResponse> responseFuture = new CompletableFuture<>();
+        correlationManager.putBrokerRequest((short) 1, (short) 1, correlationId, responseFuture);
+        CorrelationManager.Correlation brokerCorrelation = correlationManager.getBrokerCorrelation(correlationId);
+        assertThat(brokerCorrelation).isNotNull();
+
+        // when
+        CorrelationManager.Correlation brokerCorrelation2 = correlationManager.getBrokerCorrelation(correlationId);
+
+        // then
+        assertThat(brokerCorrelation2).isNull();
+    }
+
+}
diff --git a/kroxylicious-integration-test-support/src/test/java/io/kroxylicious/test/tester/KroxyliciousTestersTest.java b/kroxylicious-integration-test-support/src/test/java/io/kroxylicious/test/tester/KroxyliciousTestersTest.java
index 11b5d8d7..b0a60f97 100644
--- a/kroxylicious-integration-test-support/src/test/java/io/kroxylicious/test/tester/KroxyliciousTestersTest.java
+++ b/kroxylicious-integration-test-support/src/test/java/io/kroxylicious/test/tester/KroxyliciousTestersTest.java
@@ -251,6 +251,7 @@ class KroxyliciousTestersTest {
     void testMockRequestMockTester() {
         try (var tester = mockKafkaKroxyliciousTester(KroxyliciousConfigUtils::proxy)) {
             assertCanSendRequestsAndReceiveMockResponses(tester, tester::simpleTestClient);
+            tester.clearMock();
             assertCanSendRequestsAndReceiveMockResponses(tester, () -> tester.simpleTestClient(DEFAULT_VIRTUAL_CLUSTER));
         }
     }
@@ -326,18 +327,22 @@ class KroxyliciousTestersTest {
             var mockResponse2 = new ListTransactionsResponseData().setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code());
             tester.addMockResponseForApiKey(new ResponsePayload(ApiKeys.LIST_TRANSACTIONS, ApiKeys.LIST_TRANSACTIONS.latestVersion(), mockResponse2));
 
-            var response1 = kafkaClient
-                    .getSync(new Request(ApiKeys.DESCRIBE_ACLS, ApiKeys.DESCRIBE_ACLS.latestVersion(), "client", new DescribeAclsRequestData()));
+            Request describeAcls = new Request(ApiKeys.DESCRIBE_ACLS, ApiKeys.DESCRIBE_ACLS.latestVersion(), "client", new DescribeAclsRequestData());
+            var response1 = kafkaClient.getSync(describeAcls);
             ApiMessage response1Message = response1.payload().message();
             assertThat(response1Message).isInstanceOf(DescribeAclsResponseData.class);
             assertThat(response1Message).isEqualTo(mockResponse1);
+            assertThat(tester.getOnlyRequestForApiKey(ApiKeys.DESCRIBE_ACLS)).isEqualTo(describeAcls);
+            assertThat(tester.getRequestsForApiKey(ApiKeys.DESCRIBE_ACLS)).containsOnly(describeAcls);
 
-            var response2 = kafkaClient
-                    .getSync(new Request(ApiKeys.LIST_TRANSACTIONS, ApiKeys.LIST_TRANSACTIONS.latestVersion(), "client", new ListTransactionsRequestData()));
+            Request listTransactions = new Request(ApiKeys.LIST_TRANSACTIONS, ApiKeys.LIST_TRANSACTIONS.latestVersion(), "client", new ListTransactionsRequestData());
+            var response2 = kafkaClient.getSync(listTransactions);
             ApiMessage response2Message = response2.payload().message();
             assertInstanceOf(ListTransactionsResponseData.class, response2Message);
             assertThat(response2Message).isInstanceOf(ListTransactionsResponseData.class);
             assertThat(response2Message).isEqualTo(mockResponse2);
+            assertThat(tester.getOnlyRequestForApiKey(ApiKeys.LIST_TRANSACTIONS)).isEqualTo(listTransactions);
+            assertThat(tester.getRequestsForApiKey(ApiKeys.LIST_TRANSACTIONS)).containsOnly(listTransactions);
         }
     }
 
diff --git a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/FilterIT.java b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/FilterIT.java
index c0f5c658..a41af086 100644
--- a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/FilterIT.java
+++ b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/FilterIT.java
@@ -15,6 +15,7 @@ import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.stream.Stream;
 
 import org.apache.kafka.clients.admin.Admin;
@@ -69,6 +70,7 @@ import io.kroxylicious.testing.kafka.junit5ext.Topic;
 import static io.kroxylicious.UnknownTaggedFields.unknownTaggedFieldsToStrings;
 import static io.kroxylicious.proxy.filter.ApiVersionsMarkingFilter.INTERSECTED_API_VERSION_RANGE_TAG;
 import static io.kroxylicious.proxy.filter.ApiVersionsMarkingFilter.UPSTREAM_API_VERSION_RANGE_TAG;
+import static io.kroxylicious.proxy.filter.RequestResponseMarkingFilter.DISPATCH_THREAD;
 import static io.kroxylicious.proxy.filter.RequestResponseMarkingFilter.FILTER_NAME_TAG;
 import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.proxy;
 import static io.kroxylicious.test.tester.KroxyliciousTesters.kroxyliciousTester;
@@ -278,6 +280,74 @@ class FilterIT {
                 ForwardingStyle.ASYNCHRONOUS_REQUEST_TO_BROKER);
     }
 
+    /**
+     * This is here to test a subtle case, we want to ensure that if a Filter defers some work
+     * and completes that work from some uncontrolled thread, that other Messages enqueued at the
+     * filter are dispatched on the eventloop. Given that the eventloop is dynamically allocated
+     * to the channel, we cannot precisely check what thread we are running on, but we can
+     * assert that the user's Filter method is being invoked from the same thread.
+     */
+    @ParameterizedTest
+    @EnumSource(value = RequestResponseMarkingFilterFactory.Direction.class)
+    void supportAsyncBatchForwarding(RequestResponseMarkingFilterFactory.Direction direction) throws Exception {
+        doSupportsForwardDeferredByAsynchronousBatchRequest(direction,
+                "supportsForwardDeferredByAsynchronousBrokerRequest",
+                ForwardingStyle.ASYNCHRONOUS_DELAYED);
+    }
+
+    private void doSupportsForwardDeferredByAsynchronousBatchRequest(RequestResponseMarkingFilterFactory.Direction direction, String name,
+                                                                     ForwardingStyle forwardingStyle)
+            throws ExecutionException, InterruptedException, TimeoutException {
+        var markingFilter = new FilterDefinitionBuilder(RequestResponseMarkingFilterFactory.class.getName())
+                .withConfig("keysToMark", Set.of(LIST_TRANSACTIONS),
+                        "direction", Set.of(direction),
+                        "name", name,
+                        "forwardingStyle", forwardingStyle)
+                .build();
+        try (var tester = mockKafkaKroxyliciousTester((mockBootstrap) -> proxy(mockBootstrap).addToFilters(markingFilter));
+                var kafkaClient = tester.simpleTestClient()) {
+            tester.addMockResponseForApiKey(new ResponsePayload(LIST_TRANSACTIONS, LIST_TRANSACTIONS.latestVersion(), new ListTransactionsResponseData()));
+            ApiVersionsResponseData apiVersionsResponseData = new ApiVersionsResponseData();
+            apiVersionsResponseData.apiKeys()
+                    .add(new ApiVersionsResponseData.ApiVersion().setApiKey(FETCH.id).setMaxVersion(FETCH.latestVersion()).setMinVersion(FETCH.oldestVersion()));
+            tester.addMockResponseForApiKey(new ResponsePayload(API_VERSIONS, API_VERSIONS.latestVersion(), apiVersionsResponseData));
+
+            // In the ASYNCHRONOUS_REQUEST_TO_BROKER case, the filter will send an async list_group
+            // request to the broker and defer the forward of the list transaction response until the list groups
+            // response arrives.
+            if (forwardingStyle == ForwardingStyle.ASYNCHRONOUS_REQUEST_TO_BROKER) {
+                tester.addMockResponseForApiKey(new ResponsePayload(LIST_GROUPS, LIST_GROUPS.latestVersion(), new ListGroupsResponseData()));
+            }
+
+            var responseAFuture = kafkaClient
+                    .get(new Request(LIST_TRANSACTIONS, LIST_TRANSACTIONS.latestVersion(), "client", new ListTransactionsRequestData()));
+
+            var responseBFuture = kafkaClient
+                    .get(new Request(LIST_TRANSACTIONS, LIST_TRANSACTIONS.latestVersion(), "client", new ListTransactionsRequestData()));
+            Response responseA = responseAFuture.get(5, TimeUnit.SECONDS);
+            Response responseB = responseBFuture.get(5, TimeUnit.SECONDS);
+
+            List<Request> listTransactionRequests = tester.getRequestsForApiKey(LIST_TRANSACTIONS);
+            var requestMessageReceivedByBrokerA = listTransactionRequests.get(0).message();
+            String dispatchThreadA = getDispatchThreadName(direction, responseA, requestMessageReceivedByBrokerA);
+            var requestMessageReceivedByBrokerB = listTransactionRequests.get(1).message();
+            String dispatchThreadB = getDispatchThreadName(direction, responseB, requestMessageReceivedByBrokerB);
+            assertThat(dispatchThreadA).isNotEmpty().containsIgnoringCase("eventloop");
+            assertThat(dispatchThreadB).isNotEmpty().containsIgnoringCase("eventloop")
+                    .describedAs("filter invocations should be dispatched from the same thread").isEqualTo(dispatchThreadA);
+        }
+    }
+
+    private static String getDispatchThreadName(RequestResponseMarkingFilterFactory.Direction direction, Response responseA, ApiMessage requestMessageReceivedByBrokerA) {
+        var responseMessageReceivedByClientA = responseA.payload().message();
+
+        assertThat(requestMessageReceivedByBrokerA).isInstanceOf(ListTransactionsRequestData.class);
+        assertThat(responseMessageReceivedByClientA).isInstanceOf(ListTransactionsResponseData.class);
+
+        var target = direction == RequestResponseMarkingFilterFactory.Direction.REQUEST ? requestMessageReceivedByBrokerA : responseMessageReceivedByClientA;
+        return unknownTaggedFieldsToStrings(target, DISPATCH_THREAD).findFirst().orElse("");
+    }
+
     private void doSupportsForwardDeferredByAsynchronousRequest(RequestResponseMarkingFilterFactory.Direction direction, String name,
                                                                 ForwardingStyle forwardingStyle) {
         var markingFilter = new FilterDefinitionBuilder(RequestResponseMarkingFilterFactory.class.getName())
diff --git a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
index b5685aa7..3813b709 100644
--- a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
+++ b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
@@ -33,6 +33,7 @@ import static java.nio.charset.StandardCharsets.UTF_8;
  */
 public class RequestResponseMarkingFilter implements RequestFilter, ResponseFilter {
 
+    public static final int DISPATCH_THREAD = 499;
     public static final int FILTER_NAME_TAG = 500;
     private final FilterFactoryContext constructionContext;
     private final String name;
@@ -53,6 +54,7 @@ public class RequestResponseMarkingFilter implements RequestFilter, ResponseFilt
         if (!(direction.contains(RequestResponseMarkingFilterFactory.Direction.REQUEST) && keysToMark.contains(apiKey))) {
             return context.forwardRequest(header, body);
         }
+        tagWithDispatchThread(body);
 
         return forwardingStyle.apply(new ForwardingContext(context, constructionContext, body))
                 .thenApply(request -> applyTaggedField(request, RequestResponseMarkingFilterFactory.Direction.REQUEST, name))
@@ -64,12 +66,19 @@ public class RequestResponseMarkingFilter implements RequestFilter, ResponseFilt
         if (!(direction.contains(RequestResponseMarkingFilterFactory.Direction.RESPONSE) && keysToMark.contains(apiKey))) {
             return context.forwardResponse(header, response);
         }
+        tagWithDispatchThread(response);
 
         return forwardingStyle.apply(new ForwardingContext(context, constructionContext, response))
                 .thenApply(request -> applyTaggedField(request, RequestResponseMarkingFilterFactory.Direction.RESPONSE, name))
                 .thenCompose(taggedRequest -> context.forwardResponse(header, taggedRequest));
     }
 
+    private static void tagWithDispatchThread(ApiMessage body) {
+        Thread currentThread = Thread.currentThread();
+        String identifier = currentThread.getName() + "::" + currentThread.threadId();
+        body.unknownTaggedFields().add(new RawTaggedField(DISPATCH_THREAD, identifier.getBytes()));
+    }
+
     private ApiMessage applyTaggedField(ApiMessage body, RequestResponseMarkingFilterFactory.Direction direction, String name) {
         body.unknownTaggedFields().add(createTaggedField(direction.toString().toLowerCase(Locale.ROOT), name));
         return body;
diff --git a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
index e4dd500a..965d299f 100644
--- a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
+++ b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
@@ -120,19 +120,18 @@ class FilterHandlerTest extends FilterHarness {
     }
 
     @Test
-    void deferredRequestDelaysSubsequentRequest() {
+    void deferredRequestMethodsDispatchedOnEventLoop() {
         var req1 = new ApiVersionsRequestData().setClientSoftwareName("req1");
-        var req2 = new ApiVersionsRequestData().setClientSoftwareName("req2");
 
         var requestFutureMap = new LinkedHashMap<ApiVersionsRequestData, CompletableFuture<Void>>();
         requestFutureMap.put(req1, new CompletableFuture<>());
-        requestFutureMap.put(req2, CompletableFuture.completedFuture(null));
 
         ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> requestFutureMap.get(request)
                 .thenCompose((u) -> context.forwardRequest(header, request));
         buildChannel(filter);
 
         requestFutureMap.keySet().forEach(this::writeRequest);
+        OpaqueRequestFrame opaqueRequestFrame = writeArbitraryOpaqueRequest();
         channel.runPendingTasks();
 
         var propagated = channel.readOutbound();
@@ -145,37 +144,63 @@ class FilterHandlerTest extends FilterHarness {
         DecodedRequestFrame<?> outboundRequest1 = channel.readOutbound();
         assertThat(outboundRequest1).extracting(DecodedRequestFrame::body).isEqualTo(req1);
 
-        DecodedRequestFrame<?> outboundRequest2 = channel.readOutbound();
-        assertThat(outboundRequest2).extracting(DecodedRequestFrame::body).isEqualTo(req2);
+        OpaqueRequestFrame outboundRequest2 = channel.readOutbound();
+        assertThat(outboundRequest2).isSameAs(opaqueRequestFrame);
     }
 
     @Test
-    void deferredRequestDelaysSubsequentOpaqueRequest() {
+    void deferredRequestMethodsDispatchedOnEventloop() {
         var req1 = new ApiVersionsRequestData().setClientSoftwareName("req1");
+        var req2 = new ApiVersionsRequestData().setClientSoftwareName("req2");
 
         var requestFutureMap = new LinkedHashMap<ApiVersionsRequestData, CompletableFuture<Void>>();
         requestFutureMap.put(req1, new CompletableFuture<>());
+        requestFutureMap.put(req2, new CompletableFuture<>());
 
-        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> requestFutureMap.get(request)
-                .thenCompose((u) -> context.forwardRequest(header, request));
+        // we expect the embedded netty eventloop to run on this thread
+        AtomicReference<Thread> expectedDispatchThread = new AtomicReference<>();
+        ApiVersionsRequestFilter filter = new ApiVersionsRequestFilter() {
+
+            @Override
+            public boolean shouldHandleApiVersionsRequest(short apiVersion) {
+                if (Thread.currentThread() != expectedDispatchThread.get()) {
+                    throw new IllegalStateException("Filter method dispatched on unexpected thread!");
+                }
+                return true;
+            }
+
+            @Override
+            public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
+                                                                             FilterContext context) {
+                if (Thread.currentThread() != expectedDispatchThread.get()) {
+                    return CompletableFuture.failedFuture(new IllegalStateException("Filter method dispatched on unexpected thread!"));
+                }
+                // delay required to provoke the second request to be queued
+                return requestFutureMap.get(request)
+                        .thenCompose((u) -> context.forwardRequest(header, request));
+            }
+        };
         buildChannel(filter);
+        expectedDispatchThread.set(obtainEventLoop());
+        writeRequest(req1);
+        writeRequest(req2);
 
-        requestFutureMap.keySet().forEach(this::writeRequest);
-        OpaqueRequestFrame opaqueRequestFrame = writeArbitraryOpaqueRequest();
+        // simulate Filter completing from an uncontrolled thread
+        CompletableFuture.runAsync(() -> requestFutureMap.get(req1).complete(null)).join();
         channel.runPendingTasks();
 
-        var propagated = channel.readOutbound();
-        assertThat(propagated).isNull();
+        DecodedRequestFrame<?> outboundRequest1 = channel.readOutbound();
 
-        // complete req1's future, now expect both requests to flow.
-        requestFutureMap.get(req1).complete(null);
+        assertThat(outboundRequest1).extracting(DecodedRequestFrame::body).isEqualTo(req1);
 
+        // simulate Filter completing from an uncontrolled thread
+        CompletableFuture.runAsync(() -> requestFutureMap.get(req2).complete(null)).join();
         channel.runPendingTasks();
-        DecodedRequestFrame<?> outboundRequest1 = channel.readOutbound();
-        assertThat(outboundRequest1).extracting(DecodedRequestFrame::body).isEqualTo(req1);
 
-        OpaqueRequestFrame outboundRequest2 = channel.readOutbound();
-        assertThat(outboundRequest2).isSameAs(opaqueRequestFrame);
+        DecodedRequestFrame<?> outboundRequest2 = channel.readOutbound();
+
+        assertThat(channel.isOpen()).isTrue();
+        assertThat(outboundRequest2).extracting(DecodedRequestFrame::body).isEqualTo(req2);
     }
 
     @Test
@@ -833,7 +858,6 @@ class FilterHandlerTest extends FilterHarness {
 
     @Test
     void sendRequestChainedActionsRunOnNettyEventLoop() {
-        var eventLoopThreadFuture = new CompletableFuture<Thread>();
 
         var oobRequestBody = new FetchRequestData();
         var applyActionThread = new AtomicReference<Thread>();
@@ -853,9 +877,7 @@ class FilterHandlerTest extends FilterHarness {
         var requestFrame = writeRequest(new ApiVersionsRequestData());
 
         // capture the thread used by the embedded channel
-        channel.eventLoop().submit(() -> eventLoopThreadFuture.complete(Thread.currentThread()));
-        channel.runPendingTasks();
-        assertThat(eventLoopThreadFuture).isCompleted();
+        Thread eventloopThread = obtainEventLoop();
 
         // verify filter has sent the send request.
         InternalRequestFrame<?> propagatedOobRequest = channel.readOutbound();
@@ -870,17 +892,25 @@ class FilterHandlerTest extends FilterHarness {
         // Verify actions ran on the expected thread.
         assertThat(applyActionThread)
                 .describedAs("first chained action (apply) must run on event loop")
-                .hasValue(eventLoopThreadFuture.getNow(null));
+                .hasValue(eventloopThread);
 
         assertThat(applyAsyncActionThread)
                 .describedAs("second chained action (applySync) must run on event loop")
-                .hasValue(eventLoopThreadFuture.getNow(null));
+                .hasValue(eventloopThread);
 
         // Verify the filtered request arrived at outcome.
         var propagated = channel.readOutbound();
         assertThat(propagated).isEqualTo(requestFrame);
     }
 
+    private Thread obtainEventLoop() {
+        var eventLoopThreadFuture = new CompletableFuture<Thread>();
+        channel.eventLoop().submit(() -> eventLoopThreadFuture.complete(Thread.currentThread()));
+        channel.runPendingTasks();
+        assertThat(eventLoopThreadFuture).isCompleted();
+        return eventLoopThreadFuture.getNow(null);
+    }
+
     @Test
     void upstreamFiltersCanFilterOutOfBandRequest() {
         var oobRequestBody = new FetchRequestData();