Repository: kroxylicious/kroxylicious
Pull request #512 (2023-08-02 21:08:35)

Another way of trying to get sonar to work
### Type of change

- Bugfix

### Description

Fixes active sonar scanning 

### Additional Context

_Why are you making this pull request?_

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [ ] Write tests
- [ ] 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.
- [ ] Update documentation
- [ ] Reference relevant issue(s) and close them after merging
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).



Patch:
diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml
index 610dca1e..3e2b4988 100644
--- a/.github/workflows/maven.yml
+++ b/.github/workflows/maven.yml
@@ -50,7 +50,16 @@ jobs:
           key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}
           restore-keys: ${{ runner.os }}-m2
 
-      - name: 'Build example service'
+      - name: 'Build Kroxylicious maven project'
         env:
-          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         run: mvn -B clean verify -Pci
+      - name: Save PR number to file
+        if: github.event_name == 'pull_request' && ${{ matrix.os }} == 'ubuntu-latest' && ${{ matrix.java_version }} == '17'
+        run: echo ${{ github.event.number }} > PR_NUMBER.txt
+      - name: Archive PR number
+        if: github.event_name == 'pull_request' && ${{ matrix.os }} == 'ubuntu-latest' && ${{ matrix.java_version }} == '17'
+        uses: actions/upload-artifact@v3
+        with:
+         name: PR_NUMBER
+         path: PR_NUMBER.txt
diff --git a/.github/workflows/sonar.yml b/.github/workflows/sonar.yml
new file mode 100644
index 00000000..f8d5eece
--- /dev/null
+++ b/.github/workflows/sonar.yml
@@ -0,0 +1,73 @@
+name: Sonar Scan
+
+on:
+  workflow_run:
+    workflows: [Build]
+    types: [completed]
+
+jobs:
+  Sonar:
+    runs-on: ubuntu-latest
+    if: github.event.workflow_run.conclusion == 'success'
+    steps:
+      - name: echo event
+        run: cat $GITHUB_EVENT_PATH
+      - name: Download PR number artifact
+        if: github.event.workflow_run.event == 'pull_request'
+        uses: dawidd6/action-download-artifact@v2
+        with:
+          workflow: Build
+          run_id: ${{ github.event.workflow_run.id }}
+          name: PR_NUMBER
+      - name: Read PR_NUMBER.txt
+        if: github.event.workflow_run.event == 'pull_request'
+        id: pr_number
+        uses: juliangruber/read-file-action@v1
+        with:
+          path: ./PR_NUMBER.txt
+      - name: Request GitHub API for PR data
+        if: github.event.workflow_run.event == 'pull_request'
+        uses: octokit/request-action@v2.x
+        id: get_pr_data
+        with:
+          route: GET /repos/{full_name}/pulls/{number}
+          number: ${{ steps.pr_number.outputs.content }}
+          full_name: ${{ github.event.repository.full_name }}
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      - uses: actions/checkout@v3
+        with:
+          repository: ${{ github.event.workflow_run.head_repository.full_name }}
+          ref: ${{ github.event.workflow_run.head_branch }}
+          fetch-depth: 0
+      - name: Checkout base branch
+        if: github.event.workflow_run.event == 'pull_request'
+        run: |
+          git remote add upstream ${{ github.event.repository.clone_url }}
+          git fetch upstream
+          git checkout -B ${{ fromJson(steps.get_pr_data.outputs.data).base.ref }} upstream/${{ fromJson(steps.get_pr_data.outputs.data).base.ref }}
+          git checkout ${{ github.event.workflow_run.head_branch }}
+          git clean -ffdx && git reset --hard HEAD
+      - name: Set up JDK 20
+        uses: actions/setup-java@v3
+        with:
+          java-version: 20
+          distribution: 'temurin'
+      - name: 'Cache Maven packages'
+        uses: actions/cache@v3
+        with:
+          path: ~/.m2
+          key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}
+          restore-keys: ${{ runner.os }}-m2
+      - name: SonarCloud Scan on PR
+        if: github.event.workflow_run.event == 'pull_request'
+        run: mvn -B clean verify -Pci org.sonarsource.scanner.maven:sonar-maven-plugin:3.9.1.2184:sonar -Dsonar.projectKey=kroxylicious_kroxylicious -Dsonar.scm.revision=${{ github.event.workflow_run.head_sha }} -Dsonar.pullrequest.key=${{ fromJson(steps.get_pr_data.outputs.data).number }} -Dsonar.pullrequest.branch=${{ fromJson(steps.get_pr_data.outputs.data).head.ref }} -Dsonar.pullrequest.base=${{ fromJson(steps.get_pr_data.outputs.data).base.ref }}
+        env:
+          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      - name: SonarCloud Scan on push
+        if: github.event.workflow_run.event == 'push' && github.event.workflow_run.head_repository.full_name == github.event.repository.full_name
+        run: mvn -B clean verify -Pci org.sonarsource.scanner.maven:sonar-maven-plugin:3.9.1.2184:sonar -Dsonar.scm.revision=${{ github.event.workflow_run.head_sha }} -Dsonar.branch.name=${{ github.event.workflow_run.head_branch }}
+        env:
+          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
\ No newline at end of file
diff --git a/CHANGELOG.md b/CHANGELOG.md
index fae8b68b..acbf54fd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,8 +6,6 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 
 ## 0.3.0
 
-
-* [#494](https://github.com/kroxylicious/kroxylicious/issues/494): [Breaking] Make the Filter API fully asynchronous (filter methods must return a CompletionStage)
 * [#498](https://github.com/kroxylicious/kroxylicious/issues/498): Include the cluster name from the configuration node in the config model.
 * [#488](https://github.com/kroxylicious/kroxylicious/pull/488): Kroxylicious Bill Of Materials 
 * [#480](https://github.com/kroxylicious/kroxylicious/issues/480): Multi-tenant - add suport for the versions of OffsetFetch, FindCoordinator, and DeleteTopics used by Sarama client v1.38.1
@@ -30,11 +28,6 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 
 ### Changes, deprecations and removals
 
-The Filter API is refactored to be fully asynchronous.  Filter API methods such as `#onXxxxRequest` and `onXxxxResponse`
-now are required to return a `CompletionStage<FilterResult>`. The `FilterResult` encapsulates the message to be
-forwarded and carries orders (such as close the connection). The context provides factory methods for creating
-`FilterResult` objects.
-
 The default metrics port has changed from 9193 to 9190 to prevent port collisions
 
 Filter Authors can now implement CompositeFilter if they want a single configuration block to contribute multiple Filters
diff --git a/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml b/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml
index c73f2b19..b71b2545 100644
--- a/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml
+++ b/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml
@@ -80,11 +80,6 @@
             <artifactId>mockito-core</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.mockito</groupId>
-            <artifactId>mockito-junit-jupiter</artifactId>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>com.flipkart.zjsonpatch</groupId>
             <artifactId>zjsonpatch</artifactId>
diff --git a/kroxylicious-additional-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java b/kroxylicious-additional-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
index 9df54175..dafe9a99 100644
--- a/kroxylicious-additional-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
+++ b/kroxylicious-additional-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
@@ -6,7 +6,6 @@
 package io.kroxylicious.proxy.filter.multitenant;
 
 import java.util.Optional;
-import java.util.concurrent.CompletionStage;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
@@ -91,19 +90,17 @@ import io.kroxylicious.proxy.filter.OffsetForLeaderEpochRequestFilter;
 import io.kroxylicious.proxy.filter.OffsetForLeaderEpochResponseFilter;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 import io.kroxylicious.proxy.filter.SyncGroupRequestFilter;
 import io.kroxylicious.proxy.filter.TxnOffsetCommitRequestFilter;
 import io.kroxylicious.proxy.filter.TxnOffsetCommitResponseFilter;
 
 /**
  * Simple multi-tenant filter.
- * <br/>
+ *
  * Uses the first component of a fully-qualified host name as a tenant prefix.
  * This tenant prefix is prepended to the kafka resources name in order to present an isolated
  * environment for each tenant.
- * <br/>
+ *
  * TODO prefix other resources e.g. group names, transaction ids
  * TODO disallow the use of topic uids belonging to one tenant by another.
  */
@@ -135,84 +132,78 @@ public class MultiTenantTransformationFilter
     private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantTransformationFilter.class);
 
     @Override
-    public CompletionStage<RequestFilterResult> onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request,
-                                                                      KrpcFilterContext context) {
+    public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response,
-                                                                        KrpcFilterContext context) {
+    public void onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onDeleteTopicsRequest(short apiVersion, RequestHeaderData header, DeleteTopicsRequestData request,
-                                                                      KrpcFilterContext context) {
+    public void onDeleteTopicsRequest(short apiVersion, RequestHeaderData header, DeleteTopicsRequestData request, KrpcFilterContext context) {
         // the topicName field was present up to and including version 5
         request.setTopicNames(request.topicNames().stream().map(topic -> applyTenantPrefix(context, topic)).toList());
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, topic.topicId() != null));
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onDeleteTopicsResponse(short apiVersion, ResponseHeaderData header, DeleteTopicsResponseData response,
-                                                                        KrpcFilterContext context) {
+    public void onDeleteTopicsResponse(short apiVersion, ResponseHeaderData header, DeleteTopicsResponseData response, KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onMetadataRequest(short apiVersion, RequestHeaderData header, MetadataRequestData request, KrpcFilterContext context) {
+    public void onMetadataRequest(short apiVersion, RequestHeaderData header, MetadataRequestData request, KrpcFilterContext context) {
         if (request.topics() != null) {
             // n.b. request.topics() == null used to query all the topics.
             request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         }
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
+
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onMetadataResponse(short apiVersion, ResponseHeaderData header, MetadataResponseData response,
-                                                                    KrpcFilterContext context) {
+    public void onMetadataResponse(short apiVersion, ResponseHeaderData header, MetadataResponseData response, KrpcFilterContext context) {
         String tenantPrefix = getTenantPrefix(context);
         response.topics().removeIf(topic -> !topic.name().startsWith(tenantPrefix)); // TODO: allow kafka internal topics to be returned?
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         applyTenantPrefix(context, request::transactionalId, request::setTransactionalId, true);
         request.topicData().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
+
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
-                                                                   KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
+
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onListOffsetsRequest(short apiVersion, RequestHeaderData header, ListOffsetsRequestData request,
-                                                                     KrpcFilterContext context) {
+    public void onListOffsetsRequest(short apiVersion, RequestHeaderData header, ListOffsetsRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onListOffsetsResponse(short apiVersion, ResponseHeaderData header, ListOffsetsResponseData response,
-                                                                       KrpcFilterContext context) {
+    public void onListOffsetsResponse(short apiVersion, ResponseHeaderData header, ListOffsetsResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onOffsetFetchRequest(short apiVersion, RequestHeaderData header, OffsetFetchRequestData request,
-                                                                     KrpcFilterContext context) {
+    public void onOffsetFetchRequest(short apiVersion, RequestHeaderData header, OffsetFetchRequestData request, KrpcFilterContext context) {
         // the groupId and top-level topic fields were present up to and including version 7
         Optional.ofNullable(request.groupId()).ifPresent(groupId -> applyTenantPrefix(context, request::groupId, request::setGroupId, true));
         Optional.ofNullable(request.topics()).ifPresent(topics -> topics.forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false)));
@@ -222,158 +213,140 @@ public class MultiTenantTransformationFilter
                     .ifPresent(topics -> topics.forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false)));
         });
 
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onOffsetFetchResponse(short apiVersion, ResponseHeaderData header, OffsetFetchResponseData response,
-                                                                       KrpcFilterContext context) {
+    public void onOffsetFetchResponse(short apiVersion, ResponseHeaderData header, OffsetFetchResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         response.groups().forEach(responseGroup -> {
             removeTenantPrefix(context, responseGroup::groupId, responseGroup::setGroupId, false);
             responseGroup.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         });
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onOffsetForLeaderEpochRequest(short apiVersion, RequestHeaderData header, OffsetForLeaderEpochRequestData request,
-                                                                              KrpcFilterContext context) {
+    public void onOffsetForLeaderEpochRequest(short apiVersion, RequestHeaderData header, OffsetForLeaderEpochRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::topic, topic::setTopic, false));
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onOffsetForLeaderEpochResponse(short apiVersion, ResponseHeaderData header, OffsetForLeaderEpochResponseData response,
-                                                                                KrpcFilterContext context) {
+    public void onOffsetForLeaderEpochResponse(short apiVersion, ResponseHeaderData header, OffsetForLeaderEpochResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::topic, topic::setTopic, false));
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onOffsetCommitRequest(short apiVersion, RequestHeaderData header, OffsetCommitRequestData request,
-                                                                      KrpcFilterContext context) {
+    public void onOffsetCommitRequest(short apiVersion, RequestHeaderData header, OffsetCommitRequestData request, KrpcFilterContext context) {
         applyTenantPrefix(context, request::groupId, request::setGroupId, false);
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onOffsetCommitResponse(short apiVersion, ResponseHeaderData header, OffsetCommitResponseData response,
-                                                                        KrpcFilterContext context) {
+    public void onOffsetCommitResponse(short apiVersion, ResponseHeaderData header, OffsetCommitResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onOffsetDeleteRequest(short apiVersion, RequestHeaderData header, OffsetDeleteRequestData request,
-                                                                      KrpcFilterContext context) {
+    public void onOffsetDeleteRequest(short apiVersion, RequestHeaderData header, OffsetDeleteRequestData request, KrpcFilterContext context) {
         applyTenantPrefix(context, request::groupId, request::setGroupId, false);
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onOffsetDeleteResponse(short apiVersion, ResponseHeaderData header, OffsetDeleteResponseData response,
-                                                                        KrpcFilterContext context) {
+    public void onOffsetDeleteResponse(short apiVersion, ResponseHeaderData header, OffsetDeleteResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
+    public void onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::topic, topic::setTopic, topic.topicId() != null));
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
+    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::topic, topic::setTopic, topic.topicId() != null));
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onFindCoordinatorRequest(short apiVersion, RequestHeaderData header, FindCoordinatorRequestData request,
-                                                                         KrpcFilterContext context) {
+    public void onFindCoordinatorRequest(short apiVersion, RequestHeaderData header, FindCoordinatorRequestData request, KrpcFilterContext context) {
         // the key fields was present up to and including version 4
         Optional.ofNullable(request.key()).ifPresent(unused -> applyTenantPrefix(context, request::key, request::setKey, true));
         request.setCoordinatorKeys(request.coordinatorKeys().stream().map(key -> applyTenantPrefix(context, key)).toList());
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onFindCoordinatorResponse(short apiVersion, ResponseHeaderData header, FindCoordinatorResponseData response,
-                                                                           KrpcFilterContext context) {
+    public void onFindCoordinatorResponse(short apiVersion, ResponseHeaderData header, FindCoordinatorResponseData response, KrpcFilterContext context) {
         response.coordinators().forEach(coordinator -> removeTenantPrefix(context, coordinator::key, coordinator::setKey, false));
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onListGroupsResponse(short apiVersion, ResponseHeaderData header, ListGroupsResponseData response,
-                                                                      KrpcFilterContext context) {
+    public void onListGroupsResponse(short apiVersion, ResponseHeaderData header, ListGroupsResponseData response, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         var filteredGroups = response.groups().stream().filter(listedGroup -> listedGroup.groupId().startsWith(tenantPrefix)).toList();
         filteredGroups.forEach(listedGroup -> removeTenantPrefix(context, listedGroup::groupId, listedGroup::setGroupId, false));
         response.setGroups(filteredGroups);
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onJoinGroupRequest(short apiVersion, RequestHeaderData header, JoinGroupRequestData request,
-                                                                   KrpcFilterContext context) {
+    public void onJoinGroupRequest(short apiVersion, RequestHeaderData header, JoinGroupRequestData request, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setGroupId(tenantPrefix + request.groupId());
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onSyncGroupRequest(short apiVersion, RequestHeaderData header, SyncGroupRequestData request,
-                                                                   KrpcFilterContext context) {
+    public void onSyncGroupRequest(short apiVersion, RequestHeaderData header, SyncGroupRequestData request, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setGroupId(tenantPrefix + request.groupId());
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onLeaveGroupRequest(short apiVersion, RequestHeaderData header, LeaveGroupRequestData request,
-                                                                    KrpcFilterContext context) {
+    public void onLeaveGroupRequest(short apiVersion, RequestHeaderData header, LeaveGroupRequestData request, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setGroupId(tenantPrefix + request.groupId());
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onHeartbeatRequest(short apiVersion, RequestHeaderData header, HeartbeatRequestData request,
-                                                                   KrpcFilterContext context) {
+    public void onHeartbeatRequest(short apiVersion, RequestHeaderData header, HeartbeatRequestData request, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setGroupId(tenantPrefix + request.groupId());
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onDescribeGroupsRequest(short apiVersion, RequestHeaderData header, DescribeGroupsRequestData request,
-                                                                        KrpcFilterContext context) {
+    public void onDescribeGroupsRequest(short apiVersion, RequestHeaderData header, DescribeGroupsRequestData request, KrpcFilterContext context) {
         request.setGroups(request.groups().stream().map(group -> applyTenantPrefix(context, group)).toList());
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onDescribeGroupsResponse(short apiVersion, ResponseHeaderData header, DescribeGroupsResponseData response,
-                                                                          KrpcFilterContext context) {
+    public void onDescribeGroupsResponse(short apiVersion, ResponseHeaderData header, DescribeGroupsResponseData response, KrpcFilterContext context) {
         response.groups().forEach(group -> removeTenantPrefix(context, group::groupId, group::setGroupId, false));
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onInitProducerIdRequest(short apiVersion, RequestHeaderData header, InitProducerIdRequestData request,
-                                                                        KrpcFilterContext context) {
+    public void onInitProducerIdRequest(short apiVersion, RequestHeaderData header, InitProducerIdRequestData request, KrpcFilterContext context) {
         applyTenantPrefix(context, request::transactionalId, request::setTransactionalId, true);
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onAddPartitionsToTxnRequest(short apiVersion, RequestHeaderData header, AddPartitionsToTxnRequestData request,
-                                                                            KrpcFilterContext context) {
+    public void onAddPartitionsToTxnRequest(short apiVersion, RequestHeaderData header, AddPartitionsToTxnRequestData request, KrpcFilterContext context) {
         request.v3AndBelowTopics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         applyTenantPrefix(context, request::v3AndBelowTransactionalId, request::setV3AndBelowTransactionalId, true);
 
@@ -383,12 +356,11 @@ public class MultiTenantTransformationFilter
                 applyTenantPrefix(context, addPartitionsToTxnTopic::name, addPartitionsToTxnTopic::setName, true);
             });
         });
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onAddPartitionsToTxnResponse(short apiVersion, ResponseHeaderData header, AddPartitionsToTxnResponseData response,
-                                                                              KrpcFilterContext context) {
+    public void onAddPartitionsToTxnResponse(short apiVersion, ResponseHeaderData header, AddPartitionsToTxnResponseData response, KrpcFilterContext context) {
         response.resultsByTopicV3AndBelow().forEach(results -> removeTenantPrefix(context, results::name, results::setName, false));
 
         response.resultsByTransaction().forEach(addPartitionsToTxnResult -> {
@@ -397,67 +369,61 @@ public class MultiTenantTransformationFilter
                 removeTenantPrefix(context, topicResult::name, topicResult::setName, true);
             }
         });
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onAddOffsetsToTxnRequest(short apiVersion, RequestHeaderData header, AddOffsetsToTxnRequestData request,
-                                                                         KrpcFilterContext context) {
+    public void onAddOffsetsToTxnRequest(short apiVersion, RequestHeaderData header, AddOffsetsToTxnRequestData request, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setTransactionalId(tenantPrefix + request.transactionalId());
         request.setGroupId(tenantPrefix + request.groupId());
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onTxnOffsetCommitRequest(short apiVersion, RequestHeaderData header, TxnOffsetCommitRequestData request,
-                                                                         KrpcFilterContext context) {
+    public void onTxnOffsetCommitRequest(short apiVersion, RequestHeaderData header, TxnOffsetCommitRequestData request, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setTransactionalId(tenantPrefix + request.transactionalId());
         request.setGroupId(tenantPrefix + request.groupId());
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onTxnOffsetCommitResponse(short apiVersion, ResponseHeaderData header, TxnOffsetCommitResponseData response,
-                                                                           KrpcFilterContext context) {
+    public void onTxnOffsetCommitResponse(short apiVersion, ResponseHeaderData header, TxnOffsetCommitResponseData response, KrpcFilterContext context) {
         response.topics().forEach(results -> removeTenantPrefix(context, results::name, results::setName, false));
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onListTransactionsResponse(short apiVersion, ResponseHeaderData header, ListTransactionsResponseData response,
-                                                                            KrpcFilterContext context) {
+    public void onListTransactionsResponse(short apiVersion, ResponseHeaderData header, ListTransactionsResponseData response, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         var filteredTransactions = response.transactionStates().stream().filter(listedTxn -> listedTxn.transactionalId().startsWith(tenantPrefix)).toList();
         filteredTransactions.forEach(listedTxn -> removeTenantPrefix(context, listedTxn::transactionalId, listedTxn::setTransactionalId, false));
         response.setTransactionStates(filteredTransactions);
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onDescribeTransactionsRequest(short apiVersion, RequestHeaderData header, DescribeTransactionsRequestData request,
-                                                                              KrpcFilterContext context) {
+    public void onDescribeTransactionsRequest(short apiVersion, RequestHeaderData header, DescribeTransactionsRequestData request, KrpcFilterContext context) {
         request.setTransactionalIds(request.transactionalIds().stream().map(group -> applyTenantPrefix(context, group)).toList());
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onDescribeTransactionsResponse(short apiVersion, ResponseHeaderData header, DescribeTransactionsResponseData response,
-                                                                                KrpcFilterContext context) {
+    public void onDescribeTransactionsResponse(short apiVersion, ResponseHeaderData header, DescribeTransactionsResponseData response, KrpcFilterContext context) {
         response.transactionStates().forEach(ts -> {
             removeTenantPrefix(context, ts::transactionalId, ts::setTransactionalId, false);
             ts.topics().forEach(t -> removeTenantPrefix(context, t::topic, t::setTopic, false));
         });
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onEndTxnRequest(short apiVersion, RequestHeaderData header, EndTxnRequestData request, KrpcFilterContext context) {
+    public void onEndTxnRequest(short apiVersion, RequestHeaderData header, EndTxnRequestData request, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setTransactionalId(tenantPrefix + request.transactionalId());
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     private void applyTenantPrefix(KrpcFilterContext context, Supplier<String> getter, Consumer<String> setter, boolean ignoreEmpty) {
diff --git a/kroxylicious-additional-filters/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilter.java b/kroxylicious-additional-filters/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilter.java
index db941df4..e98b4ac9 100644
--- a/kroxylicious-additional-filters/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilter.java
+++ b/kroxylicious-additional-filters/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilter.java
@@ -8,7 +8,6 @@ package io.kroxylicious.proxy.filter.schema;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.CompletionStage;
 
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceResponseData;
@@ -21,8 +20,6 @@ import org.slf4j.LoggerFactory;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 import io.kroxylicious.proxy.filter.schema.validation.request.ProduceRequestValidationResult;
 import io.kroxylicious.proxy.filter.schema.validation.request.ProduceRequestValidator;
 import io.kroxylicious.proxy.filter.schema.validation.topic.PartitionValidationResult;
@@ -65,22 +62,21 @@ public class ProduceValidationFilter implements ProduceRequestFilter, ProduceRes
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         ProduceRequestValidationResult result = validator.validateRequest(request);
         if (result.isAnyTopicPartitionInvalid()) {
-            return handleInvalidTopicPartitions(header, request, context, result);
+            handleInvalidTopicPartitions(header, request, context, result);
         }
         else {
-            return context.forwardRequest(header, request);
+            context.forwardRequest(header, request);
         }
     }
 
-    private CompletionStage<RequestFilterResult> handleInvalidTopicPartitions(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context,
-                                                                              ProduceRequestValidationResult result) {
+    private void handleInvalidTopicPartitions(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context, ProduceRequestValidationResult result) {
         if (result.isAllTopicPartitionsInvalid()) {
             LOGGER.debug("all topic-partitions for request contained invalid data: {}", result);
             ProduceResponseData response = invalidateEntireRequest(request, result);
-            return context.requestFilterResultBuilder().shortCircuitResponse(response).completed();
+            context.forwardResponse(response);
         }
         // do not forward partial produce data if request is transactional because the whole produce must eventually succeed or fail together
         else if (request.transactionalId() == null && forwardPartialRequests) {
@@ -90,13 +86,13 @@ public class ProduceValidationFilter implements ProduceRequestFilter, ProduceRes
                 topicDatum.partitionData().removeIf(partitionProduceData -> !result.isPartitionValid(topicDatum.name(), partitionProduceData.index()));
             }
             correlatedResults.put(header.correlationId(), result);
-            return context.forwardRequest(header, request);
+            context.forwardRequest(header, request);
         }
         else {
             LOGGER.debug("some topic-partitions for transactional request with id: {}, contained invalid data: {}, invalidation entire request",
                     request.transactionalId(), result);
             ProduceResponseData response = invalidateEntireRequest(request, result);
-            return context.requestFilterResultBuilder().shortCircuitResponse(response).completed();
+            context.forwardResponse(response);
         }
     }
 
@@ -149,16 +145,15 @@ public class ProduceValidationFilter implements ProduceRequestFilter, ProduceRes
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
-                                                                   KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
         ProduceRequestValidationResult produceRequestValidationResult = correlatedResults.remove(header.correlationId());
         if (produceRequestValidationResult != null) {
             LOGGER.debug("augmenting invalid topic-partition details into response: {}", produceRequestValidationResult);
             augmentResponseWithInvalidTopicPartitions(response, produceRequestValidationResult);
-            return context.forwardResponse(header, response);
+            context.forwardResponse(response);
         }
         else {
-            return context.forwardResponse(header, response);
+            context.forwardResponse(response);
         }
     }
 
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvoker.java
index a70cb447..21b6cbc6 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvoker.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvoker.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -17,7 +15,7 @@ import org.apache.kafka.common.protocol.ApiMessage;
  * The FilterInvoker connects Kroxylicious with the concrete implementation of a KrpcFilter.
  * <p>When handling a message, we want to avoid the penalty of deserializing the bytes
  * into an ApiMessage. When Kroxylicious receives a message, all the Filters in the
- * filter chain will be consulted (via their invoker) to see if any want to handle that message.
+ * filter chainnwill be consulted (via their invoker) to see if any want to handle that message.
  * If any filter wants to handle it, then the message will be deserialized. Then
  * onRequest|onResponse will be eligible to be called in the filter chain for that
  * message (i.e. if filter A wants to handle request Y but filter B doesn't, only the
@@ -74,49 +72,34 @@ public interface FilterInvoker {
      * <p>Handle deserialized request data. It is implicit that the underlying filter
      * wants to handle this data because it indicated that with {@link #shouldHandleRequest(ApiKeys, short)}
      * </p><p>
-     * Filters must return a {@link CompletionStage<RequestFilterResult>} object.  This object
-     * encapsulates the request to be forwarded and, optionally, orders for actions such as closing the connection or
-     * dropping the request.
-     * </p><p>
-     * The {@link KrpcFilterContext} is the factory for {@link FilterResult} objects.  See
-     * {@link KrpcFilterContext#forwardRequest(RequestHeaderData, ApiMessage)} and
-     * {@link KrpcFilterContext#requestFilterResultBuilder()} for more details.
+     * Most Filters will want to call a method on {@link KrpcFilterContext} like {@link KrpcFilterContext#forwardRequest(RequestHeaderData, ApiMessage)}
+     * so that the message continues to flow through the filter chain.
      * </p>
-     * @param apiKey        the key of the message
-     * @param apiVersion    the apiVersion of the message
-     * @param header        the header of the message
-     * @param body          the body of the message
+     * @param apiKey the key of the message
+     * @param apiVersion the apiVersion of the message
+     * @param header the header of the message
+     * @param body the body of the message
      * @param filterContext contains methods to continue the filter chain and other contextual data
-     * @return a {@link CompletionStage<RequestFilterResult>}, that when complete, will yield the request to forward.
      */
-    default CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body,
-                                                           KrpcFilterContext filterContext) {
-        return filterContext.requestFilterResultBuilder().forward(header, body).completed();
+    default void onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        filterContext.forwardRequest(header, body);
     }
 
     /**
      * <p>Handle deserialized response data. It is implicit that the underlying filter
      * wants to handle this data because it indicated that with {@link #shouldHandleResponse(ApiKeys, short)}
      * </p><p>
-     * Filters must return a {@link CompletionStage<ResponseFilterResult>} object.  This object
-     * encapsulates the response to be forwarded and, optionally, orders for actions such as closing the connection or
-     * dropping the response.
-     * </p><p>
-     * The {@link KrpcFilterContext} is the factory for {@link FilterResult} objects.  See
-     * {@link KrpcFilterContext#forwardResponse(ResponseHeaderData, ApiMessage)} and
-     * {@link KrpcFilterContext#responseFilterResultBuilder()} for more details.
+     * Most Filters will want to call a method on {@link KrpcFilterContext} like {@link KrpcFilterContext#forwardResponse(ResponseHeaderData, ApiMessage)}
+     * so that the message continues to flow through the filter chain.
      * </p>
-     *
-     * @param apiKey        the key of the message
-     * @param apiVersion    the apiVersion of the message
-     * @param header        the header of the message
-     * @param body          the body of the message
+     * @param apiKey the key of the message
+     * @param apiVersion the apiVersion of the message
+     * @param header the header of the message
+     * @param body the body of the message
      * @param filterContext contains methods to continue the filter chain and other contextual data
-     * @return a {@link CompletionStage<ResponseFilterResult>}, that when complete, will yield the response to forward.
      */
-    default CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body,
-                                                             KrpcFilterContext filterContext) {
-        return filterContext.responseFilterResultBuilder().forward(header, body).completed();
+    default void onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        filterContext.forwardResponse(header, body);
     }
 
 }
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterResult.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterResult.java
deleted file mode 100644
index 64170a13..00000000
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterResult.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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.filter;
-
-import org.apache.kafka.common.protocol.ApiMessage;
-
-/**
- * The result of a filter request or response operation that encapsulates the request or response
- * to be forwarded to the next filter in the chain.  Optionally it carries orders for actions such
- * as close the connection or drop the message.
- *
- * @see FilterResultBuilder
- */
-public interface FilterResult {
-    /**
-     * the header to be forwarded to the next filter in the chain.
-     *
-     * @return header.
-     */
-    ApiMessage header();
-
-    /**
-     * the message to be forwarded to the next filter in the chain.
-     * @return header.
-     */
-    ApiMessage message();
-
-    /**
-     * signals the filter's wish that the connection will be closed. if the filter provides a
-     * message it will be forwarded before the connection is closed.
-     * @return true if the connection is to be closed.
-     */
-    boolean closeConnection();
-
-    /**
-     * signals the filter's wish that message is dropped i.e. not forward to the next filter
-     * in the chain.
-     * @return true if message is to be dropped.
-     */
-    boolean drop();
-}
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterResultBuilder.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterResultBuilder.java
deleted file mode 100644
index 4b789e6f..00000000
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterResultBuilder.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.filter;
-
-import org.apache.kafka.common.protocol.ApiMessage;
-
-import io.kroxylicious.proxy.filter.filterresultbuilder.CloseOrTerminalStage;
-import io.kroxylicious.proxy.filter.filterresultbuilder.CloseStage;
-import io.kroxylicious.proxy.filter.filterresultbuilder.TerminalStage;
-
-/**
- * Fluent builder for filter results.
- *
- * @param <H>  kafka api message header class
- * @param <FR> filter result
- */
-public interface FilterResultBuilder<H extends ApiMessage, FR extends FilterResult> extends CloseStage<FR> {
-
-    /**
-     * A forward of a request or response message to the next filter in the chain.
-     *
-     * @param header message header. May not be null.
-     * @param message api message. May not be null.  for request messages the class must have a name
-     *                that that ends with RequestData. for response messages the class must have one
-     *                that ends with ResponseData.
-     * @return next stage in the fluent builder API
-     * @throws IllegalArgumentException header or message do not meet criteria described above.
-     */
-    CloseOrTerminalStage<FR> forward(H header, ApiMessage message) throws IllegalArgumentException;
-
-    /**
-     * Signals the desire of the filter that the connection is closed.
-     *
-     * @return last stage in the fluent API.
-     */
-    TerminalStage<FR> drop();
-
-}
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/HandleNothingFilterInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/HandleNothingFilterInvoker.java
index dbe5d6dc..dc9c00c5 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/HandleNothingFilterInvoker.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/HandleNothingFilterInvoker.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -17,12 +15,12 @@ record HandleNothingFilterInvoker() implements FilterInvoker {
     static final FilterInvoker INSTANCE = new HandleNothingFilterInvoker();
 
     @Override
-    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public void onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         throw new IllegalStateException("I should never be invoked");
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public void onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         throw new IllegalStateException("I should never be invoked");
     }
 }
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
index 3cf00ea7..18688b31 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
@@ -38,32 +38,12 @@ public interface KrpcFilterContext {
     String sniHostname();
 
     /**
-     * Creates a builder for a request filter result objects.  This object encapsulates
-     * the request to forward and optionally orders for actions such as closing
-     * the connection or dropping the request.
-     * <br/>
-     * The builder returns either {@link CompletionStage<RequestFilterResult>} object
-     * ready to be returned by the request filter, or a {@link ResponseFilterResult} object.
-     * The latter facilitates asynchronous programming patterns where requests must be
-     * forwarded after other work has completed.
-     *
-     * @return builder
-     */
-    RequestFilterResultBuilder requestFilterResultBuilder();
-
-    /**
-     * Generates a completed filter results containing the given header and request.  When
-     * request filters implementations return this result, the request will be sent towards
-     * the broker, invoking upstream filters.
-     * <br/>
-     * Invoking this method is identical to invoking:
-     * {@code requestFilterResultBuilder.forward(header, request).complete()}
+     * Send a request towards the broker, invoking upstream filters.
      *
      * @param header The header to forward to the broker.
      * @param request The request to forward to the broker.
-     * @return completed filter results.
      */
-    CompletionStage<RequestFilterResult> forwardRequest(RequestHeaderData header, ApiMessage request);
+    void forwardRequest(RequestHeaderData header, ApiMessage request);
 
     /**
      * Send a message from a filter towards the broker, invoking upstream filters
@@ -80,32 +60,35 @@ public interface KrpcFilterContext {
     <T extends ApiMessage> CompletionStage<T> sendRequest(short apiVersion, ApiMessage request);
 
     /**
-     * Generates a completed filter results containing the given header and response.  When
-     * response filters implementations return this result, the response will be sent towards
-     * the client, invoking downstream filters.
-     * <br/>
-     * Invoking this method is identical to invoking:
-     * {@code responseFilterResultBuilder.forward(header, response).complete()}
-     *
-     * @param header The header to forward to the broker.
-     * @param response The request to forward to the broker.
-     * @return completed filter results.
+     * Send a response towards the client, invoking downstream filters.
+     * <p>If this is invoked while the message is flowing downstream towards the broker, then
+     * it will not be sent to the broker. So this method can be used to generate responses in
+     * the proxy.</p>
+     * @param header The header to forward to the client.
+     * @param response The response to forward to the client.
+     * @throws AssertionError if response is logically inconsistent, for example responding with request data
+     * or responding with a produce response to a fetch request. It is up to specific implementations to
+     * determine what logically inconsistent means.
      */
-    CompletionStage<ResponseFilterResult> forwardResponse(ResponseHeaderData header, ApiMessage response);
+    void forwardResponse(ResponseHeaderData header, ApiMessage response);
 
     /**
-     * Creates a builder for a request filter result objects.  This object encapsulates
-     * the response to forward and optionally orders for actions such as closing
-     * the connection or dropping the response.
-     * <br/>
-     * The builder returns either {@link CompletionStage<ResponseFilterResult>} object
-     * ready to be returned by the response filter, or a {@link ResponseFilterResult} object.
-     * The latter facilitates asynchronous programming patterns where responses must be
-     * forwarded after other work has completed.
-     *
-     * @return builder
+     * Send a response towards the client, invoking downstream filters.
+     * <p>If this is invoked while the message is flowing downstream towards the broker, then
+     * it will not be sent to the broker. So this method can be used to generate responses in
+     * the proxy. In this case response headers will be created with a correlationId matching the request</p>
+     * @param response The response to forward to the client.
+     * @throws AssertionError if response is logically inconsistent, for example responding with request data
+     * or responding with a produce response to a fetch request. It is up to specific implementations to
+     * determine what logically inconsistent means.
+     */
+    void forwardResponse(ApiMessage response);
+
+    /**
+     * Allows a filter to decide to close the connection. The client will be disconnected.  The client is free
+     * to retry the connection.  The client will receive no indication why the connection was closed.
      */
-    ResponseFilterResultBuilder responseFilterResultBuilder();
+    void closeConnection();
 
     // TODO an API to allow a filter to add/remove another filter from the pipeline
 }
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 c761a75e..7dc78aea 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
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ApiMessage;
@@ -37,16 +35,13 @@ public interface RequestFilter extends KrpcFilter {
 
     /**
      * Handle the request
-     *
-     * @param apiKey        key of the request
-     * @param header        header of the request
-     * @param body          body of the request
+     * @param apiKey key of the request
+     * @param header header of the request
+     * @param body body of the request
      * @param filterContext context containing methods to continue the filter chain and other contextual data
-     * @return CompletionStage that, when complete, will yield a RequestFilterResult containing the
-     *         request to be forwarded.
      */
-    CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey,
-                                                   RequestHeaderData header,
-                                                   ApiMessage body,
-                                                   KrpcFilterContext filterContext);
+    void onRequest(ApiKeys apiKey,
+                   RequestHeaderData header,
+                   ApiMessage body,
+                   KrpcFilterContext filterContext);
 }
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterInvoker.java
index 20f1a969..ba1d5f5c 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterInvoker.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterInvoker.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ApiMessage;
@@ -15,8 +13,8 @@ import org.apache.kafka.common.protocol.ApiMessage;
 record RequestFilterInvoker(RequestFilter filter) implements FilterInvoker {
 
     @Override
-    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-        return filter.onRequest(apiKey, header, body, filterContext);
+    public void onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        filter.onRequest(apiKey, header, body, filterContext);
     }
 
     @Override
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterResult.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterResult.java
deleted file mode 100644
index 5db3d065..00000000
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterResult.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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.filter;
-
-/**
- * A specialization of the {@link FilterResult} for request filters.
- * <br>
- * A request filter may, rather than forwarding a request towards the broker, opt to
- * send a response towards the client instead. This is called a short circuit response. It
- * is useful for implementing validating filters.
- */
-public interface RequestFilterResult extends FilterResult {
-
-    /**
-     * If true, the request is carrying a short circuit response.
-     *
-     * @return true if carrying a short circuit response.
-     */
-    boolean shortCircuitResponse();
-}
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterResultBuilder.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterResultBuilder.java
deleted file mode 100644
index 4ad85930..00000000
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterResultBuilder.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.filter;
-
-import org.apache.kafka.common.message.RequestHeaderData;
-import org.apache.kafka.common.message.ResponseHeaderData;
-import org.apache.kafka.common.protocol.ApiMessage;
-
-import io.kroxylicious.proxy.filter.filterresultbuilder.CloseOrTerminalStage;
-
-/**
- * Builder for request filter results.
- * <br/>
- * See {@link RequestFilterResult} for a description of short-circuit responses.
- */
-public interface RequestFilterResultBuilder extends FilterResultBuilder<RequestHeaderData, RequestFilterResult> {
-
-    /**
-     * A short-circuit response towards the client.
-     *
-     * @param header response header. May be null.
-     * @param message response message. May not be null.  the response messages the class must have one
-     *                that ends with ResponseData.
-     * @return next stage in the fluent builder API
-     * @throws IllegalArgumentException header or message do not meet criteria described above.
-     */
-    CloseOrTerminalStage<RequestFilterResult> shortCircuitResponse(ResponseHeaderData header, ApiMessage message) throws IllegalArgumentException;
-
-    /**
-     * A short-circuit response towards the client.
-     *
-     * @param message response message. May not be null.  the response messages the class must have one
-     *                that ends with ResponseData.
-     * @return next stage in the fluent builder API
-     * @throws IllegalArgumentException header or message do not meet criteria described above.
-     */
-    CloseOrTerminalStage<RequestFilterResult> shortCircuitResponse(ApiMessage message) throws IllegalArgumentException;
-
-}
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestResponseInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestResponseInvoker.java
index 4acd982e..1338f604 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestResponseInvoker.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestResponseInvoker.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -16,13 +14,13 @@ import org.apache.kafka.common.protocol.ApiMessage;
 record RequestResponseInvoker(RequestFilter requestFilter, ResponseFilter responseFilter) implements FilterInvoker {
 
     @Override
-    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-        return requestFilter.onRequest(apiKey, header, body, filterContext);
+    public void onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        requestFilter.onRequest(apiKey, header, body, filterContext);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-        return responseFilter.onResponse(apiKey, header, body, filterContext);
+    public void onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        responseFilter.onResponse(apiKey, header, body, filterContext);
     }
 
     @Override
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 dffd64e4..e5ffaea9 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
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ApiMessage;
@@ -41,11 +39,9 @@ public interface ResponseFilter extends KrpcFilter {
      * @param header header of the response
      * @param body body of the response
      * @param filterContext context containing methods to continue the filter chain and other contextual data
-     * @return CompletionStage that, when complete, will yield a ResponseFilterResult containing the
-     *         response to be forwarded.
      */
-    CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey,
-                                                     ResponseHeaderData header,
-                                                     ApiMessage body,
-                                                     KrpcFilterContext filterContext);
+    void onResponse(ApiKeys apiKey,
+                    ResponseHeaderData header,
+                    ApiMessage body,
+                    KrpcFilterContext filterContext);
 }
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterInvoker.java
index 1d6d16cd..c0bc7746 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterInvoker.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterInvoker.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ApiMessage;
@@ -15,8 +13,8 @@ import org.apache.kafka.common.protocol.ApiMessage;
 record ResponseFilterInvoker(ResponseFilter filter) implements FilterInvoker {
 
     @Override
-    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-        return filter.onResponse(apiKey, header, body, filterContext);
+    public void onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        filter.onResponse(apiKey, header, body, filterContext);
     }
 
     @Override
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterResult.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterResult.java
deleted file mode 100644
index 6fc44128..00000000
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterResult.java
+++ /dev/null
@@ -1,14 +0,0 @@
-/*
- * 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.filter;
-
-/**
- * A specialization of the {@link FilterResult} for response filters.
- */
-public interface ResponseFilterResult extends FilterResult {
-
-}
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterResultBuilder.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterResultBuilder.java
deleted file mode 100644
index 3d8930f3..00000000
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterResultBuilder.java
+++ /dev/null
@@ -1,15 +0,0 @@
-/*
- * 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.filter;
-
-import org.apache.kafka.common.message.ResponseHeaderData;
-
-/**
- * Builder for response filter results.
- */
-public interface ResponseFilterResultBuilder extends FilterResultBuilder<ResponseHeaderData, ResponseFilterResult> {
-}
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 2be85d13..209c4a2f 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
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -22,22 +20,22 @@ import org.apache.kafka.common.protocol.ApiMessage;
 record SafeInvoker(FilterInvoker invoker) implements FilterInvoker {
 
     @Override
-    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public void onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         if (invoker.shouldHandleRequest(apiKey, apiVersion)) {
-            return invoker.onRequest(apiKey, apiVersion, header, body, filterContext);
+            invoker.onRequest(apiKey, apiVersion, header, body, filterContext);
         }
         else {
-            return filterContext.forwardRequest(header, body);
+            filterContext.forwardRequest(header, body);
         }
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public void onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         if (invoker.shouldHandleResponse(apiKey, apiVersion)) {
-            return invoker.onResponse(apiKey, apiVersion, header, body, filterContext);
+            invoker.onResponse(apiKey, apiVersion, header, body, filterContext);
         }
         else {
-            return filterContext.responseFilterResultBuilder().forward(header, body).completed();
+            filterContext.forwardResponse(header, body);
         }
     }
 
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/filterresultbuilder/CloseOrTerminalStage.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/filterresultbuilder/CloseOrTerminalStage.java
deleted file mode 100644
index 873714db..00000000
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/filterresultbuilder/CloseOrTerminalStage.java
+++ /dev/null
@@ -1,17 +0,0 @@
-/*
- * 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.filter.filterresultbuilder;
-
-import io.kroxylicious.proxy.filter.FilterResult;
-
-/**
- * Interface supporting the {@link io.kroxylicious.proxy.filter.FilterResultBuilder} fluent API.
- *
- * @param <FR> filter result
- */
-public interface CloseOrTerminalStage<FR extends FilterResult> extends TerminalStage<FR>, CloseStage<FR> {
-}
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/filterresultbuilder/CloseStage.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/filterresultbuilder/CloseStage.java
deleted file mode 100644
index acaa9e07..00000000
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/filterresultbuilder/CloseStage.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * 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.filter.filterresultbuilder;
-
-import io.kroxylicious.proxy.filter.FilterResult;
-
-/**
- * Interface supporting the {@link io.kroxylicious.proxy.filter.FilterResultBuilder} fluent API.
- *
- * @param <FR> filter result
- */
-public interface CloseStage<FR extends FilterResult> {
-    /**
-     * Signals the desire of the filter that the connection is closed.
-     *
-     * @return next stage in the fluent API.
-     */
-    TerminalStage<FR> withCloseConnection();
-}
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/filterresultbuilder/TerminalStage.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/filterresultbuilder/TerminalStage.java
deleted file mode 100644
index 4603d99e..00000000
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/filterresultbuilder/TerminalStage.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.filter.filterresultbuilder;
-
-import java.util.concurrent.CompletionStage;
-
-import io.kroxylicious.proxy.filter.FilterResult;
-
-/**
- * Interface supporting the {@link io.kroxylicious.proxy.filter.FilterResultBuilder} fluent API.
- *
- * @param <FR> filter result
- */
-public interface TerminalStage<FR extends FilterResult> {
-    /**
-     * Constructs the filter result.
-     *
-     * @return filter result
-     */
-    FR build();
-
-    /**
-     * Produces the filter result contained within a completed {@link CompletionStage}.
-     * @return completion stage contain the filter result.
-     */
-    CompletionStage<FR> completed();
-}
diff --git a/kroxylicious-api/src/main/templates/Kproxy/Filter.ftl b/kroxylicious-api/src/main/templates/Kproxy/Filter.ftl
index e20b57f4..a2a20feb 100644
--- a/kroxylicious-api/src/main/templates/Kproxy/Filter.ftl
+++ b/kroxylicious-api/src/main/templates/Kproxy/Filter.ftl
@@ -8,8 +8,6 @@
 <#assign
   dataClass="${messageSpec.name}Data"
   filterClass="${messageSpec.name}Filter"
-  filterResultClass="${messageSpec.type?lower_case?cap_first}FilterResult"
-  headerClass="${messageSpec.type?lower_case?cap_first}HeaderData"
   msgType=messageSpec.type?lower_case
 />
 /*
@@ -30,8 +28,6 @@
  */
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.${messageSpec.name}Data;
 <#if messageSpec.type?lower_case == 'response'>
 import org.apache.kafka.common.message.ResponseHeaderData;
@@ -59,7 +55,7 @@ public interface ${filterClass} extends KrpcFilter {
 
     /**
      * Handle the given {@code ${msgType}},
-     * returning the {@code ${dataClass}} instance to be passed to the next filter.
+     * returning the {@code ${messageSpec.name}Data} instance to be passed to the next filter.
      * The implementation may modify the given {@code data} in-place and return it,
      * or instantiate a new one.
      *
@@ -67,9 +63,7 @@ public interface ${filterClass} extends KrpcFilter {
      * @param header <#if messageSpec.type?lower_case == 'response'>response<#else>request</#if> header.
      * @param ${msgType} The KRPC message to handle.
      * @param context The context.
-     * @return CompletionStage that, when complete, will yield a ${filterResultClass} containing the
-     *         ${messageSpec.type?lower_case} to be forwarded.
      */
-     CompletionStage<${filterResultClass}> on${messageSpec.name}(short apiVersion, ${headerClass} header, ${dataClass} ${msgType}, KrpcFilterContext context);
+    void on${messageSpec.name}(short apiVersion, <#if messageSpec.type?lower_case == 'response'>ResponseHeaderData<#else>RequestHeaderData</#if> header, ${dataClass} ${msgType}, KrpcFilterContext context);
 
 }
\ No newline at end of file
diff --git a/kroxylicious-api/src/main/templates/Kproxy/FilterInvoker.ftl b/kroxylicious-api/src/main/templates/Kproxy/FilterInvoker.ftl
index 18db9847..7bfd81fb 100644
--- a/kroxylicious-api/src/main/templates/Kproxy/FilterInvoker.ftl
+++ b/kroxylicious-api/src/main/templates/Kproxy/FilterInvoker.ftl
@@ -8,8 +8,6 @@
 <#assign
   dataClass="${messageSpec.name}Data"
   filterClass="${messageSpec.name}Filter"
-  filterResultClass="${messageSpec.type?lower_case?cap_first}FilterResult"
-  headerClass="${messageSpec.type?lower_case?cap_first}HeaderData"
   filterInvokerClass="${messageSpec.name}FilterInvoker"
   msgType=messageSpec.type?lower_case
 />
@@ -31,8 +29,6 @@
  */
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.${messageSpec.name}Data;
 
 <#if messageSpec.type?lower_case == 'response'>
@@ -62,7 +58,7 @@ class ${filterInvokerClass} implements FilterInvoker {
     }
 
     @Override
-    public CompletionStage<${filterResultClass}> on<#if messageSpec.type?lower_case == 'response'>Response<#else>Request</#if>(ApiKeys apiKey, short apiVersion, ${headerClass} header, ApiMessage body, KrpcFilterContext filterContext) {
-        return filter.on${messageSpec.name}(apiVersion, header, (${dataClass}) body, filterContext);
+    public void on<#if messageSpec.type?lower_case == 'response'>Response<#else>Request</#if>(ApiKeys apiKey, short apiVersion, <#if messageSpec.type?lower_case == 'response'>Response<#else>Request</#if>HeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        filter.on${messageSpec.name}(apiVersion, header, (${messageSpec.name}Data) body, filterContext);
     }
 }
\ No newline at end of file
diff --git a/kroxylicious-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl b/kroxylicious-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl
index 4056d7df..6ac5be2c 100644
--- a/kroxylicious-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl
+++ b/kroxylicious-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl
@@ -23,8 +23,6 @@
  */
 package ${outputPackage};
 
-import java.util.concurrent.CompletionStage;
-
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
@@ -69,14 +67,14 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
      */
     @Override
     @SuppressWarnings("SwitchStatementWithTooFewBranches")
-    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey,
-                                                          short apiVersion,
-                                                          RequestHeaderData header,
-                                                          ApiMessage body,
-                                                          KrpcFilterContext filterContext) {
+    public void onRequest(ApiKeys apiKey,
+                           short apiVersion,
+                           RequestHeaderData header,
+                           ApiMessage body,
+                           KrpcFilterContext filterContext) {
         // We wrap the array lookup in a switch based on the API Key as it supports JIT optimisations around method dispatch.
         // See the InvokerDispatchBenchmark micro benchmark for a comparison
-        return switch (apiKey) {
+        switch (apiKey) {
 <#list messageSpecs as messageSpec>
     <#if messageSpec.type?lower_case == 'request'>
             case ${retrieveApiKey(messageSpec)} ->
@@ -84,7 +82,7 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
     </#if>
 </#list>
             default -> throw new IllegalStateException("Unsupported RPC " + apiKey);
-        };
+        }
 
     }
 
@@ -98,14 +96,14 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
      */
     @Override
     @SuppressWarnings("SwitchStatementWithTooFewBranches")
-    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey,
-                                                            short apiVersion,
-                                                            ResponseHeaderData header,
-                                                            ApiMessage body,
-                                                            KrpcFilterContext filterContext) {
+    public void onResponse(ApiKeys apiKey,
+                            short apiVersion,
+                            ResponseHeaderData header,
+                            ApiMessage body,
+                            KrpcFilterContext filterContext) {
         // We wrap the array lookup in a switch based on the API Key as it supports JIT optimisations around method dispatch.
         // See the InvokerDispatchBenchmark micro benchmark for a comparison
-        return switch (apiKey) {
+        switch (apiKey) {
 <#list messageSpecs as messageSpec>
     <#if messageSpec.type?lower_case == 'response'>
             case ${retrieveApiKey(messageSpec)} ->
@@ -113,7 +111,7 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
     </#if>
 </#list>
             default -> throw new IllegalStateException("Unsupported RPC " + apiKey);
-        };
+        }
 
     }
 
diff --git a/kroxylicious-sample/pom.xml b/kroxylicious-sample/pom.xml
index 523cced8..23aac130 100644
--- a/kroxylicious-sample/pom.xml
+++ b/kroxylicious-sample/pom.xml
@@ -61,11 +61,6 @@
             <artifactId>mockito-core</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.mockito</groupId>
-            <artifactId>mockito-junit-jupiter</artifactId>
-            <scope>test</scope>
-        </dependency>
     </dependencies>
 
 </project>
\ No newline at end of file
diff --git a/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleFetchResponseFilter.java b/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleFetchResponseFilter.java
index e38b618a..4364dd64 100644
--- a/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleFetchResponseFilter.java
+++ b/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleFetchResponseFilter.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.sample;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 
@@ -16,7 +14,6 @@ import io.micrometer.core.instrument.Timer;
 
 import io.kroxylicious.proxy.filter.FetchResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 import io.kroxylicious.sample.config.SampleFilterConfig;
 import io.kroxylicious.sample.util.SampleFilterTransformer;
 
@@ -51,17 +48,15 @@ public class SampleFetchResponseFilter implements FetchResponseFilter {
     /**
      * Handle the given response, transforming the data in-place according to the configuration, forwarding
      * the FetchResponseData instance onward.
-     *
      * @param apiVersion the apiVersion of the response
-     * @param header     response header.
-     * @param response   The KRPC message to handle.
-     * @param context    The context.
-     * @return CompletionStage that will yield a {@link ResponseFilterResult} containing the response to be forwarded.
+     * @param header response header.
+     * @param response The KRPC message to handle.
+     * @param context The context.
      */
     @Override
-    public CompletionStage<ResponseFilterResult> onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
+    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
         this.timer.record(() -> applyTransformation(response, context)); // We're timing this to report how long it takes through Micrometer
-        return context.forwardResponse(header, response);
+        context.forwardResponse(header, response);
     }
 
     /**
diff --git a/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleProduceRequestFilter.java b/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleProduceRequestFilter.java
index 1b54800a..f80f9bf9 100644
--- a/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleProduceRequestFilter.java
+++ b/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleProduceRequestFilter.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.sample;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceRequestData.PartitionProduceData;
 import org.apache.kafka.common.message.RequestHeaderData;
@@ -17,7 +15,6 @@ import io.micrometer.core.instrument.Timer;
 
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
 import io.kroxylicious.sample.config.SampleFilterConfig;
 import io.kroxylicious.sample.util.SampleFilterTransformer;
 
@@ -52,18 +49,15 @@ public class SampleProduceRequestFilter implements ProduceRequestFilter {
     /**
      * Handle the given request, transforming the data in-place according to the configuration, and returning
      * the ProduceRequestData instance to be passed to the next filter.
-     *
      * @param apiVersion the apiVersion of the request
-     * @param header     request header.
-     * @param request    The KRPC message to handle.
-     * @param context    The context.
-     * @return CompletionStage that will yield a {@link RequestFilterResult} containing the request to be forwarded.
+     * @param header request header.
+     * @param request The KRPC message to handle.
+     * @param context The context.
      */
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         this.timer.record(() -> applyTransformation(request, context)); // We're timing this to report how long it takes through Micrometer
-
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     /**
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/NetFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/NetFilter.java
index 8bad87a5..360b8b06 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/NetFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/NetFilter.java
@@ -58,12 +58,12 @@ public interface NetFilter {
         String authorizedId();
 
         /**
-         * @return The name of the client software, if known via ApiVersions request. Otherwise, null.
+         * @return The name of the client software, if known via ApiVersions request. Otherwise null.
          */
         String clientSoftwareName();
 
         /**
-         * @return The version of the client software, if known via ApiVersions request. Otherwise, null.
+         * @return The version of the client software, if known via ApiVersions request. Otherwise null.
          */
         String clientSoftwareVersion();
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
index 7ceb28f0..3fef507b 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
@@ -25,16 +25,10 @@ import io.netty.channel.ChannelPromise;
 
 import io.kroxylicious.proxy.filter.KrpcFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.RequestFilterResultBuilder;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResultBuilder;
 import io.kroxylicious.proxy.frame.DecodedFrame;
 import io.kroxylicious.proxy.frame.DecodedResponseFrame;
 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.ByteBufOutputStream;
 
 /**
@@ -99,7 +93,8 @@ class DefaultFilterContext implements KrpcFilterContext {
      * @param header The header
      * @param message The message
      */
-    protected void forwardRequestInternal(RequestHeaderData header, ApiMessage message) {
+    @Override
+    public void forwardRequest(RequestHeaderData header, ApiMessage message) {
         if (decodedFrame.body() != message) {
             throw new IllegalStateException();
         }
@@ -172,11 +167,6 @@ class DefaultFilterContext implements KrpcFilterContext {
         return filterStage;
     }
 
-    @Override
-    public CompletionStage<ResponseFilterResult> forwardResponse(ResponseHeaderData header, ApiMessage response) {
-        return responseFilterResultBuilder().forward(header, response).completed();
-    }
-
     /**
      * Forward a request to the next filter in the chain
      * (or to the downstream client).
@@ -184,7 +174,8 @@ class DefaultFilterContext implements KrpcFilterContext {
      * @param header The header
      * @param response The message
      */
-    protected void forwardResponseInternal(ResponseHeaderData header, ApiMessage response) {
+    @Override
+    public void forwardResponse(ResponseHeaderData header, ApiMessage response) {
         // check it's a response
         String name = response.getClass().getName();
         if (!name.endsWith("ResponseData")) {
@@ -209,21 +200,17 @@ class DefaultFilterContext implements KrpcFilterContext {
     }
 
     @Override
-    public ResponseFilterResultBuilder responseFilterResultBuilder() {
-        return new ResponseFilterResultBuilderImpl();
-    }
-
-    @Override
-    public CompletionStage<RequestFilterResult> forwardRequest(RequestHeaderData header, ApiMessage request) {
-        return requestFilterResultBuilder().forward(header, request).completed();
+    public void forwardResponse(ApiMessage response) {
+        if (decodedFrame instanceof RequestFrame) {
+            this.forwardResponse(new ResponseHeaderData().setCorrelationId(decodedFrame.correlationId()), response);
+        }
+        else {
+            this.forwardResponse((ResponseHeaderData) decodedFrame.header(), response);
+        }
     }
 
     @Override
-    public RequestFilterResultBuilder requestFilterResultBuilder() {
-        return new RequestFilterResultBuilderImpl();
-    }
-
-    protected void closeConnection() {
+    public void closeConnection() {
         this.channelContext.close().addListener(future -> {
             LOGGER.debug("{} closed.", channelDescriptor());
         });
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 bed0a1f1..b881a4b5 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
@@ -8,8 +8,6 @@ package io.kroxylicious.proxy.internal;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 
-import org.apache.kafka.common.message.RequestHeaderData;
-import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiMessage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -32,7 +30,8 @@ import io.kroxylicious.proxy.internal.util.Assertions;
  * A {@code ChannelInboundHandler} (for handling requests from downstream)
  * that applies a single {@link KrpcFilter}.
  */
-public class FilterHandler extends ChannelDuplexHandler {
+public class FilterHandler
+        extends ChannelDuplexHandler {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(FilterHandler.class);
     private final KrpcFilter filter;
@@ -59,49 +58,7 @@ public class FilterHandler extends ChannelDuplexHandler {
                 LOGGER.debug("{}: Dispatching downstream {} request to filter{}: {}",
                         ctx.channel(), decodedFrame.apiKey(), filterDescriptor(), msg);
             }
-
-            var stage = invoker.onRequest(decodedFrame.apiKey(), decodedFrame.apiVersion(), decodedFrame.header(),
-                    decodedFrame.body(), filterContext);
-            if (stage == null) {
-                LOGGER.error("{}: Filter{} for {} request unexpectedly returned null. This is a coding error in the filter. Closing connection.",
-                        ctx.channel(), filterDescriptor(), decodedFrame.apiKey());
-                filterContext.closeConnection();
-                return;
-            }
-            stage.whenComplete((requestFilterResult, t) -> {
-                // maybe better to run the whole thing on the netty thread.
-
-                if (t != null) {
-                    LOGGER.warn("{}: Filter{} for {} request ended exceptionally - closing connection",
-                            ctx.channel(), filterDescriptor(), decodedFrame.apiKey(), t);
-                    filterContext.closeConnection();
-                    return;
-                }
-
-                if (requestFilterResult.drop()) {
-                    if (LOGGER.isDebugEnabled()) {
-                        LOGGER.debug("{}: Filter{} drops {} request",
-                                ctx.channel(), filterDescriptor(), decodedFrame.apiKey());
-                    }
-                }
-                else if (requestFilterResult.message() != null) {
-                    if (requestFilterResult.shortCircuitResponse()) {
-                        // this is the short circuit path
-                        var header = requestFilterResult.header() == null ? new ResponseHeaderData() : ((ResponseHeaderData) requestFilterResult.header());
-                        header.setCorrelationId(decodedFrame.correlationId());
-                        filterContext.forwardResponseInternal(header, requestFilterResult.message());
-                    }
-                    else {
-                        var header = requestFilterResult.header() == null ? decodedFrame.header() : requestFilterResult.header();
-                        filterContext.forwardRequestInternal((RequestHeaderData) header, requestFilterResult.message());
-                    }
-                }
-
-                if (requestFilterResult.closeConnection()) {
-                    filterContext.closeConnection();
-                }
-
-            });
+            invoker.onRequest(decodedFrame.apiKey(), decodedFrame.apiVersion(), decodedFrame.header(), decodedFrame.body(), filterContext);
 
         }
         else {
@@ -132,38 +89,7 @@ public class FilterHandler extends ChannelDuplexHandler {
                     LOGGER.debug("{}: Dispatching upstream {} response to filter {}: {}",
                             ctx.channel(), decodedFrame.apiKey(), filterDescriptor(), msg);
                 }
-                var stage = invoker.onResponse(decodedFrame.apiKey(), decodedFrame.apiVersion(),
-                        decodedFrame.header(), decodedFrame.body(), filterContext);
-                if (stage == null) {
-                    LOGGER.error("{}: Filter{} for {} response unexpectedly returned null. This is a coding error in the filter. Closing connection.",
-                            ctx.channel(), filterDescriptor(), decodedFrame.apiKey());
-                    filterContext.closeConnection();
-                    return;
-                }
-
-                stage.whenComplete((responseFilterResult, t) -> {
-                    if (t != null) {
-                        LOGGER.warn("{}: Filter{} for {} response ended exceptionally - closing connection",
-                                ctx.channel(), filterDescriptor(), decodedFrame.apiKey(), t);
-                        filterContext.closeConnection();
-                        return;
-                    }
-                    if (responseFilterResult.drop()) {
-                        if (LOGGER.isDebugEnabled()) {
-                            LOGGER.debug("{}: Filter{} drops {} response",
-                                    ctx.channel(), filterDescriptor(), decodedFrame.apiKey());
-                        }
-                    }
-                    else if (responseFilterResult.message() != null) {
-                        ResponseHeaderData header = responseFilterResult.header() == null ? decodedFrame.header() : (ResponseHeaderData) responseFilterResult.header();
-                        filterContext.forwardResponseInternal(header, responseFilterResult.message());
-                    }
-                    if (responseFilterResult.closeConnection()) {
-                        filterContext.closeConnection();
-                    }
-
-                });
-
+                invoker.onResponse(decodedFrame.apiKey(), decodedFrame.apiVersion(), decodedFrame.header(), decodedFrame.body(), filterContext);
             }
         }
         else {
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ApiVersionsFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ApiVersionsFilter.java
index 8b2ab21d..ca4957be 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ApiVersionsFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ApiVersionsFilter.java
@@ -7,7 +7,6 @@ package io.kroxylicious.proxy.internal.filter;
 
 import java.util.HashSet;
 import java.util.Set;
-import java.util.concurrent.CompletionStage;
 
 import org.apache.kafka.common.message.ApiVersionsResponseData;
 import org.apache.kafka.common.message.ResponseHeaderData;
@@ -17,7 +16,6 @@ import org.slf4j.LoggerFactory;
 
 import io.kroxylicious.proxy.filter.ApiVersionsResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 
 /**
  * Changes an API_VERSIONS response so that a client sees the intersection of supported version ranges for each
@@ -74,9 +72,8 @@ public class ApiVersionsFilter implements ApiVersionsResponseFilter {
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData data,
-                                                                       KrpcFilterContext context) {
+    public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData data, KrpcFilterContext context) {
         intersectApiVersions(context.channelDescriptor(), data);
-        return context.forwardResponse(header, data);
+        context.forwardResponse(header, data);
     }
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java
index 178948a1..a5e850f8 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java
@@ -7,7 +7,6 @@ package io.kroxylicious.proxy.internal.filter;
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.CompletionStage;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
 import java.util.function.ObjIntConsumer;
@@ -28,7 +27,6 @@ import io.kroxylicious.proxy.filter.DescribeClusterResponseFilter;
 import io.kroxylicious.proxy.filter.FindCoordinatorResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.MetadataResponseFilter;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 import io.kroxylicious.proxy.internal.net.EndpointReconciler;
 import io.kroxylicious.proxy.model.VirtualCluster;
 import io.kroxylicious.proxy.service.HostPort;
@@ -50,31 +48,29 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onMetadataResponse(short apiVersion, ResponseHeaderData header, MetadataResponseData data, KrpcFilterContext context) {
+    public void onMetadataResponse(short apiVersion, ResponseHeaderData header, MetadataResponseData data, KrpcFilterContext context) {
         var nodeMap = new HashMap<Integer, HostPort>();
         for (MetadataResponseBroker broker : data.brokers()) {
             nodeMap.put(broker.nodeId(), new HostPort(broker.host(), broker.port()));
             apply(context, broker, MetadataResponseBroker::nodeId, MetadataResponseBroker::host, MetadataResponseBroker::port, MetadataResponseBroker::setHost,
                     MetadataResponseBroker::setPort);
         }
-        return doReconcileThenForwardResponse(header, data, context, nodeMap);
+        doReconcileThenForwardResponse(header, data, context, nodeMap);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onDescribeClusterResponse(short apiVersion, ResponseHeaderData header, DescribeClusterResponseData data,
-                                                                           KrpcFilterContext context) {
+    public void onDescribeClusterResponse(short apiVersion, ResponseHeaderData header, DescribeClusterResponseData data, KrpcFilterContext context) {
         var nodeMap = new HashMap<Integer, HostPort>();
         for (DescribeClusterBroker broker : data.brokers()) {
             nodeMap.put(broker.brokerId(), new HostPort(broker.host(), broker.port()));
             apply(context, broker, DescribeClusterBroker::brokerId, DescribeClusterBroker::host, DescribeClusterBroker::port, DescribeClusterBroker::setHost,
                     DescribeClusterBroker::setPort);
         }
-        return doReconcileThenForwardResponse(header, data, context, nodeMap);
+        doReconcileThenForwardResponse(header, data, context, nodeMap);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onFindCoordinatorResponse(short apiVersion, ResponseHeaderData header, FindCoordinatorResponseData data,
-                                                                           KrpcFilterContext context) {
+    public void onFindCoordinatorResponse(short apiVersion, ResponseHeaderData header, FindCoordinatorResponseData data, KrpcFilterContext context) {
         // Version 4+
         for (Coordinator coordinator : data.coordinators()) {
             // If the coordinator is not yet available, the server returns a nodeId of -1.
@@ -87,7 +83,7 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
             apply(context, data, FindCoordinatorResponseData::nodeId, FindCoordinatorResponseData::host, FindCoordinatorResponseData::port,
                     FindCoordinatorResponseData::setHost, FindCoordinatorResponseData::setPort);
         }
-        return context.responseFilterResultBuilder().forward(header, data).completed();
+        context.forwardResponse(header, data);
     }
 
     private <T> void apply(KrpcFilterContext context, T broker, Function<T, Integer> nodeIdGetter, Function<T, String> hostGetter, ToIntFunction<T> portGetter,
@@ -103,12 +99,11 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
         portSetter.accept(broker, downstreamAddress.port());
     }
 
-    private CompletionStage<ResponseFilterResult> doReconcileThenForwardResponse(ResponseHeaderData header, ApiMessage data, KrpcFilterContext context,
-                                                                                 Map<Integer, HostPort> nodeMap) {
+    private void doReconcileThenForwardResponse(ResponseHeaderData header, ApiMessage data, KrpcFilterContext context, Map<Integer, HostPort> nodeMap) {
         // https://github.com/kroxylicious/kroxylicious/issues/351
         // Using synchronous approach for now
         reconciler.reconcile(virtualCluster, nodeMap).toCompletableFuture().join();
-        LOGGER.debug("Endpoint reconciliation complete for virtual cluster {}", virtualCluster);
-        return context.responseFilterResultBuilder().forward(header, data).completed();
+        context.forwardResponse(header, data);
+        LOGGER.debug("Endpoint reconciliation complete for  virtual cluster {}", virtualCluster);
     }
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearner.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearner.java
index 39182460..2de82316 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearner.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearner.java
@@ -6,8 +6,6 @@
 package io.kroxylicious.proxy.internal.filter;
 
 import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
 
 import org.apache.kafka.common.message.MetadataRequestData;
 import org.apache.kafka.common.message.MetadataResponseData;
@@ -19,7 +17,6 @@ import org.slf4j.LoggerFactory;
 
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.RequestFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
 
 /**
  * An internal filter that causes the system to eagerly learn the cluster's topology by spontaneously emitting
@@ -45,9 +42,9 @@ public class EagerMetadataLearner implements RequestFilter {
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         if (KAFKA_PRELUDE.contains(apiKey)) {
-            return filterContext.requestFilterResultBuilder().forward(header, body).completed();
+            filterContext.forwardRequest(header, body);
         }
         else {
             final short apiVersion = determineMetadataApiVersion(header);
@@ -56,25 +53,19 @@ public class EagerMetadataLearner implements RequestFilter {
             boolean useClientRequest = apiKey.equals(ApiKeys.METADATA) && apiVersion == header.requestApiVersion();
             var request = useClientRequest ? (MetadataRequestData) body : new MetadataRequestData();
 
-            var future = new CompletableFuture<RequestFilterResult>();
             var unused = filterContext.<MetadataResponseData> sendRequest(apiVersion, request)
                     .thenAccept(metadataResponseData -> {
-                        // closing the connection is important. This client connection is connected to bootstrap (it could
-                        // be any broker or maybe not something else). we must close the connection to force the client to
-                        // connect again.
-                        var builder = filterContext.requestFilterResultBuilder();
                         if (useClientRequest) {
-                            // The client's request matched our out-of-band message, so we may as well return the
+                            // The client's requested matched our out-of-band request, so we may as well return the
                             // response.
-                            future.complete(builder.shortCircuitResponse(metadataResponseData).withCloseConnection().build());
-                        }
-                        else {
-                            future.complete(builder.withCloseConnection().build());
-
+                            filterContext.forwardResponse(metadataResponseData);
                         }
+                        // closing the connection is important. This client connection is connected to bootstrap (it could
+                        // be any broker or maybe not something else). we must close the connection to force the client to
+                        // connect again.
+                        filterContext.closeConnection();
                         LOGGER.info("Closing upstream bootstrap connection {} now that endpoint reconciliation is complete.", filterContext.channelDescriptor());
                     });
-            return future;
         }
     }
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java
index 18123208..2788984e 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java
@@ -9,8 +9,6 @@ import java.lang.reflect.InvocationTargetException;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
 import java.util.stream.Collectors;
 
 import org.apache.kafka.common.Uuid;
@@ -33,7 +31,6 @@ import org.slf4j.LoggerFactory;
 import io.kroxylicious.proxy.config.BaseConfig;
 import io.kroxylicious.proxy.filter.FetchResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 import io.kroxylicious.proxy.internal.util.MemoryRecordsHelper;
 
 /**
@@ -74,8 +71,7 @@ public class FetchResponseTransformationFilter implements FetchResponseFilter {
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData fetchResponse,
-                                                                 KrpcFilterContext context) {
+    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData fetchResponse, KrpcFilterContext context) {
         List<MetadataRequestData.MetadataRequestTopic> requestTopics = fetchResponse.responses().stream()
                 .filter(t -> t.topic().isEmpty())
                 .map(fetchableTopicResponse -> {
@@ -86,29 +82,24 @@ public class FetchResponseTransformationFilter implements FetchResponseFilter {
                 .collect(Collectors.toList());
         if (!requestTopics.isEmpty()) {
             LOGGER.debug("Fetch response contains {} unknown topic ids, lookup via Metadata request: {}", requestTopics.size(), requestTopics);
-            var future = new CompletableFuture<ResponseFilterResult>();
             // TODO Can't necessarily use HIGHEST_SUPPORTED_VERSION, must use highest supported version
             context.<MetadataResponseData> sendRequest(MetadataRequestData.HIGHEST_SUPPORTED_VERSION,
                     new MetadataRequestData()
                             .setTopics(requestTopics))
                     .thenAccept(metadataResponse -> {
-                        Map<Uuid, String> uidToName = metadataResponse.topics().stream().collect(Collectors.toMap(MetadataResponseData.MetadataResponseTopic::topicId,
-                                MetadataResponseData.MetadataResponseTopic::name));
+                        Map<Uuid, String> uidToName = metadataResponse.topics().stream().collect(Collectors.toMap(ti -> ti.topicId(), ti -> ti.name()));
                         LOGGER.debug("Metadata response yields {}, updating original Fetch response", uidToName);
                         for (var fetchableTopicResponse : fetchResponse.responses()) {
                             fetchableTopicResponse.setTopic(uidToName.get(fetchableTopicResponse.topicId()));
                         }
                         applyTransformation(context, fetchResponse);
                         LOGGER.debug("Forwarding original Fetch response");
-
-                        var value = context.responseFilterResultBuilder().forward(header, fetchResponse).build();
-                        future.complete(value);
+                        context.forwardResponse(header, fetchResponse);
                     });
-            return future;
         }
         else {
             applyTransformation(context, fetchResponse);
-            return context.forwardResponse(header, fetchResponse);
+            context.forwardResponse(header, fetchResponse);
         }
     }
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FilterResultBuilderImpl.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FilterResultBuilderImpl.java
deleted file mode 100644
index ae944727..00000000
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FilterResultBuilderImpl.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * 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.filter;
-
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
-
-import org.apache.kafka.common.protocol.ApiMessage;
-
-import io.micrometer.common.lang.NonNull;
-
-import io.kroxylicious.proxy.filter.FilterResult;
-import io.kroxylicious.proxy.filter.FilterResultBuilder;
-import io.kroxylicious.proxy.filter.filterresultbuilder.CloseOrTerminalStage;
-import io.kroxylicious.proxy.filter.filterresultbuilder.TerminalStage;
-
-public abstract class FilterResultBuilderImpl<H extends ApiMessage, FR extends FilterResult>
-        implements FilterResultBuilder<H, FR>, CloseOrTerminalStage<FR> {
-    private ApiMessage message;
-    private ApiMessage header;
-    private boolean closeConnection;
-    private boolean drop;
-
-    protected FilterResultBuilderImpl() {
-    }
-
-    @Override
-    public CloseOrTerminalStage<FR> forward(@NonNull H header, @NonNull ApiMessage message) {
-        validateForward(header, message);
-        this.header = header;
-        this.message = message;
-
-        return this;
-    }
-
-    protected void validateForward(H header, ApiMessage message) {
-        if (header == null) {
-            throw new IllegalArgumentException("header may not be null");
-        }
-        if (message == null) {
-            throw new IllegalArgumentException("message may not be null");
-        }
-    }
-
-    ApiMessage header() {
-        return header;
-    }
-
-    ApiMessage message() {
-        return message;
-    }
-
-    @Override
-    public TerminalStage<FR> withCloseConnection() {
-        this.closeConnection = true;
-        return this;
-    }
-
-    boolean closeConnection() {
-        return closeConnection;
-    }
-
-    @Override
-    public TerminalStage<FR> drop() {
-        this.drop = true;
-        return this;
-    }
-
-    public boolean isDrop() {
-        return drop;
-    }
-
-    @Override
-    public CompletionStage<FR> completed() {
-        return CompletableFuture.completedStage(build());
-    }
-}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ProduceRequestTransformationFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ProduceRequestTransformationFilter.java
index 8015a655..6309aeeb 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ProduceRequestTransformationFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ProduceRequestTransformationFilter.java
@@ -9,7 +9,6 @@ import java.lang.reflect.InvocationTargetException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.Iterator;
-import java.util.concurrent.CompletionStage;
 
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceRequestData.PartitionProduceData;
@@ -27,7 +26,6 @@ import com.fasterxml.jackson.annotation.JsonProperty;
 import io.kroxylicious.proxy.config.BaseConfig;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
 import io.kroxylicious.proxy.internal.util.MemoryRecordsHelper;
 
 /**
@@ -74,9 +72,9 @@ public class ProduceRequestTransformationFilter implements ProduceRequestFilter
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData data, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData data, KrpcFilterContext context) {
         applyTransformation(context, data);
-        return context.forwardRequest(header, data);
+        context.forwardRequest(header, data);
     }
 
     private void applyTransformation(KrpcFilterContext ctx, ProduceRequestData req) {
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/RequestFilterResultBuilderImpl.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/RequestFilterResultBuilderImpl.java
deleted file mode 100644
index 044c8dca..00000000
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/RequestFilterResultBuilderImpl.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * 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.filter;
-
-import org.apache.kafka.common.message.RequestHeaderData;
-import org.apache.kafka.common.message.ResponseHeaderData;
-import org.apache.kafka.common.protocol.ApiMessage;
-
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.RequestFilterResultBuilder;
-import io.kroxylicious.proxy.filter.filterresultbuilder.CloseOrTerminalStage;
-
-public class RequestFilterResultBuilderImpl extends FilterResultBuilderImpl<RequestHeaderData, RequestFilterResult>
-        implements RequestFilterResultBuilder {
-
-    private static final String REQUEST_DATA_NAME_SUFFIX = "RequestData";
-    private static final String RESPONSE_DATA_NAME_SUFFIX = "ResponseData";
-    private ResponseHeaderData shortCircuitHeader;
-    private ApiMessage shortCircuitResponse;
-
-    public RequestFilterResultBuilderImpl() {
-    }
-
-    @Override
-    protected void validateForward(RequestHeaderData header, ApiMessage message) {
-        super.validateForward(header, message);
-        if (message != null && !message.getClass().getSimpleName().endsWith(REQUEST_DATA_NAME_SUFFIX)) {
-            throw new IllegalArgumentException("class name " + message.getClass().getName() + " does not have expected suffix " + REQUEST_DATA_NAME_SUFFIX);
-        }
-    }
-
-    @Override
-    public CloseOrTerminalStage<RequestFilterResult> shortCircuitResponse(ResponseHeaderData header, ApiMessage message) {
-        validateShortCircuitResponse(header, message);
-        this.shortCircuitHeader = header;
-        this.shortCircuitResponse = message;
-        return this;
-    }
-
-    @Override
-    public CloseOrTerminalStage<RequestFilterResult> shortCircuitResponse(ApiMessage message) {
-        validateShortCircuitResponse(null, message);
-        this.shortCircuitResponse = message;
-        return this;
-    }
-
-    private void validateShortCircuitResponse(ResponseHeaderData header, ApiMessage message) {
-        if (message == null) {
-            throw new IllegalArgumentException("message may not be null");
-        }
-        if (!message.getClass().getSimpleName().endsWith(RESPONSE_DATA_NAME_SUFFIX)) {
-            throw new IllegalArgumentException("class name " + message.getClass().getName() + " does not have expected suffix " + RESPONSE_DATA_NAME_SUFFIX);
-        }
-    }
-
-    @Override
-    public RequestFilterResult build() {
-
-        return new RequestFilterResult() {
-
-            @Override
-            public boolean shortCircuitResponse() {
-                return shortCircuitResponse != null;
-            }
-
-            @Override
-            public ApiMessage header() {
-
-                return shortCircuitResponse == null ? RequestFilterResultBuilderImpl.this.header() : shortCircuitHeader;
-            }
-
-            @Override
-            public ApiMessage message() {
-                return shortCircuitResponse == null ? RequestFilterResultBuilderImpl.this.message() : shortCircuitResponse;
-            }
-
-            @Override
-            public boolean closeConnection() {
-                return RequestFilterResultBuilderImpl.this.closeConnection();
-            }
-
-            @Override
-            public boolean drop() {
-                return RequestFilterResultBuilderImpl.this.isDrop();
-            }
-        };
-
-    }
-}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ResponseFilterResultBuilderImpl.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ResponseFilterResultBuilderImpl.java
deleted file mode 100644
index 57112c47..00000000
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ResponseFilterResultBuilderImpl.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.filter;
-
-import org.apache.kafka.common.message.ResponseHeaderData;
-import org.apache.kafka.common.protocol.ApiMessage;
-
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResultBuilder;
-
-public class ResponseFilterResultBuilderImpl extends FilterResultBuilderImpl<ResponseHeaderData, ResponseFilterResult>
-        implements ResponseFilterResultBuilder {
-
-    private static final String RESPONSE_DATA_NAME_SUFFIX = "ResponseData";
-
-    @Override
-    protected void validateForward(ResponseHeaderData header, ApiMessage message) {
-        super.validateForward(header, message);
-        if (message != null && !message.getClass().getSimpleName().endsWith(RESPONSE_DATA_NAME_SUFFIX)) {
-            throw new IllegalArgumentException("class name " + message.getClass().getName() + " does not have expected suffix " + RESPONSE_DATA_NAME_SUFFIX);
-        }
-    }
-
-    @Override
-    public ResponseFilterResult build() {
-        return new ResponseFilterResult() {
-            @Override
-            public ApiMessage header() {
-                return ResponseFilterResultBuilderImpl.this.header();
-            }
-
-            @Override
-            public ApiMessage message() {
-                return ResponseFilterResultBuilderImpl.this.message();
-            }
-
-            @Override
-            public boolean closeConnection() {
-                return ResponseFilterResultBuilderImpl.this.closeConnection();
-            }
-
-            @Override
-            public boolean drop() {
-                return ResponseFilterResultBuilderImpl.this.isDrop();
-            }
-        };
-    }
-
-}
diff --git a/pom.xml b/pom.xml
index 561846f9..c5bb95cc 100644
--- a/pom.xml
+++ b/pom.xml
@@ -458,6 +458,11 @@
                     <artifactId>build-helper-maven-plugin</artifactId>
                     <version>3.4.0</version>
                 </plugin>
+                <plugin>
+                    <groupId>org.jacoco</groupId>
+                    <artifactId>jacoco-maven-plugin</artifactId>
+                    <version>${jacoco-maven-plugin.version}</version>
+                </plugin>
             </plugins>
         </pluginManagement>
         <plugins>
@@ -715,7 +720,6 @@
                     <plugin>
                         <groupId>org.jacoco</groupId>
                         <artifactId>jacoco-maven-plugin</artifactId>
-                        <version>${jacoco-maven-plugin.version}</version>
                         <executions>
                             <execution>
                                 <id>prepare-agent</id>
@@ -736,20 +740,6 @@
                             </execution>
                         </executions>
                     </plugin>
-                    <plugin>
-                        <groupId>org.sonarsource.scanner.maven</groupId>
-                        <artifactId>sonar-maven-plugin</artifactId>
-                        <version>3.9.1.2184</version>
-                        <executions>
-                            <execution>
-                                <id>sonar-scan</id>
-                                <goals>
-                                    <goal>sonar</goal>
-                                </goals>
-                                <phase>verify</phase>
-                            </execution>
-                        </executions>
-                    </plugin>
                 </plugins>
             </build>
         </profile>


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
index 7178dfa5..441525fd 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
@@ -242,6 +242,7 @@ public class KrpcFilterIT {
 
     @Test
     public void shouldModifyFetchMessage(KafkaCluster cluster, Admin admin) throws Exception {
+        var proxyAddress = HostPort.parse("localhost:9192");
 
         admin.createTopics(List.of(
                 new NewTopic(TOPIC_1, 1, (short) 1),
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CompositePrefixingFixedClientIdFilter.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CompositePrefixingFixedClientIdFilter.java
index 20fb6530..c8261875 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CompositePrefixingFixedClientIdFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CompositePrefixingFixedClientIdFilter.java
@@ -7,7 +7,6 @@
 package io.kroxylicious.proxy.filter;
 
 import java.util.List;
-import java.util.concurrent.CompletionStage;
 
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -28,9 +27,9 @@ public class CompositePrefixingFixedClientIdFilter implements CompositeFilter {
 
     private class PrefixingFilter implements RequestFilter {
         @Override
-        public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
             header.setClientId(config.prefix + header.clientId());
-            return filterContext.forwardRequest(header, body);
+            filterContext.forwardRequest(header, body);
         }
     }
 
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CreateTopicRejectFilter.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CreateTopicRejectFilter.java
index 4b040a50..8a2283cc 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CreateTopicRejectFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CreateTopicRejectFilter.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.CreateTopicsRequestData;
 import org.apache.kafka.common.message.CreateTopicsResponseData;
 import org.apache.kafka.common.message.RequestHeaderData;
@@ -18,8 +16,7 @@ public class CreateTopicRejectFilter implements CreateTopicsRequestFilter {
     public static final String ERROR_MESSAGE = "rejecting all topics";
 
     @Override
-    public CompletionStage<RequestFilterResult> onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request,
-                                                                      KrpcFilterContext context) {
+    public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
         CreateTopicsResponseData response = new CreateTopicsResponseData();
         CreateTopicsResponseData.CreatableTopicResultCollection topics = new CreateTopicsResponseData.CreatableTopicResultCollection();
         allocateByteBufToTestKroxyliciousReleasesIt(context);
@@ -30,7 +27,7 @@ public class CreateTopicRejectFilter implements CreateTopicsRequestFilter {
             topics.add(result);
         });
         response.setTopics(topics);
-        return context.requestFilterResultBuilder().shortCircuitResponse(response).completed();
+        context.forwardResponse(response);
     }
 
     private static void allocateByteBufToTestKroxyliciousReleasesIt(KrpcFilterContext context) {
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/FixedClientIdFilter.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/FixedClientIdFilter.java
index 7b76df75..d9361025 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/FixedClientIdFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/FixedClientIdFilter.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -37,13 +35,23 @@ public class FixedClientIdFilter implements RequestFilter, ResponseFilter {
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public boolean shouldHandleRequest(ApiKeys apiKey, short apiVersion) {
+        return true;
+    }
+
+    @Override
+    public boolean shouldHandleResponse(ApiKeys apiKey, short apiVersion) {
+        return true;
+    }
+
+    @Override
+    public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         header.setClientId(clientId);
-        return filterContext.forwardRequest(header, body);
+        filterContext.forwardRequest(header, body);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-        return filterContext.forwardResponse(header, body);
+    public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        filterContext.forwardResponse(header, body);
     }
 }
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/OutOfBandSendFilter.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/OutOfBandSendFilter.java
index 2f00b31e..d599ef74 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/OutOfBandSendFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/OutOfBandSendFilter.java
@@ -7,7 +7,6 @@
 package io.kroxylicious.proxy.filter;
 
 import java.util.Objects;
-import java.util.concurrent.CompletionStage;
 import java.util.stream.Collectors;
 
 import org.apache.kafka.common.message.CreateTopicsRequestData;
@@ -56,23 +55,21 @@ public class OutOfBandSendFilter implements DescribeClusterRequestFilter, Descri
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onDescribeClusterRequest(short apiVersion, RequestHeaderData header, DescribeClusterRequestData request,
-                                                                         KrpcFilterContext context) {
+    public void onDescribeClusterRequest(short apiVersion, RequestHeaderData header, DescribeClusterRequestData request, KrpcFilterContext context) {
         ApiKeys apiKeyToSend = config.apiKeyToSend;
         ApiMessage message = createApiMessage(apiKeyToSend);
         context.sendRequest(apiKeyToSend.latestVersion(), message).thenAccept(apiMessage -> {
             // expected to execute before onDescribeClusterResponse becase sendRequest called before forwardRequest
             values = unknownTaggedFieldsToStrings(apiMessage, config.tagIdToCollect).collect(Collectors.joining(","));
         });
-        return context.forwardRequest(header, request);
+        context.forwardRequest(header, request);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onDescribeClusterResponse(short apiVersion, ResponseHeaderData header, DescribeClusterResponseData response,
-                                                                           KrpcFilterContext context) {
+    public void onDescribeClusterResponse(short apiVersion, ResponseHeaderData header, DescribeClusterResponseData response, KrpcFilterContext context) {
         response.setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code())
                 .setErrorMessage("filterNameTaggedFieldsFromOutOfBandResponse: " + values);
-        return context.forwardResponse(header, response);
+        context.forwardResponse(response);
     }
 
     private static ApiMessage createApiMessage(ApiKeys apiKeyToSend) {
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
index 9ed8696b..bffd8947 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
@@ -7,7 +7,6 @@
 package io.kroxylicious.proxy.filter;
 
 import java.util.Set;
-import java.util.concurrent.CompletionStage;
 
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.message.ResponseHeaderData;
@@ -42,19 +41,19 @@ public class RequestResponseMarkingFilter implements RequestFilter, ResponseFilt
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         if (keysToMark.contains(apiKey)) {
             body.unknownTaggedFields().add(createTaggedField("request"));
         }
-        return filterContext.forwardRequest(header, body);
+        filterContext.forwardRequest(header, body);
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         if (keysToMark.contains(apiKey)) {
             body.unknownTaggedFields().add(createTaggedField("response"));
         }
-        return filterContext.forwardResponse(header, body);
+        filterContext.forwardResponse(header, body);
     }
 
     private RawTaggedField createTaggedField(String type) {
diff --git a/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java b/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
index d351cb72..996ec724 100644
--- a/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
+++ b/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
@@ -8,7 +8,6 @@ package io.kroxylicious.proxy.filter.multitenant;
 
 import java.io.IOException;
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
 import java.util.regex.Pattern;
 import java.util.stream.Stream;
 
@@ -17,14 +16,11 @@ import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ApiMessage;
-import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
 
 import com.flipkart.zjsonpatch.JsonDiff;
 import com.google.common.reflect.ClassPath;
@@ -33,8 +29,6 @@ import com.google.common.reflect.ClassPath.ResourceInfo;
 import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.filter.FilterInvoker;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 import io.kroxylicious.test.requestresponsetestdef.ApiMessageTestDef;
 import io.kroxylicious.test.requestresponsetestdef.RequestResponseTestDef;
 
@@ -42,16 +36,15 @@ import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.collect.Iterables.getOnlyElement;
 import static io.kroxylicious.test.requestresponsetestdef.KafkaApiMessageConverter.requestConverterFor;
 import static io.kroxylicious.test.requestresponsetestdef.KafkaApiMessageConverter.responseConverterFor;
-import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.Mock.Strictness.LENIENT;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-@ExtendWith(MockitoExtension.class)
 class MultiTenantTransformationFilterTest {
     private static final Pattern TEST_RESOURCE_FILTER = Pattern.compile(
             String.format("%s/.*\\.yaml", MultiTenantTransformationFilterTest.class.getPackageName().replace(".", "/")));
-    private static final String TENANT_1 = "tenant1.kafka.example.com";
 
     private static List<ResourceInfo> getTestResources() throws IOException {
         var resources = ClassPath.from(MultiTenantTransformationFilterTest.class.getClassLoader()).getResources().stream()
@@ -69,21 +62,14 @@ class MultiTenantTransformationFilterTest {
 
     private final FilterInvoker invoker = getOnlyElement(FilterAndInvoker.build(filter)).invoker();
 
-    @Mock(strictness = LENIENT)
-    private KrpcFilterContext context;
+    private final KrpcFilterContext context = mock(KrpcFilterContext.class);
 
-    @Captor
-    private ArgumentCaptor<RequestHeaderData> requestHeaderDataCaptor;
+    private final ArgumentCaptor<ApiMessage> apiMessageCaptor = ArgumentCaptor.forClass(ApiMessage.class);
 
-    @Mock(strictness = LENIENT)
-    private RequestFilterResult requestFilterResult;
-    @Captor
-    private ArgumentCaptor<ResponseHeaderData> responseHeaderDataArgumentCaptor;
-    @Mock(strictness = LENIENT)
-    private ResponseFilterResult responseFilterResult;
-
-    @Captor
-    private ArgumentCaptor<ApiMessage> apiMessageCaptor;
+    @BeforeEach
+    public void beforeEach() {
+        when(context.sniHostname()).thenReturn("tenant1.kafka.example.com");
+    }
 
     public static Stream<Arguments> requests() throws Exception {
         return RequestResponseTestDef.requestResponseTestDefinitions(getTestResources()).filter(td -> td.request() != null)
@@ -92,24 +78,16 @@ class MultiTenantTransformationFilterTest {
 
     @ParameterizedTest(name = "{0}")
     @MethodSource(value = "requests")
-    void requestsTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef requestTestDef)
-            throws Exception {
+    void requestsTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef requestTestDef) {
         var request = requestTestDef.message();
         // marshalled the request object back to json, this is used for the comparison later.
         var requestWriter = requestConverterFor(apiMessageType).writer();
         var marshalled = requestWriter.apply(request, header.requestApiVersion());
 
-        when(context.sniHostname()).thenReturn(TENANT_1);
-        when(requestFilterResult.message()).thenAnswer(invocation -> apiMessageCaptor.getValue());
-        when(requestFilterResult.header()).thenAnswer(invocation -> requestHeaderDataCaptor.getValue());
-        when(context.forwardRequest(requestHeaderDataCaptor.capture(), apiMessageCaptor.capture())).thenAnswer(
-                invocation -> CompletableFuture.completedStage(requestFilterResult));
+        invoker.onRequest(ApiKeys.forId(apiMessageType.apiKey()), header.requestApiVersion(), header, request, context);
+        verify(context).forwardRequest(any(), apiMessageCaptor.capture());
 
-        var stage = invoker.onRequest(ApiKeys.forId(apiMessageType.apiKey()), header.requestApiVersion(), header, request, context);
-        assertThat(stage).isCompleted();
-        var forward = stage.toCompletableFuture().get().message();
-
-        var filtered = requestWriter.apply(forward, header.requestApiVersion());
+        var filtered = requestWriter.apply(apiMessageCaptor.getValue(), header.requestApiVersion());
         assertEquals(requestTestDef.expectedPatch(), JsonDiff.asJson(marshalled, filtered));
     }
 
@@ -120,26 +98,18 @@ class MultiTenantTransformationFilterTest {
 
     @ParameterizedTest(name = "{0}")
     @MethodSource(value = "responses")
-    void responseTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef responseTestDef)
-            throws Exception {
+    void responseTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef responseTestDef) {
         var response = responseTestDef.message();
         // marshalled the response object back to json, this is used for comparison later.
         var responseWriter = responseConverterFor(apiMessageType).writer();
 
         var marshalled = responseWriter.apply(response, header.requestApiVersion());
 
-        when(context.sniHostname()).thenReturn(TENANT_1);
-        when(responseFilterResult.message()).thenAnswer(invocation -> apiMessageCaptor.getValue());
-        when(responseFilterResult.header()).thenAnswer(invocation -> responseHeaderDataArgumentCaptor.getValue());
-        when(context.forwardResponse(responseHeaderDataArgumentCaptor.capture(), apiMessageCaptor.capture())).thenAnswer(
-                invocation -> CompletableFuture.completedStage(responseFilterResult));
-
         ResponseHeaderData headerData = new ResponseHeaderData();
-        var stage = invoker.onResponse(ApiKeys.forId(apiMessageType.apiKey()), header.requestApiVersion(), headerData, response, context);
-        assertThat(stage).isCompleted();
-        var forward = stage.toCompletableFuture().get().message();
+        invoker.onResponse(ApiKeys.forId(apiMessageType.apiKey()), header.requestApiVersion(), headerData, response, context);
+        verify(context).forwardResponse(any(), apiMessageCaptor.capture());
 
-        var filtered = responseWriter.apply(forward, header.requestApiVersion());
+        var filtered = responseWriter.apply(apiMessageCaptor.getValue(), header.requestApiVersion());
         assertEquals(responseTestDef.expectedPatch(), JsonDiff.asJson(marshalled, filtered));
     }
 }
diff --git a/kroxylicious-api/src/test/java/io/kroxylicious/proxy/filter/FilterInvokersTest.java b/kroxylicious-api/src/test/java/io/kroxylicious/proxy/filter/FilterInvokersTest.java
index 2152785f..e2559c6d 100644
--- a/kroxylicious-api/src/test/java/io/kroxylicious/proxy/filter/FilterInvokersTest.java
+++ b/kroxylicious-api/src/test/java/io/kroxylicious/proxy/filter/FilterInvokersTest.java
@@ -7,7 +7,6 @@
 package io.kroxylicious.proxy.filter;
 
 import java.util.List;
-import java.util.concurrent.CompletionStage;
 import java.util.stream.Stream;
 
 import org.apache.kafka.common.message.ApiVersionsRequestData;
@@ -41,8 +40,12 @@ class FilterInvokersTest {
     public void testCompositeFilter() {
         MultipleSpecificFilter filterA = new MultipleSpecificFilter();
         RequestResponseFilter filterB = new RequestResponseFilter();
-        RequestFilter filterC = (apiKey, header, body, filterContext) -> null;
-        ResponseFilter filterD = (apiKey, header, body, filterContext) -> null;
+        RequestFilter filterC = (apiKey, header, body, filterContext) -> {
+
+        };
+        ResponseFilter filterD = (apiKey, header, body, filterContext) -> {
+
+        };
         List<FilterAndInvoker> from = FilterInvokers.from(
                 (CompositeFilter) () -> List.of((CompositeFilter) () -> List.of(filterA, filterB), (CompositeFilter) () -> List.of(filterC, filterD)));
         List<KrpcFilter> filters = from.stream().map(FilterAndInvoker::filter).toList();
@@ -64,9 +67,15 @@ class FilterInvokersTest {
     }
 
     public static Stream<KrpcFilter> validFilters() {
-        ApiVersionsRequestFilter singleSpecificMessageFilter = (apiVersion, header, request, context) -> null;
-        RequestFilter requestFilter = (apiKey, header, body, filterContext) -> null;
-        ResponseFilter responseFilter = (apiKey, header, body, filterContext) -> null;
+        ApiVersionsRequestFilter singleSpecificMessageFilter = (apiVersion, header, request, context) -> {
+
+        };
+        RequestFilter requestFilter = (apiKey, header, body, filterContext) -> {
+
+        };
+        ResponseFilter responseFilter = (apiKey, header, body, filterContext) -> {
+
+        };
         return Stream.of(singleSpecificMessageFilter,
                 new MultipleSpecificFilter(),
                 requestFilter,
@@ -79,13 +88,13 @@ class FilterInvokersTest {
     static class RequestResponseFilter implements RequestFilter, ResponseFilter {
 
         @Override
-        public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-            return null;
+        public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+
         }
 
         @Override
-        public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-            return null;
+        public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+
         }
     }
 
@@ -108,17 +117,13 @@ class FilterInvokersTest {
     static class MultipleSpecificFilter implements ApiVersionsRequestFilter, ApiVersionsResponseFilter {
 
         @Override
-        public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
-                                                                         KrpcFilterContext context) {
+        public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
 
-            return null;
         }
 
         @Override
-        public CompletionStage<ResponseFilterResult> onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response,
-                                                                           KrpcFilterContext context) {
+        public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
 
-            return null;
         }
     }
 
@@ -129,9 +134,8 @@ class FilterInvokersTest {
         }
 
         @Override
-        public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
 
-            return null;
         }
     }
 
@@ -148,7 +152,9 @@ class FilterInvokersTest {
     static class TooDeeplyNestedCompositeFilter implements CompositeFilter {
         @Override
         public List<KrpcFilter> getFilters() {
-            return List.of((CompositeFilter) () -> List.of((CompositeFilter) () -> List.of((ApiVersionsRequestFilter) (apiVersion, header, request, context) -> null)));
+            return List.of((CompositeFilter) () -> List.of((CompositeFilter) () -> List.of((ApiVersionsRequestFilter) (apiVersion, header, request, context) -> {
+
+            })));
         }
     }
 
@@ -159,19 +165,16 @@ class FilterInvokersTest {
         }
 
         @Override
-        public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
 
-            return null;
         }
     }
 
     static class SpecificAndCompositeFilter implements ApiVersionsRequestFilter, CompositeFilter {
 
         @Override
-        public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
-                                                                         KrpcFilterContext context) {
+        public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
 
-            return null;
         }
 
         @Override
@@ -183,32 +186,26 @@ class FilterInvokersTest {
     static class SpecificAndResponseFilter implements ApiVersionsRequestFilter, ResponseFilter {
 
         @Override
-        public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
-                                                                         KrpcFilterContext context) {
+        public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
 
-            return null;
         }
 
         @Override
-        public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
 
-            return null;
         }
     }
 
     static class SpecificAndRequestFilter implements ApiVersionsRequestFilter, RequestFilter {
 
         @Override
-        public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
 
-            return null;
         }
 
         @Override
-        public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
-                                                                         KrpcFilterContext context) {
+        public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
 
-            return null;
         }
     }
 
diff --git a/kroxylicious-sample/src/test/java/io/kroxylicious/sample/SampleFetchResponseFilterTest.java b/kroxylicious-sample/src/test/java/io/kroxylicious/sample/SampleFetchResponseFilterTest.java
index 611d475a..56e34e04 100644
--- a/kroxylicious-sample/src/test/java/io/kroxylicious/sample/SampleFetchResponseFilterTest.java
+++ b/kroxylicious-sample/src/test/java/io/kroxylicious/sample/SampleFetchResponseFilterTest.java
@@ -10,7 +10,6 @@ import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
 
 import org.apache.kafka.common.message.ApiMessageType;
 import org.apache.kafka.common.message.FetchResponseData;
@@ -25,21 +24,19 @@ import org.apache.kafka.common.record.TimestampType;
 import org.apache.kafka.common.utils.ByteBufferOutputStream;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Captor;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
 import org.mockito.stubbing.Answer;
 
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 import io.kroxylicious.sample.config.SampleFilterConfig;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-@ExtendWith(MockitoExtension.class)
 class SampleFetchResponseFilterTest {
 
     private static final short API_VERSION = ApiMessageType.FETCH.highestSupportedVersion(true); // this is arbitrary for our filter
@@ -48,19 +45,12 @@ class SampleFetchResponseFilterTest {
     private static final String POST_TRANSFORM_VALUE = "this is what the value will be transformed to";
     private static final String CONFIG_FIND_VALUE = "from";
     private static final String CONFIG_REPLACE_VALUE = "to";
-    @Mock
-    private KrpcFilterContext context;
 
-    @Mock(strictness = Mock.Strictness.LENIENT)
-    private ResponseFilterResult responseFilterResult;
-    @Captor
-    private ArgumentCaptor<Integer> bufferInitialCapacity;
+    private KrpcFilterContext context;
 
     @Captor
-    private ArgumentCaptor<ResponseHeaderData> responseHeaderDataCaptor;
+    private ArgumentCaptor<ApiMessage> apiMessageCaptor = ArgumentCaptor.forClass(ApiMessage.class);
 
-    @Captor
-    private ArgumentCaptor<ApiMessage> apiMessageCaptor;
     private SampleFetchResponseFilter filter;
     private ResponseHeaderData headerData;
 
@@ -76,12 +66,11 @@ class SampleFetchResponseFilterTest {
      * Unit Test: Checks that transformation is applied when response data contains configured value.
      */
     @Test
-    public void willTransformFetchResponseTest() throws Exception {
+    public void willTransformFetchResponseTest() {
         var responseData = buildFetchResponseData(PRE_TRANSFORM_VALUE);
-        var stage = filter.onFetchResponse(API_VERSION, headerData, responseData, context);
-
-        var response = stage.toCompletableFuture().get().message();
-        var unpackedResponse = unpackFetchResponseData(((FetchResponseData) response));
+        filter.onFetchResponse(API_VERSION, headerData, responseData, context);
+        verify(context).forwardResponse(any(), apiMessageCaptor.capture());
+        var unpackedResponse = unpackFetchResponseData((FetchResponseData) apiMessageCaptor.getValue());
         // We only put 1 record in, we should only get 1 record back, and
         // We should see that the unpacked response value has changed from the input value, and
         // We should see that the unpacked response value has been transformed to the correct value
@@ -95,24 +84,21 @@ class SampleFetchResponseFilterTest {
      * value.
      */
     @Test
-    public void wontTransformFetchResponseTest() throws Exception {
+    public void wontTransformFetchResponseTest() {
         var responseData = buildFetchResponseData(NO_TRANSFORM_VALUE);
-        var stage = filter.onFetchResponse(API_VERSION, headerData, responseData, context);
-
-        var response = stage.toCompletableFuture().get().message();
-        var unpackedResponse = unpackFetchResponseData((FetchResponseData) response);
+        filter.onFetchResponse(API_VERSION, headerData, responseData, context);
+        verify(context).forwardResponse(any(), apiMessageCaptor.capture());
+        var unpackedResponse = unpackFetchResponseData((FetchResponseData) apiMessageCaptor.getValue());
         // We only put 1 record in, we should only get 1 record back, and
         // We should see that the unpacked response value has not changed from the input value
         assertThat(unpackedResponse).containsExactly(NO_TRANSFORM_VALUE);
     }
 
     private void setupContextMock() {
-        when(context.forwardResponse(responseHeaderDataCaptor.capture(), apiMessageCaptor.capture())).thenAnswer(
-                invocation -> CompletableFuture.completedStage(responseFilterResult));
-        when(responseFilterResult.message()).thenAnswer(invocation -> apiMessageCaptor.getValue());
-        when(responseFilterResult.header()).thenAnswer(invocation -> responseHeaderDataCaptor.getValue());
-
-        when(context.createByteBufferOutputStream(bufferInitialCapacity.capture())).thenAnswer(
+        context = mock(KrpcFilterContext.class);
+        // create stub for createByteBufferOutputStream method
+        ArgumentCaptor<Integer> argument = ArgumentCaptor.forClass(Integer.class);
+        when(context.createByteBufferOutputStream(argument.capture())).thenAnswer(
                 (Answer<ByteBufferOutputStream>) invocation -> {
                     Object[] args = invocation.getArguments();
                     Integer size = (Integer) args[0];
diff --git a/kroxylicious-sample/src/test/java/io/kroxylicious/sample/SampleProduceRequestFilterTest.java b/kroxylicious-sample/src/test/java/io/kroxylicious/sample/SampleProduceRequestFilterTest.java
index 4e538cef..2ce3840e 100644
--- a/kroxylicious-sample/src/test/java/io/kroxylicious/sample/SampleProduceRequestFilterTest.java
+++ b/kroxylicious-sample/src/test/java/io/kroxylicious/sample/SampleProduceRequestFilterTest.java
@@ -10,7 +10,6 @@ import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
 
 import org.apache.kafka.common.message.ApiMessageType;
 import org.apache.kafka.common.message.ProduceRequestData;
@@ -25,21 +24,19 @@ import org.apache.kafka.common.record.TimestampType;
 import org.apache.kafka.common.utils.ByteBufferOutputStream;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Captor;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
 import org.mockito.stubbing.Answer;
 
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
 import io.kroxylicious.sample.config.SampleFilterConfig;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-@ExtendWith(MockitoExtension.class)
 class SampleProduceRequestFilterTest {
 
     private static final short API_VERSION = ApiMessageType.PRODUCE.highestSupportedVersion(true); // this is arbitrary for our filter
@@ -49,18 +46,10 @@ class SampleProduceRequestFilterTest {
     private static final String CONFIG_FIND_VALUE = "from";
     private static final String CONFIG_REPLACE_VALUE = "to";
 
-    @Mock
     private KrpcFilterContext context;
 
-    @Mock(strictness = Mock.Strictness.LENIENT)
-    private RequestFilterResult requestFilterResult;
     @Captor
-    private ArgumentCaptor<Integer> bufferInitialCapacity;
-
-    @Captor
-    private ArgumentCaptor<ApiMessage> apiMessageCaptor;
-    @Captor
-    private ArgumentCaptor<RequestHeaderData> requestHeaderDataCaptor;
+    private ArgumentCaptor<ApiMessage> apiMessageCaptor = ArgumentCaptor.forClass(ApiMessage.class);
 
     private SampleProduceRequestFilter filter;
     private RequestHeaderData headerData;
@@ -77,13 +66,11 @@ class SampleProduceRequestFilterTest {
      * Unit Test: Checks that transformation is applied when request data contains configured value.
      */
     @Test
-    public void willTransformProduceRequestTest() throws Exception {
+    public void willTransformProduceRequestTest() {
         var requestData = buildProduceRequestData(PRE_TRANSFORM_VALUE);
-
-        var stage = filter.onProduceRequest(API_VERSION, headerData, requestData, context);
-        assertThat(stage).isCompleted();
-        var forwardedRequest = stage.toCompletableFuture().get().message();
-        var unpackedRequest = unpackProduceRequestData((ProduceRequestData) forwardedRequest);
+        filter.onProduceRequest(API_VERSION, headerData, requestData, context);
+        verify(context).forwardRequest(any(), apiMessageCaptor.capture());
+        var unpackedRequest = unpackProduceRequestData((ProduceRequestData) apiMessageCaptor.getValue());
         // We should see that the unpacked request value has changed from the input value, and
         // We should see that the unpacked request value has been transformed to the correct value
         assertThat(unpackedRequest)
@@ -96,23 +83,21 @@ class SampleProduceRequestFilterTest {
      * value.
      */
     @Test
-    public void wontTransformProduceRequestTest() throws Exception {
+    public void wontTransformProduceRequestTest() {
         var requestData = buildProduceRequestData(NO_TRANSFORM_VALUE);
-        var stage = filter.onProduceRequest(API_VERSION, headerData, requestData, context);
-        assertThat(stage).isCompleted();
-        var forwardedRequest = stage.toCompletableFuture().get().message();
-        var unpackedRequest = unpackProduceRequestData((ProduceRequestData) forwardedRequest);
+        filter.onProduceRequest(API_VERSION, headerData, requestData, context);
+        verify(context).forwardRequest(any(), apiMessageCaptor.capture());
+        var unpackedRequest = unpackProduceRequestData((ProduceRequestData) apiMessageCaptor.getValue());
         // We should see that the unpacked request value has not changed from the input value
         assertThat(unpackedRequest)
                 .containsExactly(NO_TRANSFORM_VALUE);
     }
 
     private void setupContextMock() {
-        when(context.forwardRequest(requestHeaderDataCaptor.capture(), apiMessageCaptor.capture())).thenAnswer(invocation -> CompletableFuture.completedStage(requestFilterResult));
-        when(requestFilterResult.message()).thenAnswer(invocation -> apiMessageCaptor.getValue());
-        when(requestFilterResult.header()).thenAnswer(invocation -> requestHeaderDataCaptor.getValue());
-
-        when(context.createByteBufferOutputStream(bufferInitialCapacity.capture())).thenAnswer(
+        context = mock(KrpcFilterContext.class);
+        // create stub for createByteBufferOutputStream method
+        ArgumentCaptor<Integer> argument = ArgumentCaptor.forClass(Integer.class);
+        when(context.createByteBufferOutputStream(argument.capture())).thenAnswer(
                 (Answer<ByteBufferOutputStream>) invocation -> {
                     Object[] args = invocation.getArguments();
                     Integer size = (Integer) args[0];
diff --git a/kroxylicious-unit-test/pom.xml b/kroxylicious-unit-test/pom.xml
index 32de5067..4d961e74 100644
--- a/kroxylicious-unit-test/pom.xml
+++ b/kroxylicious-unit-test/pom.xml
@@ -23,11 +23,6 @@
     <description>Classes to help unit test Filter implementations</description>
 
     <dependencies>
-        <dependency>
-            <groupId>io.kroxylicious</groupId>
-            <artifactId>kroxylicious-api</artifactId>
-        </dependency>
-
         <!-- third party dependencies - runtime and compile -->
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
diff --git a/kroxylicious-unit-test/src/main/java/io/kroxylicious/test/filter/TestHarnessResponseFilterResult.java b/kroxylicious-unit-test/src/main/java/io/kroxylicious/test/filter/TestHarnessResponseFilterResult.java
deleted file mode 100644
index 3d57d3d3..00000000
--- a/kroxylicious-unit-test/src/main/java/io/kroxylicious/test/filter/TestHarnessResponseFilterResult.java
+++ /dev/null
@@ -1,15 +0,0 @@
-/*
- * 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.filter;
-
-import org.apache.kafka.common.message.ResponseHeaderData;
-import org.apache.kafka.common.protocol.ApiMessage;
-
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
-
-public record TestHarnessResponseFilterResult(ResponseHeaderData header, ApiMessage message, boolean closeConnection, boolean drop) implements ResponseFilterResult {
-}
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 f761b96c..f6ccb617 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
@@ -24,8 +24,6 @@ import org.junit.jupiter.api.Test;
 import io.kroxylicious.proxy.filter.ApiVersionsRequestFilter;
 import io.kroxylicious.proxy.filter.ApiVersionsResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 import io.kroxylicious.proxy.frame.DecodedRequestFrame;
 import io.kroxylicious.proxy.frame.DecodedResponseFrame;
 import io.kroxylicious.proxy.future.InternalCompletionStage;
@@ -44,8 +42,7 @@ public class FilterHandlerTest extends FilterHarness {
 
     @Test
     public void testForwardRequest() {
-        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> context.requestFilterResultBuilder().forward(header, request)
-                .completed();
+        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> context.forwardRequest(header, request);
         buildChannel(filter);
         var frame = writeRequest(new ApiVersionsRequestData());
         var propagated = channel.readOutbound();
@@ -61,10 +58,8 @@ public class FilterHandlerTest extends FilterHarness {
             }
 
             @Override
-            public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
-                                                                             KrpcFilterContext context) {
+            public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
                 fail("Should not be called");
-                return null;
             }
         };
         buildChannel(filter);
@@ -77,19 +72,14 @@ public class FilterHandlerTest extends FilterHarness {
     public void testDropRequest() {
         ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> {
             /* don't call forwardRequest => drop the request */
-            return context.requestFilterResultBuilder().drop().completed();
         };
         buildChannel(filter);
         var frame = writeRequest(new ApiVersionsRequestData());
-        var propagated = channel.readOutbound();
-        assertNull(propagated);
-
     }
 
     @Test
     public void testForwardResponse() {
-        ApiVersionsResponseFilter filter = (apiVersion, header, response, context) -> context.responseFilterResultBuilder().forward(header, response)
-                .completed();
+        ApiVersionsResponseFilter filter = (apiVersion, header, response, context) -> context.forwardResponse(header, response);
         buildChannel(filter);
         var frame = writeResponse(new ApiVersionsResponseData());
         var propagated = channel.readInbound();
@@ -133,10 +123,8 @@ public class FilterHandlerTest extends FilterHarness {
             }
 
             @Override
-            public CompletionStage<ResponseFilterResult> onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response,
-                                                                               KrpcFilterContext context) {
+            public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
                 fail("Should not be called");
-                return null;
             }
         };
         buildChannel(filter);
@@ -148,13 +136,10 @@ public class FilterHandlerTest extends FilterHarness {
     @Test
     public void testDropResponse() {
         ApiVersionsResponseFilter filter = (apiVersion, header, response, context) -> {
-            return context.responseFilterResultBuilder().drop().completed();
+            /* don't call forwardRequest => drop the request */
         };
         buildChannel(filter);
         var frame = writeResponse(new ApiVersionsResponseData());
-        var propagated = channel.readInbound();
-        assertNull(propagated);
-
     }
 
     @Test
@@ -165,7 +150,6 @@ public class FilterHandlerTest extends FilterHarness {
             assertNull(fut[0],
                     "Expected to only be called once");
             fut[0] = (InternalCompletionStage<ApiMessage>) context.sendRequest((short) 3, body);
-            return CompletableFuture.completedStage(null);
         };
 
         buildChannel(filter);
@@ -211,7 +195,6 @@ public class FilterHandlerTest extends FilterHarness {
             assertNull(fut[0],
                     "Expected to only be called once");
             fut[0] = context.sendRequest((short) 3, body);
-            return CompletableFuture.completedStage(null);
         };
 
         buildChannel(filter);
@@ -240,7 +223,6 @@ public class FilterHandlerTest extends FilterHarness {
             assertNull(fut[0],
                     "Expected to only be called once");
             fut[0] = context.sendRequest((short) 3, body);
-            return CompletableFuture.completedStage(null);
         };
 
         buildChannel(filter);
@@ -270,7 +252,6 @@ public class FilterHandlerTest extends FilterHarness {
             assertNull(fut[0],
                     "Expected to only be called once");
             fut[0] = context.sendRequest((short) 3, body);
-            return CompletableFuture.completedStage(null);
         };
 
         buildChannel(filter, 50L);
@@ -286,7 +267,7 @@ public class FilterHandlerTest extends FilterHarness {
         assertFalse(q.isDone(),
                 "Future should not be finished yet");
 
-        // timeout the message
+        // timeout the request
         Thread.sleep(60L);
         channel.runPendingTasks();
 
@@ -310,7 +291,7 @@ public class FilterHandlerTest extends FilterHarness {
     private static ApiVersionsResponseFilter taggingApiVersionsResponseFilter(String tag) {
         return (apiVersion, header, response, context) -> {
             response.unknownTaggedFields().add(new RawTaggedField(ARBITRARY_TAG, tag.getBytes(UTF_8)));
-            return context.responseFilterResultBuilder().forward(header, response).completed();
+            context.forwardResponse(header, response);
         };
     }
 
@@ -322,7 +303,7 @@ public class FilterHandlerTest extends FilterHarness {
     private static ApiVersionsRequestFilter taggingApiVersionsRequestFilter(String tag) {
         return (apiVersion, header, request, context) -> {
             request.unknownTaggedFields().add(new RawTaggedField(ARBITRARY_TAG, tag.getBytes(UTF_8)));
-            return context.requestFilterResultBuilder().forward(header, request).completed();
+            context.forwardRequest(header, request);
         };
     }
 
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaAuthnHandlerTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaAuthnHandlerTest.java
index ecd10c57..bb5c438d 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaAuthnHandlerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaAuthnHandlerTest.java
@@ -11,7 +11,6 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
@@ -50,7 +49,6 @@ import io.netty.channel.embedded.EmbeddedChannel;
 
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ResponseFilter;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 import io.kroxylicious.proxy.frame.BareSaslRequest;
 import io.kroxylicious.proxy.frame.BareSaslResponse;
 import io.kroxylicious.proxy.frame.DecodedRequestFrame;
@@ -210,9 +208,8 @@ public class KafkaAuthnHandlerTest {
             }
 
             @Override
-            public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+            public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
 
-                return null;
             }
         }, new CompletableFuture<>(), true);
 
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderTest.java
index ba059281..7a3b2283 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderTest.java
@@ -8,7 +8,6 @@ package io.kroxylicious.proxy.internal.codec;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.CompletionStage;
 
 import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.RequestHeaderData;
@@ -22,7 +21,6 @@ import io.netty.buffer.Unpooled;
 import io.kroxylicious.proxy.filter.ApiVersionsRequestFilter;
 import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
 import io.kroxylicious.proxy.frame.DecodedRequestFrame;
 import io.kroxylicious.proxy.frame.OpaqueRequestFrame;
 
@@ -31,7 +29,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 public class RequestDecoderTest extends AbstractCodecTest {
 
     @ParameterizedTest
-    @MethodSource("requestApiVersions")
+    @MethodSource("requestApiVersions()")
     public void testApiVersionsExactlyOneFrame_decoded(short apiVersion) throws Exception {
         assertEquals(12,
                 exactlyOneFrame_decoded(apiVersion,
@@ -43,15 +41,14 @@ public class RequestDecoderTest extends AbstractCodecTest {
                         new KafkaRequestDecoder(
                                 DecodePredicate
                                         .forFilters(FilterAndInvoker.build(
-                                                (ApiVersionsRequestFilter) (version, header, request, context) -> context.requestFilterResultBuilder()
-                                                        .forward(header, request).completed()))),
+                                                (ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request)))),
                         DecodedRequestFrame.class,
                         (RequestHeaderData header) -> header),
                 "Unexpected correlation id");
     }
 
     @ParameterizedTest
-    @MethodSource("requestApiVersions")
+    @MethodSource("requestApiVersions()")
     public void testApiVersionsExactlyOneFrame_opaque(short apiVersion) throws Exception {
         assertEquals(12,
                 exactlyOneFrame_encoded(apiVersion,
@@ -67,10 +64,9 @@ public class RequestDecoderTest extends AbstractCodecTest {
                                     }
 
                                     @Override
-                                    public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header,
-                                                                                                     ApiVersionsRequestData request,
-                                                                                                     KrpcFilterContext context) {
-                                        return context.requestFilterResultBuilder().forward(header, request).completed();
+                                    public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
+                                                                     KrpcFilterContext context) {
+                                        context.forwardRequest(header, request);
                                     }
                                 }))),
                         OpaqueRequestFrame.class),
@@ -78,7 +74,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
     }
 
     @ParameterizedTest
-    @MethodSource("requestApiVersions")
+    @MethodSource("requestApiVersions()")
     public void testApiVersionsFrameLessOneByte(short apiVersion) throws Exception {
         RequestHeaderData encodedHeader = exampleRequestHeader(apiVersion);
         ApiVersionsRequestData encodedBody = exampleApiVersionsRequest();
@@ -91,9 +87,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
                 DecodePredicate.forFilters(
-                        FilterAndInvoker.build((ApiVersionsRequestFilter) (version, header, request, context) -> {
-                            return context.requestFilterResultBuilder().forward(header, request).completed();
-                        })))
+                        FilterAndInvoker.build((ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request))))
                 .decode(null, byteBuf, messages);
 
         assertEquals(List.of(), messageClasses(messages));
@@ -112,9 +106,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
                 DecodePredicate.forFilters(
-                        FilterAndInvoker
-                                .build((ApiVersionsRequestFilter) (version, header, request, context) -> context.requestFilterResultBuilder().forward(header, request)
-                                        .completed())))
+                        FilterAndInvoker.build((ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request))))
                 .decode(null, byteBuf, messages);
 
         assertEquals(List.of(), messageClasses(messages));
@@ -122,19 +114,19 @@ public class RequestDecoderTest extends AbstractCodecTest {
     }
 
     @ParameterizedTest
-    @MethodSource("requestApiVersions")
+    @MethodSource("requestApiVersions()")
     public void testApiVersionsFrameFirst3Bytes(short apiVersion) throws Exception {
         doTestApiVersionsFrameFirstNBytes(apiVersion, 3, 0);
     }
 
     @ParameterizedTest
-    @MethodSource("requestApiVersions")
+    @MethodSource("requestApiVersions()")
     public void testApiVersionsFrameFirst5Bytes(short apiVersion) throws Exception {
         doTestApiVersionsFrameFirstNBytes(apiVersion, 5, 0);
     }
 
     @ParameterizedTest
-    @MethodSource("requestApiVersions")
+    @MethodSource("requestApiVersions()")
     public void testApiVersionsExactlyTwoFrames(short apiVersion) throws Exception {
         RequestHeaderData encodedHeader = exampleRequestHeader(apiVersion);
 
@@ -157,9 +149,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
                 DecodePredicate.forFilters(
-                        FilterAndInvoker
-                                .build((ApiVersionsRequestFilter) (version, head, request, context) -> context.requestFilterResultBuilder().forward(header, request)
-                                        .completed())))
+                        FilterAndInvoker.build((ApiVersionsRequestFilter) (version, head, request, context) -> context.forwardRequest(header, request))))
                 .decode(null, byteBuf, messages);
 
         assertEquals(List.of(DecodedRequestFrame.class, DecodedRequestFrame.class), messageClasses(messages));
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilterTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilterTest.java
index 05b3dda2..8e0451df 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilterTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilterTest.java
@@ -19,11 +19,14 @@ import org.apache.kafka.common.message.ApiMessageType;
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.junit.jupiter.MockitoExtension;
@@ -44,6 +47,7 @@ import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.collect.Iterables.getOnlyElement;
 import static io.kroxylicious.test.requestresponsetestdef.KafkaApiMessageConverter.responseConverterFor;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -76,6 +80,9 @@ class BrokerAddressFilterTest {
     @Mock
     private KrpcFilterContext context;
 
+    @Captor
+    private ArgumentCaptor<ApiMessage> apiMessageCaptor;
+
     private BrokerAddressFilter filter;
 
     private FilterInvoker invoker;
@@ -107,36 +114,31 @@ class BrokerAddressFilterTest {
     @ParameterizedTest(name = "{0}")
     @MethodSource(value = "nodeInfoCarryingResponses")
     void nodeInfoCarryingResponsesTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header,
-                                              ApiMessageTestDef responseTestDef)
-            throws Exception {
+                                              ApiMessageTestDef responseTestDef) {
         filterResponseAndVerify(apiMessageType, header, responseTestDef);
     }
 
     @ParameterizedTest(name = "{0}")
     @MethodSource(value = "completeClusterInfoCarryingResponses")
     void reconcileCachesUpstreamAddress(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header,
-                                        ApiMessageTestDef responseTestDef)
-            throws Exception {
+                                        ApiMessageTestDef responseTestDef) {
 
         filterResponseAndVerify(apiMessageType, header, responseTestDef);
         verify(endpointReconciler, times(1)).reconcile(Mockito.eq(virtualCluster), Mockito.anyMap());
     }
 
-    private void filterResponseAndVerify(ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef responseTestDef) throws Exception {
+    private void filterResponseAndVerify(ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef responseTestDef) {
         var response = responseTestDef.message();
         // marshalled the response object back to json, this is used for comparison later.
         var responseWriter = responseConverterFor(apiMessageType).writer();
 
         var marshalled = responseWriter.apply(response, header.requestApiVersion());
 
-        when(context.responseFilterResultBuilder()).thenReturn(new ResponseFilterResultBuilderImpl());
-
         ResponseHeaderData headerData = new ResponseHeaderData();
-        var stage = invoker.onResponse(ApiKeys.forId(apiMessageType.apiKey()), header.requestApiVersion(), headerData, response, context);
-        assertThat(stage).isCompleted();
-        var value = stage.toCompletableFuture().get().message();
+        invoker.onResponse(ApiKeys.forId(apiMessageType.apiKey()), header.requestApiVersion(), headerData, response, context);
+        verify(context).forwardResponse(any(), apiMessageCaptor.capture());
 
-        var filtered = responseWriter.apply(value, header.requestApiVersion());
+        var filtered = responseWriter.apply(apiMessageCaptor.getValue(), header.requestApiVersion());
         assertThat(JsonDiff.asJson(marshalled, filtered)).isEqualTo(responseTestDef.expectedPatch());
     }
 
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearnerTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearnerTest.java
index 084c26d0..8dfc306d 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearnerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearnerTest.java
@@ -36,10 +36,11 @@ import org.mockito.junit.jupiter.MockitoExtension;
 
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 
-import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.anyShort;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -65,9 +66,8 @@ class EagerMetadataLearnerTest {
     @ParameterizedTest(name = "{0}")
     @MethodSource("preludeRequests")
     public void forwardsRequestsOfKafkaPrelude(String name, ApiKeys apiKey, RequestHeaderData header, ApiMessage request) {
-        when(context.requestFilterResultBuilder()).thenReturn(new RequestFilterResultBuilderImpl());
-        var stage = learner.onRequest(apiKey, header, request, context);
-        assertThat(stage).isCompleted();
+        learner.onRequest(apiKey, header, request, context);
+        verify(context).forwardRequest(header, request);
     }
 
     public static Stream<Arguments> postPreludeRequests() {
@@ -81,25 +81,22 @@ class EagerMetadataLearnerTest {
 
     @ParameterizedTest(name = "{0}")
     @MethodSource("postPreludeRequests")
-    public void spontaneouslyEmitsMetadataRequest(String name, ApiKeys apiKey, RequestHeaderData header, ApiMessage request) throws Exception {
+    public void spontaneouslyEmitsMetadataRequest(String name, ApiKeys apiKey, RequestHeaderData header, ApiMessage request) {
         var metadataResponse = new MetadataResponseData();
         metadataResponse.brokers().add(new MetadataResponseData.MetadataResponseBroker().setNodeId(1).setHost("localhost").setPort(1234));
 
-        when(context.requestFilterResultBuilder()).thenReturn(new RequestFilterResultBuilderImpl());
         when(context.sendRequest(anyShort(), isA(MetadataRequestData.class))).thenReturn(CompletableFuture.completedStage(metadataResponse));
-        var stage = learner.onRequest(apiKey, header, request, context);
-        assertThat(stage).isCompleted();
-        var result = stage.toCompletableFuture().get();
+        learner.onRequest(apiKey, header, request, context);
 
         if (apiKey == ApiKeys.METADATA) {
             // if caller's request is a metadata request, then the filter must forward it with fidelity
             verify(context).sendRequest(eq(header.requestApiVersion()), eq(request));
-            assertThat(result.message()).isEqualTo(metadataResponse);
         }
         else {
             verify(context).sendRequest(anyShort(), isA(MetadataRequestData.class));
         }
-        assertThat(result.closeConnection()).isTrue();
+        verify(context, apiKey.equals(ApiKeys.METADATA) ? times(1) : never()).forwardResponse(metadataResponse);
+        verify(context).closeConnection();
     }
 
     private static Arguments toArgs(String name, AbstractRequest request) {
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/RequestFilterResultBuilderTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/RequestFilterResultBuilderTest.java
deleted file mode 100644
index 90c18e4f..00000000
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/RequestFilterResultBuilderTest.java
+++ /dev/null
@@ -1,129 +0,0 @@
-/*
- * 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.filter;
-
-import org.apache.kafka.common.message.FetchRequestData;
-import org.apache.kafka.common.message.FetchResponseData;
-import org.apache.kafka.common.message.RequestHeaderData;
-import org.apache.kafka.common.message.ResponseHeaderData;
-import org.junit.jupiter.api.Test;
-
-import io.kroxylicious.proxy.filter.RequestFilterResultBuilder;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-class RequestFilterResultBuilderTest {
-
-    private final RequestFilterResultBuilder builder = new RequestFilterResultBuilderImpl();
-
-    @Test
-    void forwardRequest() {
-        var request = new FetchRequestData();
-        var header = new RequestHeaderData();
-        var result = builder.forward(header, request).build();
-        assertThat(result.message()).isEqualTo(request);
-        assertThat(result.header()).isEqualTo(header);
-        assertThat(result.closeConnection()).isFalse();
-        assertThat(result.drop()).isFalse();
-    }
-
-    @Test
-    void forwardRejectResponseData() {
-        var res = new FetchResponseData();
-        var header = new RequestHeaderData();
-        assertThatThrownBy(() -> builder.forward(header, res)).isInstanceOf(IllegalArgumentException.class);
-    }
-
-    @Test
-    void forwardRejectNullResponseData() {
-        var header = new RequestHeaderData();
-        assertThatThrownBy(() -> builder.forward(header, null)).isInstanceOf(IllegalArgumentException.class);
-    }
-
-    @Test
-    void forwardRejectsNullHeader() {
-        var req = new FetchRequestData();
-        assertThatThrownBy(() -> builder.forward(null, req)).isInstanceOf(IllegalArgumentException.class);
-    }
-
-    @Test
-    void bareCloseConnection() {
-        var result = builder.withCloseConnection().build();
-        assertThat(result.closeConnection()).isTrue();
-    }
-
-    @Test
-    void forwardWithCloseConnection() {
-        var request = new FetchRequestData();
-        var header = new RequestHeaderData();
-
-        var result = builder.forward(header, request).withCloseConnection().build();
-        assertThat(result.message()).isEqualTo(request);
-        assertThat(result.header()).isEqualTo(header);
-        assertThat(result.closeConnection()).isTrue();
-    }
-
-    @Test
-    void shortCircuit() {
-        var res = new FetchResponseData();
-        var result = builder.shortCircuitResponse(res).build();
-        assertThat(result.message()).isEqualTo(res);
-        assertThat(result.header()).isNull();
-        assertThat(result.closeConnection()).isFalse();
-    }
-
-    @Test
-    void shortCircuitResultWithCloseConnection() {
-        var res = new FetchResponseData();
-        var result = builder.shortCircuitResponse(res).withCloseConnection().build();
-        assertThat(result.message()).isEqualTo(res);
-        assertThat(result.header()).isNull();
-        assertThat(result.closeConnection()).isTrue();
-    }
-
-    @Test
-    void shortCircuitHeaderAndResponseData() {
-        var res = new FetchResponseData();
-        var header = new ResponseHeaderData();
-        var result = builder.shortCircuitResponse(header, res).build();
-        assertThat(result.message()).isEqualTo(res);
-        assertThat(result.header()).isEqualTo(header);
-        assertThat(result.closeConnection()).isFalse();
-    }
-
-    @Test
-    void shortCircuitRejectsRequestData() {
-        var req = new FetchRequestData();
-        assertThatThrownBy(() -> builder.shortCircuitResponse(req)).isInstanceOf(IllegalArgumentException.class);
-    }
-
-    @Test
-    void shortCircuitRejectsNullRequestData() {
-        assertThatThrownBy(() -> builder.shortCircuitResponse(null)).isInstanceOf(IllegalArgumentException.class);
-    }
-
-    @Test
-    void drop() {
-        var result = builder.drop().build();
-        assertThat(result.drop()).isTrue();
-        assertThat(result.message()).isNull();
-        assertThat(result.header()).isNull();
-    }
-
-    @Test
-    void completedApi() throws Exception {
-        var request = new FetchRequestData();
-        var header = new RequestHeaderData();
-        var future = builder.forward(header, request).completed();
-        assertThat(future).isCompleted();
-        var result = future.toCompletableFuture().get();
-        assertThat(result.message()).isEqualTo(request);
-        assertThat(result.header()).isEqualTo(header);
-    }
-
-}
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/ResponseFilterResultBuilderTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/ResponseFilterResultBuilderTest.java
deleted file mode 100644
index 356c6b68..00000000
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/ResponseFilterResultBuilderTest.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * 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.filter;
-
-import org.apache.kafka.common.message.FetchRequestData;
-import org.apache.kafka.common.message.FetchResponseData;
-import org.apache.kafka.common.message.ResponseHeaderData;
-import org.junit.jupiter.api.Test;
-
-import io.kroxylicious.proxy.filter.ResponseFilterResultBuilder;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-class ResponseFilterResultBuilderTest {
-
-    private final ResponseFilterResultBuilder builder = new ResponseFilterResultBuilderImpl();
-
-    @Test
-    void forwardResponse() {
-        var res = new FetchResponseData();
-        var header = new ResponseHeaderData();
-        var result = builder.forward(header, res).build();
-        assertThat(result.message()).isEqualTo(res);
-        assertThat(result.header()).isEqualTo(header);
-        assertThat(result.closeConnection()).isFalse();
-        assertThat(result.drop()).isFalse();
-    }
-
-    @Test
-    void forwardRejectsRequestData() {
-        var req = new FetchRequestData();
-        var header = new ResponseHeaderData();
-        assertThatThrownBy(() -> builder.forward(header, req)).isInstanceOf(IllegalArgumentException.class);
-    }
-
-    @Test
-    void forwardRejectsNullRequestData() {
-        var header = new ResponseHeaderData();
-        assertThatThrownBy(() -> builder.forward(header, null)).isInstanceOf(IllegalArgumentException.class);
-    }
-
-    @Test
-    void bareCloseConnection() {
-        var result = builder.withCloseConnection().build();
-        assertThat(result.closeConnection()).isTrue();
-    }
-
-    @Test
-    void forwardResponseWithCloseConnection() {
-        var res = new FetchResponseData();
-        var header = new ResponseHeaderData();
-        var result = builder.forward(header, res).withCloseConnection().build();
-        assertThat(result.message()).isEqualTo(res);
-        assertThat(result.header()).isEqualTo(header);
-        assertThat(result.closeConnection()).isTrue();
-    }
-
-    @Test
-    void drop() {
-        var result = builder.drop().build();
-        assertThat(result.drop()).isTrue();
-        assertThat(result.message()).isNull();
-        assertThat(result.header()).isNull();
-    }
-
-    @Test
-    void completedApi() throws Exception {
-        var res = new FetchResponseData();
-        var header = new ResponseHeaderData();
-        var future = builder.forward(header, res).completed();
-        assertThat(future).isCompleted();
-        var result = future.toCompletableFuture().get();
-        assertThat(result.message()).isEqualTo(res);
-        assertThat(result.header()).isEqualTo(header);
-    }
-}
diff --git a/performance-tests/pom.xml b/performance-tests/pom.xml
index 901a8749..93e5c507 100644
--- a/performance-tests/pom.xml
+++ b/performance-tests/pom.xml
@@ -23,7 +23,6 @@
     <properties>
         <jmh.version>1.36</jmh.version>
         <uberjar.name>benchmarks</uberjar.name>
-        <sonar.skip>true</sonar.skip>
     </properties>
 
     <dependencies>
diff --git a/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
index f8d4873e..d47ca7fd 100644
--- a/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
+++ b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
@@ -38,10 +38,6 @@ import io.kroxylicious.proxy.filter.FilterInvoker;
 import io.kroxylicious.proxy.filter.FilterInvokers;
 import io.kroxylicious.proxy.filter.KrpcFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.RequestFilterResultBuilder;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResultBuilder;
 import io.kroxylicious.proxy.filter.SpecificFilterInvoker;
 
 // try hard to make shouldHandleXYZ to observe different receivers concrete types, saving unrolling to bias a specific call-site to a specific concrete type
@@ -174,29 +170,28 @@ public class InvokerDispatchBenchmark {
         }
 
         @Override
-        public <T extends ApiMessage> CompletionStage<T> sendRequest(short apiVersion, ApiMessage request) {
-            return null;
+        public void forwardRequest(RequestHeaderData header, ApiMessage request) {
+
         }
 
         @Override
-        public CompletionStage<ResponseFilterResult> forwardResponse(ResponseHeaderData header, ApiMessage response) {
+        public <T extends ApiMessage> CompletionStage<T> sendRequest(short apiVersion, ApiMessage request) {
             return null;
         }
 
         @Override
-        public RequestFilterResultBuilder requestFilterResultBuilder() {
-            return null;
+        public void forwardResponse(ResponseHeaderData header, ApiMessage response) {
+
         }
 
         @Override
-        public CompletionStage<RequestFilterResult> forwardRequest(RequestHeaderData header, ApiMessage request) {
-            return null;
+        public void forwardResponse(ApiMessage response) {
+
         }
 
         @Override
-        public ResponseFilterResultBuilder responseFilterResultBuilder() {
-            return null;
-        }
+        public void closeConnection() {
 
+        }
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/EightInterfaceFilter.java b/performance-tests/src/main/java/io/kroxylicious/filters/EightInterfaceFilter.java
index 1023de78..5e372c9d 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/EightInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/EightInterfaceFilter.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.filters;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.CreateTopicsRequestData;
 import org.apache.kafka.common.message.CreateTopicsResponseData;
 import org.apache.kafka.common.message.DeleteTopicsRequestData;
@@ -28,62 +26,45 @@ import io.kroxylicious.proxy.filter.DescribeGroupsResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 
 public class EightInterfaceFilter implements ProduceResponseFilter, ProduceRequestFilter, CreateTopicsRequestFilter, CreateTopicsResponseFilter,
         DeleteTopicsRequestFilter, DeleteTopicsResponseFilter, DescribeGroupsRequestFilter, DescribeGroupsResponseFilter {
 
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
-        return null;
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
-                                                                   KrpcFilterContext context) {
-        return null;
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request,
-                                                                      KrpcFilterContext context) {
+    public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
 
-        return null;
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response,
-                                                                        KrpcFilterContext context) {
+    public void onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
 
-        return null;
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onDeleteTopicsRequest(short apiVersion, RequestHeaderData header, DeleteTopicsRequestData request,
-                                                                      KrpcFilterContext context) {
+    public void onDeleteTopicsRequest(short apiVersion, RequestHeaderData header, DeleteTopicsRequestData request, KrpcFilterContext context) {
 
-        return null;
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onDeleteTopicsResponse(short apiVersion, ResponseHeaderData header, DeleteTopicsResponseData response,
-                                                                        KrpcFilterContext context) {
+    public void onDeleteTopicsResponse(short apiVersion, ResponseHeaderData header, DeleteTopicsResponseData response, KrpcFilterContext context) {
 
-        return null;
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onDescribeGroupsRequest(short apiVersion, RequestHeaderData header, DescribeGroupsRequestData request,
-                                                                        KrpcFilterContext context) {
+    public void onDescribeGroupsRequest(short apiVersion, RequestHeaderData header, DescribeGroupsRequestData request, KrpcFilterContext context) {
 
-        return null;
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onDescribeGroupsResponse(short apiVersion, ResponseHeaderData header, DescribeGroupsResponseData response,
-                                                                          KrpcFilterContext context) {
+    public void onDescribeGroupsResponse(short apiVersion, ResponseHeaderData header, DescribeGroupsResponseData response, KrpcFilterContext context) {
 
-        return null;
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter.java b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter.java
index 92d3dfdc..c3a22e0c 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.filters;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.CreateTopicsRequestData;
 import org.apache.kafka.common.message.CreateTopicsResponseData;
 import org.apache.kafka.common.message.ProduceRequestData;
@@ -20,33 +18,24 @@ import io.kroxylicious.proxy.filter.CreateTopicsResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 
 public class FourInterfaceFilter implements ProduceResponseFilter, ProduceRequestFilter, CreateTopicsRequestFilter, CreateTopicsResponseFilter {
 
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
-        return null;
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
-                                                                   KrpcFilterContext context) {
-        return null;
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request,
-                                                                      KrpcFilterContext context) {
+    public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
 
-        return null;
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response,
-                                                                        KrpcFilterContext context) {
+    public void onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
 
-        return null;
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter0.java b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter0.java
index 6bd2e4b2..db734d4e 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter0.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter0.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.filters;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.ApiVersionsResponseData;
 import org.apache.kafka.common.message.ProduceRequestData;
@@ -21,37 +19,28 @@ import io.kroxylicious.proxy.filter.ApiVersionsResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 
 import static io.kroxylicious.benchmarks.InvokerDispatchBenchmark.CONSUME_TOKENS;
 
 public class FourInterfaceFilter0 implements ProduceResponseFilter, ProduceRequestFilter, ApiVersionsRequestFilter, ApiVersionsResponseFilter {
 
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
-                                                                   KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
-                                                                     KrpcFilterContext context) {
+    public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response,
-                                                                       KrpcFilterContext context) {
+    public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter1.java b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter1.java
index 46f1675a..36c690a8 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter1.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter1.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.filters;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.ApiVersionsResponseData;
 import org.apache.kafka.common.message.ProduceRequestData;
@@ -21,36 +19,27 @@ import io.kroxylicious.proxy.filter.ApiVersionsResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 
 import static io.kroxylicious.benchmarks.InvokerDispatchBenchmark.CONSUME_TOKENS;
 
 public class FourInterfaceFilter1 implements ProduceResponseFilter, ProduceRequestFilter, ApiVersionsRequestFilter, ApiVersionsResponseFilter {
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
-                                                                   KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
-                                                                     KrpcFilterContext context) {
+    public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response,
-                                                                       KrpcFilterContext context) {
+    public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter2.java b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter2.java
index a5de5eda..eaa7240b 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter2.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter2.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.filters;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.FetchRequestData;
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.message.ProduceRequestData;
@@ -21,35 +19,28 @@ import io.kroxylicious.proxy.filter.FetchResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 
 import static io.kroxylicious.benchmarks.InvokerDispatchBenchmark.CONSUME_TOKENS;
 
 public class FourInterfaceFilter2 implements ProduceResponseFilter, ProduceRequestFilter, FetchRequestFilter, FetchResponseFilter {
 
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
-                                                                   KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
+    public void onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
+    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter3.java b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter3.java
index 8b156aba..90c65d97 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter3.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter3.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.filters;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.FetchRequestData;
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.message.ProduceRequestData;
@@ -21,35 +19,28 @@ import io.kroxylicious.proxy.filter.FetchResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 
 import static io.kroxylicious.benchmarks.InvokerDispatchBenchmark.CONSUME_TOKENS;
 
 public class FourInterfaceFilter3 implements ProduceResponseFilter, ProduceRequestFilter, FetchRequestFilter, FetchResponseFilter {
 
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
-                                                                   KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
+    public void onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
+    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
-        return null;
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/OneInterfaceFilter.java b/performance-tests/src/main/java/io/kroxylicious/filters/OneInterfaceFilter.java
index cf727a7b..bacb5e2f 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/OneInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/OneInterfaceFilter.java
@@ -6,20 +6,15 @@
 
 package io.kroxylicious.filters;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.ProduceResponseData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 
 public class OneInterfaceFilter implements ProduceResponseFilter {
 
     @Override
-    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
-                                                                   KrpcFilterContext context) {
-        return null;
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter.java b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter.java
index 0b1c9d2c..e1c3459d 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.filters;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceResponseData;
 import org.apache.kafka.common.message.RequestHeaderData;
@@ -16,19 +14,14 @@ import org.apache.kafka.common.message.ResponseHeaderData;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 
 public class TwoInterfaceFilter implements ProduceResponseFilter, ProduceRequestFilter {
 
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
-        return null;
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
-                                                                   KrpcFilterContext context) {
-        return null;
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter0.java b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter0.java
index 5a2bcd37..9c249754 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter0.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter0.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.filters;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceResponseData;
 import org.apache.kafka.common.message.RequestHeaderData;
@@ -16,19 +14,14 @@ import org.apache.kafka.common.message.ResponseHeaderData;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 
 public class TwoInterfaceFilter0 implements ProduceResponseFilter, ProduceRequestFilter {
 
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
-        return null;
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
-                                                                   KrpcFilterContext context) {
-        return null;
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter1.java b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter1.java
index 05a06e6b..a58cd4b7 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter1.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter1.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.filters;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceResponseData;
 import org.apache.kafka.common.message.RequestHeaderData;
@@ -16,19 +14,14 @@ import org.apache.kafka.common.message.ResponseHeaderData;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.ProduceRequestFilter;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 
 public class TwoInterfaceFilter1 implements ProduceResponseFilter, ProduceRequestFilter {
 
     @Override
-    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
-        return null;
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
     }
 
     @Override
-    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
-                                                                   KrpcFilterContext context) {
-        return null;
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/proxy/filter/ArrayFilterInvoker.java b/performance-tests/src/main/java/io/kroxylicious/proxy/filter/ArrayFilterInvoker.java
index e02940f9..f7275cca 100644
--- a/performance-tests/src/main/java/io/kroxylicious/proxy/filter/ArrayFilterInvoker.java
+++ b/performance-tests/src/main/java/io/kroxylicious/proxy/filter/ArrayFilterInvoker.java
@@ -9,7 +9,6 @@ import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.OptionalInt;
-import java.util.concurrent.CompletionStage;
 
 import org.apache.kafka.common.message.ApiMessageType;
 import org.apache.kafka.common.message.RequestHeaderData;
@@ -445,42 +444,38 @@ public class ArrayFilterInvoker implements FilterInvoker {
 
     /**
      * Apply the filter to the given {@code header} and {@code body} using the given {@code filterContext}.
-     *
-     * @param apiKey        The request api key.
-     * @param apiVersion    The request api version.
-     * @param header        The request header.
-     * @param body          The request body.
+     * @param apiKey The request api key.
+     * @param apiVersion The request api version.
+     * @param header The request header.
+     * @param body The request body.
      * @param filterContext The filter context.
-     * @return
      */
     @Override
-    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey,
-                                                          short apiVersion,
-                                                          RequestHeaderData header,
-                                                          ApiMessage body,
-                                                          KrpcFilterContext filterContext) {
+    public void onRequest(ApiKeys apiKey,
+                          short apiVersion,
+                          RequestHeaderData header,
+                          ApiMessage body,
+                          KrpcFilterContext filterContext) {
         FilterInvoker invoker = requestInvokers[apiKey.id];
-        return invoker.onRequest(apiKey, apiVersion, header, body, filterContext);
+        invoker.onRequest(apiKey, apiVersion, header, body, filterContext);
     }
 
     /**
      * Apply the filter to the given {@code header} and {@code body} using the given {@code filterContext}.
-     *
-     * @param apiKey        The request api key.
-     * @param apiVersion    The api version.
-     * @param header        The request header.
-     * @param body          The request body.
+     * @param apiKey The request api key.
+     * @param apiVersion The api version.
+     * @param header The request header.
+     * @param body The request body.
      * @param filterContext The filter context.
-     * @return
      */
     @Override
-    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey,
-                                                            short apiVersion,
-                                                            ResponseHeaderData header,
-                                                            ApiMessage body,
-                                                            KrpcFilterContext filterContext) {
+    public void onResponse(ApiKeys apiKey,
+                           short apiVersion,
+                           ResponseHeaderData header,
+                           ApiMessage body,
+                           KrpcFilterContext filterContext) {
         FilterInvoker invoker = responseInvokers[apiKey.id];
-        return invoker.onResponse(apiKey, apiVersion, header, body, filterContext);
+        invoker.onResponse(apiKey, apiVersion, header, body, filterContext);
     }
 
     /**
diff --git a/performance-tests/src/main/java/io/kroxylicious/proxy/filter/SpecificFilterInvoker.java b/performance-tests/src/main/java/io/kroxylicious/proxy/filter/SpecificFilterInvoker.java
index d986796f..d3d586dd 100644
--- a/performance-tests/src/main/java/io/kroxylicious/proxy/filter/SpecificFilterInvoker.java
+++ b/performance-tests/src/main/java/io/kroxylicious/proxy/filter/SpecificFilterInvoker.java
@@ -6,8 +6,6 @@
 
 package io.kroxylicious.proxy.filter;
 
-import java.util.concurrent.CompletionStage;
-
 import org.apache.kafka.common.message.AddOffsetsToTxnRequestData;
 import org.apache.kafka.common.message.AddOffsetsToTxnResponseData;
 import org.apache.kafka.common.message.AddPartitionsToTxnRequestData;
@@ -161,23 +159,21 @@ public class SpecificFilterInvoker implements FilterInvoker {
         this.filter = filter;
     }
 
-    /**X
+    /**
      * Apply the filter to the given {@code header} and {@code body} using the given {@code filterContext}.
-     *
-     * @param apiKey        The request api key.
-     * @param apiVersion    The request api version.
-     * @param header        The request header.
-     * @param body          The request body.
+     * @param apiKey The request api key.
+     * @param apiVersion The request api version.
+     * @param header The request header.
+     * @param body The request body.
      * @param filterContext The filter context.
-     * @return
      */
     @Override
-    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey,
-                                                          short apiVersion,
-                                                          RequestHeaderData header,
-                                                          ApiMessage body,
-                                                          KrpcFilterContext filterContext) {
-        return switch (apiKey) {
+    public void onRequest(ApiKeys apiKey,
+                          short apiVersion,
+                          RequestHeaderData header,
+                          ApiMessage body,
+                          KrpcFilterContext filterContext) {
+        switch (apiKey) {
             case ADD_OFFSETS_TO_TXN ->
                 ((AddOffsetsToTxnRequestFilter) filter).onAddOffsetsToTxnRequest(apiVersion, header, (AddOffsetsToTxnRequestData) body, filterContext);
             case ADD_PARTITIONS_TO_TXN ->
@@ -282,26 +278,24 @@ public class SpecificFilterInvoker implements FilterInvoker {
             case WRITE_TXN_MARKERS ->
                 ((WriteTxnMarkersRequestFilter) filter).onWriteTxnMarkersRequest(apiVersion, header, (WriteTxnMarkersRequestData) body, filterContext);
             default -> throw new IllegalStateException("Unsupported RPC " + apiKey);
-        };
+        }
     }
 
     /**
      * Apply the filter to the given {@code header} and {@code body} using the given {@code filterContext}.
-     *
-     * @param apiKey        The request api key.
-     * @param apiVersion    The api version.
-     * @param header        The request header.
-     * @param body          The request body.
+     * @param apiKey The request api key.
+     * @param apiVersion The api version.
+     * @param header The request header.
+     * @param body The request body.
      * @param filterContext The filter context.
-     * @return
      */
     @Override
-    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey,
-                                                            short apiVersion,
-                                                            ResponseHeaderData header,
-                                                            ApiMessage body,
-                                                            KrpcFilterContext filterContext) {
-        return switch (apiKey) {
+    public void onResponse(ApiKeys apiKey,
+                           short apiVersion,
+                           ResponseHeaderData header,
+                           ApiMessage body,
+                           KrpcFilterContext filterContext) {
+        switch (apiKey) {
             case ADD_OFFSETS_TO_TXN ->
                 ((AddOffsetsToTxnResponseFilter) filter).onAddOffsetsToTxnResponse(apiVersion, header, (AddOffsetsToTxnResponseData) body, filterContext);
             case ADD_PARTITIONS_TO_TXN ->
@@ -414,7 +408,7 @@ public class SpecificFilterInvoker implements FilterInvoker {
             case WRITE_TXN_MARKERS ->
                 ((WriteTxnMarkersResponseFilter) filter).onWriteTxnMarkersResponse(apiVersion, header, (WriteTxnMarkersResponseData) body, filterContext);
             default -> throw new IllegalStateException("Unsupported RPC " + apiKey);
-        };
+        }
     }
 
     /**