Repository: kroxylicious/kroxylicious
Pull request #494 (2023-07-26 16:44:18)

Async Filter API
### Type of change

- Enhancement / new feature

### Description

Make the Filter API fully asynchronous.   The key idea is to have the `onXXXRequest|Response` methods return an CompletionStage<FilterResult> that yields the header/message to forward and provides instructions such as 'close-connection' or 'drop request/response'.

Why: some methods of the Filter API are already asynchronous, other parts are synchronously.   I think by having a fully asynchronous API, a simpler API will result.

**Note - the underlying implementation still does not prevent requests/responses going out of order. My intention is fix that only once/if we agree that the API shape is right. **

#### Sample Gestures

I'm illustrating the request side below.  The response side follows the exactly the same pattern except the `onXXXResponse` methods return a `CompletionStage<ResponseFilterResult>`.
 

##### Simple forward 

(filter returns a completed future containing the result)

```java
    public CompletionStage<? extends FilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
        // mutate request
        return context.completedForwardRequest(header, request);
    }
```
 
##### In-band request

(returns a uncompleted future, which is completed by the async request with the result)

```java
    @Override
    public CompletionStage<? extends FilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
        var builder = context.requestFilterResultBuilder();
        var future = new CompletableFuture<RequestFilterResult>();
        context.sendRequest(..., ...).thenAccept(response -> {
            // augment request based on response
            builder.withMessage(request);
            future.complete(builder.build());
        });
        return future;
    }
```

##### Out of band request

```java
    @Override
    public CompletionStage<? extends FilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
        var builder = context.requestFilterResultBuilder();
        // dispatches an async request
        context.sendRequest(..., ...).thenAccept(response -> {
            ....
        });
        return context.completedForwardRequest(header, request);
    }
```

##### Short circuit (without connection close)

(a filter that returns an response, bypassing the trip to the broker)

```java
    @Override
    public CompletionStage<? extends FilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
        var response = make error response.
        return context.completedForwardResponse(response);
    }
```

##### Short circuit (with connection close)

(a filter that returns an response, bypassing the trip to the broker, and closes the connection)

```java
    @Override
    public CompletionStage<? extends FilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
        var builder = context.responseFilterResultBuilder()
                .withMessage(errorResponse)
                .withCloseConnection(true);
        
        return CompletableFuture.completedStage(builder.build());
    }
```

##### Alternative way to close connection

(if the future yields an exception, that will always close the connection (without a response to the client))

```java
    public CompletionStage<? extends FilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
        return CompletableFuture.failedStage(new IllegalStateException("i'm out of here"));
    }
```

##### Drop request

(This one isn't implemented, but it would look like this.  Drop request only makes sense for PublishRequests where acks=0)

    public CompletionStage<? extends FilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
        var builder = context.responseFilterResultBuilder()
                .withDrop(true);
        return CompletableFuture.completedStage(builder.build());
    }



### Additional Context

_Why are you making this pull request?_

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [x] Write tests
- [x] Make sure all tests pass
- [ ] Review performance test results. Ensure that any degradations to performance numbers are understood and justified.
- [x] Make sure all Sonar-Lint warnings are addressed or are justifiably ignored.
- [ ] Update documentation
- [x] Reference relevant issue(s) and close them after merging
- [x] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [13 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![7.3%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/10-16px.png '7.3%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [7.3% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [13 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![7.2%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/10-16px.png '7.2%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [7.2% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [13 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![7.2%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/10-16px.png '7.2%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [7.2% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [14 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![6.1%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/10-16px.png '6.1%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [6.1% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
k-wall: @robobario @SamBarker please take a look at the API proposal.   If you think it is somewhere in the right solar system, feel free to collaborate on the branch to make improvements to it. push to the branch or raise PRs, whatever works best for you. 
robobario: One thing I'm wondering about is which thing should be queued once some async work is kicked off? You could queue the messages or you could chain the futures together so that the channel interactions are ordered. 

Depending on what the user is doing, queuing up the channel interactions could be a more efficient. Say there is some async work to offload to another non-netty thread, we could kick off this work for multiple messages, have the work complete in any order but interact with the channel in the order messages were handled.

And from the other angle, the user might expect each message to be handled fully before the filter moves on to the next for some reason. Maybe they expect to update some state in the Filter before moving on.
robobario: On the proposed drop API should we somehow try to make them mutually exclusive in the API?

```
context.responseFilterResultBuilder()
            .withMessage(msg)
            .withDrop(true);
```

would be a contradiction right? I'm wondering if it's worth having a `context.dropBuilder()` or something. Or we could validate it in the `build()`. Or have `context.responseFilterResultBuilder().drop().withClose(true).build()` or `context.responseFilterResultBuilder().forward().withMessage(msg).withHeader(header).build()`.
robobario: So some pros:

1. Same programming model for "extra requests" and standard proxied request. So you could do things like chain the sendRequestFuture through some transformations and then return that future.
2. User are pushed to return a single forward (or other termination like drop/close) for each request/response, where with the current context methods they can do odd things like call `forwardRequest` or `forwardResponse` multiple times.

I'm coming down on the side of this being a better API than what I've been prototyping because it removes that ambiguity.
robobario: I think I'd like to see if we can make it typesafe like:

```
    public CompletionStage<ResponseFilterResult<ProduceRequestData>> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context)
```

so that the user will be made to supply `ProduceRequestData` back to it (or close/drop).
k-wall: > One thing I'm wondering about is which thing should be queued once some async work is kicked off? You could queue the messages or you could chain the futures together so that the channel interactions are ordered.

I'm not sure I'm quite following.

My mental model:

* Apache Kafka requires that responses go back to the client are in same order that the requests were sent.  The only exception being Produce(acks=0) requests that have no response. 
* As Kroxylicious filters are stateful, I think the same ordering guarantees needs to exist.
   * A request filter sees the requests in exactly the same order that the client sent them.
   * A response filter sees the responses in exactly the same order as the broker sent them.

If Kroxylicious Filters didn't provide that guarantee, Filters could see messages out of order.  It would be hard for Filter Author to reason about the system.
 
I don't see any way to achieve this without pausing the channel. I know that Netty performs best when channels flow free and there will be performance penalties.  We can mitigate that by:
a) optimising a fast path where the filter future is already complete - there's no need to touch autoread in this case
b) through user education, teach kroxylicious user to adopt patterns to minimise the need for out of band/in band requests by using caching.

In the case of internal use-cases, like the `EndpointRegisty#reconcile`, for the majority of connections, the reconcile future will be complete.  That should most the vast majority of cases follow the a) path.



> 
> Depending on what the user is doing, queuing up the channel interactions could be a more efficient. Say there is some async work to offload to another non-netty thread, we could kick off this work for multiple messages, have the work complete in any order but interact with the channel in the order messages were handled.
> 
> And from the other angle, the user might expect each message to be handled fully before the filter moves on to the next for some reason. Maybe they expect to update some state in the Filter before moving on.


k-wall: Hey @franz1981 in https://github.com/kroxylicious/kroxylicious/issues/351#issuecomment-1573472551 you expressed a performance concern about toggling autoread.  Can we talk about that more in this thread?

In Kroxylicious we envisage that some filters might need to perform asynchronous tasks.  In some of those cases, the filter might need to await for completion before the filter can pass the message to the next filter in the chain.  Example use-cases:

1. looking up an encryption key from a 3rd party KMS so Kroxy knows the key to use to cipher a payload message.  This will likely boil down to a REST API to another system.
2. resolving a topic UID to a topic name.  This is implemented by sending an additional request (DescribeTopic) to the same Kafka Cluster that is being proxied.  This request/response will go down the same channel.

For 1, it is undesirable to block netty's thread as this will impede progress that could be made servicing other connections. For 2, blocking netty's thread isn't an option.  The additional request is going down the same channel, so control has to go back to Netty's event loop to allow the additional request to be sent and the response received.

Apache Kafka relies on message ordering. I think this means when a Kroxylicious Filter needs to do async work, we must stop reading from the client, otherwise message will go out of order.   Is toggling autoread is the correct way to achieve this in Netty?  Are there other approaches to consider?

My *hope* is that the number of cases where filters actually needs to do async work will be kept small. In other words, mostly it won't follow these expensive code paths. Going back to the example, in the topic encryption case, we'll be employing cache so that it is only first connection that incurs the cost of the call to KMS. Subsequent connections will use the cached value, so their connections won't go the expensive route.

Does this approach sound reasonable?   Any pitfalls?
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [23 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![5.4%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/10-16px.png '5.4%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [5.4% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [23 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![5.6%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/10-16px.png '5.6%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [5.6% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
franz1981: > Can we talk about that more in this thread?

I'll be on PTO starting from today afternoon till 18 Aug, but I'll try answering as you need more clarifications on it...

> I don't see any way to achieve this without pausing the channel.

That's ok, but you can buffer it further to avoid it to happen too frequently.
The other thing is: you'll never get a full control of what's read out of the network with Netty anyway, because it tries hard to read in batches. 
Let's say you disable autoread (always, from the start) and you are expected to call read every-time you can read some data, when it happens you still are at the mercy of the size of the estimated receiver buffer from Netty + the data actually resident in the TCP read buffers (meaning that you can still read more than expected). 
Applying some backpressure is still possible IMO, but it's key to accept that will always be a "best-effort" (async and not precise) mechanism.

I suggest to use an intermediate receiver buffer, so:
- buffer the received data into it
- when this buffer is full, pause the channel (that become the very last resort) till the expected event happen
- or, if we don't want to keep processing data into it, suspend filling it (it will still batch some into it for some time, TBH, but will eventually stop)

This is exactly what Vertx does too (see https://github.com/eclipse-vertx/vert.x/blob/a4d85b9ee33d41a21ec5d0f268b4fa12242d3bc0/src/main/java/io/vertx/core/net/impl/NetSocketImpl.java#L330 for more info how)



sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [14 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![6.3%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/10-16px.png '6.3%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [6.3% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [37 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![4.8%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/5-16px.png '4.8%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [4.8% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
k-wall: @robobario  @SamBarker  please take a look at the latest commit.   What do you think of this API?

I'm planning to take a look at `io.kroxylicious.proxy.filter.FilterResultBuilder#withDropMessage` tomorrow.

I also want to understand what the semantics of:

`return filterContext.requestFilterResultBuilder().completedFilterResult()`

ought to mean?

I think that should be short-hand for "forward the pipeline's existing request/header, with any mutations that I've made to the objects". IOW, the same as:

`return filterContext.requestFilterResultBuilder().withHeader(header).withMessage(request).completedFilterResult()`

WDYT?




robobario: > @robobario @SamBarker please take a look at the latest commit. What do you think of this API?
> 
> I'm planning to take a look at `io.kroxylicious.proxy.filter.FilterResultBuilder#withDropMessage` tomorrow.
> 
> I also want to understand what the semantics of:
> 
> `return filterContext.requestFilterResultBuilder().completedFilterResult()`
> 
> ought to mean?
> 
> I think that should be short-hand for "forward the pipeline's existing request/header, with any mutations that I've made to the objects". IOW, the same as:
> 
> `return filterContext.requestFilterResultBuilder().withHeader(header).withMessage(request).completedFilterResult()`
> 
> WDYT?

I don't know that we should give users these implicit shorthand for the same reason we only have the one `forwardRequest` method on `KrpcFilterContext`. If we make them supply headers and message we will have less ways to do the same thing and has some symmetry with the parameters arriving in the methods the Filter Author is implementing. I think it's more surprising if `completedFilterResult` can act implicitly on the objects passed in as params.

One thing I was hoping was that the fluent API would push them into valid constellations, so rather than having to validate that they haven't supplied headers along with a short-circuit response, we could arrange things so that there is no way to supply headers.

For example (on a branch where I've added in the parameters for RequestFilterResult and KrpcFilterContext):
1. forward a header and message
```
@Override
    public CompletionStage<RequestFilterResult<ProduceRequestData>> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request,
                                                                                     KrpcFilterContext<ProduceRequestData> context) {
        return context.requestFilterResultBuilder2().forward(header, request).build().asCompletedStage();
    }
```
2. forward a short-circuit response
```
@Override
    public CompletionStage<RequestFilterResult<ProduceRequestData>> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request,
                                                                                     KrpcFilterContext<ProduceRequestData> context) {
        return context.requestFilterResultBuilder2().shortCircuitResponse(header, request).build().asCompletedStage();
    }
```
3. drop
```
@Override
    public CompletionStage<RequestFilterResult<ProduceRequestData>> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request,
                                                                                     KrpcFilterContext<ProduceRequestData> context) {
        return context.requestFilterResultBuilder2().drop().build().asCompletedStage();
    }
```
Where any of those options can be combined with `closeConnection` like.
@Override
    public CompletionStage<RequestFilterResult<ProduceRequestData>> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request,
                                                                                     KrpcFilterContext<ProduceRequestData> context) {
        return context.requestFilterResultBuilder2().forward(header, request).withCloseConnection(true).build().asCompletedStage();
    }
robobario: FYI there's this PR from last week, I should have mentioned https://github.com/k-wall/kroxylicious/pull/4 which is an implementation of the queueing logic, plus integration test.

As you said, maybe better to get the API changes decided and merged first, but it's there for later.
k-wall: > FYI there's this PR from last week, I should have mentioned [k-wall#4](https://github.com/k-wall/kroxylicious/pull/4) which is an implementation of the queueing logic, plus integration test.
> 
> As you said, maybe better to get the API changes decided and merged first, but it's there for later.

Fantastic, thank you for looking at that.
k-wall: > we could arrange things so that there is no way to supply headers for a short-circuit response.

I think it is valid to supply response headers with a short circuit response.  That'd be the mechanism the user would use to set tagged values.

k-wall: > I was hoping was that the fluent API would push the Author to provide valid arrangements, 

@robobario thank you. You are right.  I'll iterate in that direction this morning
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [13 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![4.3%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/5-16px.png '4.3%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [4.3% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [13 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![4.3%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/5-16px.png '4.3%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [4.3% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
k-wall: > > I was hoping was that the fluent API would push the Author to provide valid arrangements,
> 
> @robobario thank you. You are right. I'll iterate in that direction this morning

@robobario Following your suggestions, I implemented as an opinionated builder that restricts what the user can do.  I agree it is much more pleasing API.

sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [13 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![4.1%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/5-16px.png '4.1%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [4.1% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [13 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![4.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/5-16px.png '4.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [4.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=494)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL) [13 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=494&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494) No Coverage information  
[![4.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/5-16px.png '4.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list) [4.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=494&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
k-wall: @robobario @SamBarker could you re-review.  I'm hoping that this API-scope change is nearing good shape.  It should maintain current behaviour.

My intent is that we merge this change, then work on getting the  request/response queuing/back pressure working  in separate PR(s).

robobario: Another half-baked idea for the future that might alleviate @SamBarker's doubts about this being a more complicated API. We could offer synchronous versions of all the interfaces as a convenience like:

``` java
class DoNothingFilter implements AsyncRequestFilter {
    @Override
    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey,
                                                          RequestHeaderData header,
                                                          ApiMessage body,
                                                          KrpcFilterContext filterContext) {
           return filterContext.forwardRequest(header, body).completed();
    }
}

class DoNothingFilter implements RequestFilter {
    @Override
    public RequestFilterResult onRequest(ApiKeys apiKey,
                                                          RequestHeaderData header,
                                                          ApiMessage body,
                                                          KrpcFilterContext filterContext) {
        return filterContext.forwardRequest(header, body);
    }
}
```
(where `forwardRequest` returns RequestFilterResult, rather than stage).
k-wall: @SamBarker are you okay to approve this PR. The nits will be squashed.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index acbf54fd..fae8b68b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,8 @@ 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
@@ -28,6 +30,11 @@ 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 b71b2545..c73f2b19 100644
--- a/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml
+++ b/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml
@@ -80,6 +80,11 @@
             <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 dafe9a99..9df54175 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,6 +6,7 @@
 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;
 
@@ -90,17 +91,19 @@ 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.
  */
@@ -132,78 +135,84 @@ public class MultiTenantTransformationFilter
     private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantTransformationFilter.class);
 
     @Override
-    public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request,
+                                                                      KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response,
+                                                                        KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onDeleteTopicsRequest(short apiVersion, RequestHeaderData header, DeleteTopicsRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> 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));
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onDeleteTopicsResponse(short apiVersion, ResponseHeaderData header, DeleteTopicsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onDeleteTopicsResponse(short apiVersion, ResponseHeaderData header, DeleteTopicsResponseData response,
+                                                                        KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onMetadataRequest(short apiVersion, RequestHeaderData header, MetadataRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> 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));
         }
-        context.forwardRequest(header, request);
-
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onMetadataResponse(short apiVersion, ResponseHeaderData header, MetadataResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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));
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> 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));
-        context.forwardRequest(header, request);
-
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
+                                                                   KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        context.forwardResponse(header, response);
-
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onListOffsetsRequest(short apiVersion, RequestHeaderData header, ListOffsetsRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onListOffsetsRequest(short apiVersion, RequestHeaderData header, ListOffsetsRequestData request,
+                                                                     KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onListOffsetsResponse(short apiVersion, ResponseHeaderData header, ListOffsetsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onListOffsetsResponse(short apiVersion, ResponseHeaderData header, ListOffsetsResponseData response,
+                                                                       KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onOffsetFetchRequest(short apiVersion, RequestHeaderData header, OffsetFetchRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> 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)));
@@ -213,140 +222,158 @@ public class MultiTenantTransformationFilter
                     .ifPresent(topics -> topics.forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false)));
         });
 
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onOffsetFetchResponse(short apiVersion, ResponseHeaderData header, OffsetFetchResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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));
         });
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onOffsetForLeaderEpochRequest(short apiVersion, RequestHeaderData header, OffsetForLeaderEpochRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onOffsetForLeaderEpochRequest(short apiVersion, RequestHeaderData header, OffsetForLeaderEpochRequestData request,
+                                                                              KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::topic, topic::setTopic, false));
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onOffsetForLeaderEpochResponse(short apiVersion, ResponseHeaderData header, OffsetForLeaderEpochResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onOffsetForLeaderEpochResponse(short apiVersion, ResponseHeaderData header, OffsetForLeaderEpochResponseData response,
+                                                                                KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::topic, topic::setTopic, false));
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onOffsetCommitRequest(short apiVersion, RequestHeaderData header, OffsetCommitRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> 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));
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onOffsetCommitResponse(short apiVersion, ResponseHeaderData header, OffsetCommitResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onOffsetCommitResponse(short apiVersion, ResponseHeaderData header, OffsetCommitResponseData response,
+                                                                        KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onOffsetDeleteRequest(short apiVersion, RequestHeaderData header, OffsetDeleteRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> 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));
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onOffsetDeleteResponse(short apiVersion, ResponseHeaderData header, OffsetDeleteResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onOffsetDeleteResponse(short apiVersion, ResponseHeaderData header, OffsetDeleteResponseData response,
+                                                                        KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::topic, topic::setTopic, topic.topicId() != null));
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::topic, topic::setTopic, topic.topicId() != null));
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onFindCoordinatorRequest(short apiVersion, RequestHeaderData header, FindCoordinatorRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> 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());
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onFindCoordinatorResponse(short apiVersion, ResponseHeaderData header, FindCoordinatorResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onFindCoordinatorResponse(short apiVersion, ResponseHeaderData header, FindCoordinatorResponseData response,
+                                                                           KrpcFilterContext context) {
         response.coordinators().forEach(coordinator -> removeTenantPrefix(context, coordinator::key, coordinator::setKey, false));
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onListGroupsResponse(short apiVersion, ResponseHeaderData header, ListGroupsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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);
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onJoinGroupRequest(short apiVersion, RequestHeaderData header, JoinGroupRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onJoinGroupRequest(short apiVersion, RequestHeaderData header, JoinGroupRequestData request,
+                                                                   KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setGroupId(tenantPrefix + request.groupId());
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onSyncGroupRequest(short apiVersion, RequestHeaderData header, SyncGroupRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onSyncGroupRequest(short apiVersion, RequestHeaderData header, SyncGroupRequestData request,
+                                                                   KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setGroupId(tenantPrefix + request.groupId());
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onLeaveGroupRequest(short apiVersion, RequestHeaderData header, LeaveGroupRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onLeaveGroupRequest(short apiVersion, RequestHeaderData header, LeaveGroupRequestData request,
+                                                                    KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setGroupId(tenantPrefix + request.groupId());
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onHeartbeatRequest(short apiVersion, RequestHeaderData header, HeartbeatRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onHeartbeatRequest(short apiVersion, RequestHeaderData header, HeartbeatRequestData request,
+                                                                   KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setGroupId(tenantPrefix + request.groupId());
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onDescribeGroupsRequest(short apiVersion, RequestHeaderData header, DescribeGroupsRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onDescribeGroupsRequest(short apiVersion, RequestHeaderData header, DescribeGroupsRequestData request,
+                                                                        KrpcFilterContext context) {
         request.setGroups(request.groups().stream().map(group -> applyTenantPrefix(context, group)).toList());
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onDescribeGroupsResponse(short apiVersion, ResponseHeaderData header, DescribeGroupsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onDescribeGroupsResponse(short apiVersion, ResponseHeaderData header, DescribeGroupsResponseData response,
+                                                                          KrpcFilterContext context) {
         response.groups().forEach(group -> removeTenantPrefix(context, group::groupId, group::setGroupId, false));
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onInitProducerIdRequest(short apiVersion, RequestHeaderData header, InitProducerIdRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onInitProducerIdRequest(short apiVersion, RequestHeaderData header, InitProducerIdRequestData request,
+                                                                        KrpcFilterContext context) {
         applyTenantPrefix(context, request::transactionalId, request::setTransactionalId, true);
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onAddPartitionsToTxnRequest(short apiVersion, RequestHeaderData header, AddPartitionsToTxnRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> 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);
 
@@ -356,11 +383,12 @@ public class MultiTenantTransformationFilter
                 applyTenantPrefix(context, addPartitionsToTxnTopic::name, addPartitionsToTxnTopic::setName, true);
             });
         });
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onAddPartitionsToTxnResponse(short apiVersion, ResponseHeaderData header, AddPartitionsToTxnResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onAddPartitionsToTxnResponse(short apiVersion, ResponseHeaderData header, AddPartitionsToTxnResponseData response,
+                                                                              KrpcFilterContext context) {
         response.resultsByTopicV3AndBelow().forEach(results -> removeTenantPrefix(context, results::name, results::setName, false));
 
         response.resultsByTransaction().forEach(addPartitionsToTxnResult -> {
@@ -369,61 +397,67 @@ public class MultiTenantTransformationFilter
                 removeTenantPrefix(context, topicResult::name, topicResult::setName, true);
             }
         });
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onAddOffsetsToTxnRequest(short apiVersion, RequestHeaderData header, AddOffsetsToTxnRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onAddOffsetsToTxnRequest(short apiVersion, RequestHeaderData header, AddOffsetsToTxnRequestData request,
+                                                                         KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setTransactionalId(tenantPrefix + request.transactionalId());
         request.setGroupId(tenantPrefix + request.groupId());
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onTxnOffsetCommitRequest(short apiVersion, RequestHeaderData header, TxnOffsetCommitRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> 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));
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onTxnOffsetCommitResponse(short apiVersion, ResponseHeaderData header, TxnOffsetCommitResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onTxnOffsetCommitResponse(short apiVersion, ResponseHeaderData header, TxnOffsetCommitResponseData response,
+                                                                           KrpcFilterContext context) {
         response.topics().forEach(results -> removeTenantPrefix(context, results::name, results::setName, false));
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onListTransactionsResponse(short apiVersion, ResponseHeaderData header, ListTransactionsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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);
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onDescribeTransactionsRequest(short apiVersion, RequestHeaderData header, DescribeTransactionsRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onDescribeTransactionsRequest(short apiVersion, RequestHeaderData header, DescribeTransactionsRequestData request,
+                                                                              KrpcFilterContext context) {
         request.setTransactionalIds(request.transactionalIds().stream().map(group -> applyTenantPrefix(context, group)).toList());
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onDescribeTransactionsResponse(short apiVersion, ResponseHeaderData header, DescribeTransactionsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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));
         });
-        context.forwardResponse(header, response);
+        return context.forwardResponse(header, response);
     }
 
     @Override
-    public void onEndTxnRequest(short apiVersion, RequestHeaderData header, EndTxnRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onEndTxnRequest(short apiVersion, RequestHeaderData header, EndTxnRequestData request, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setTransactionalId(tenantPrefix + request.transactionalId());
-        context.forwardRequest(header, request);
+        return 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 e98b4ac9..db941df4 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,6 +8,7 @@ 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;
@@ -20,6 +21,8 @@ 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;
@@ -62,21 +65,22 @@ public class ProduceValidationFilter implements ProduceRequestFilter, ProduceRes
     }
 
     @Override
-    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         ProduceRequestValidationResult result = validator.validateRequest(request);
         if (result.isAnyTopicPartitionInvalid()) {
-            handleInvalidTopicPartitions(header, request, context, result);
+            return handleInvalidTopicPartitions(header, request, context, result);
         }
         else {
-            context.forwardRequest(header, request);
+            return context.forwardRequest(header, request);
         }
     }
 
-    private void handleInvalidTopicPartitions(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context, ProduceRequestValidationResult result) {
+    private CompletionStage<RequestFilterResult> 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);
-            context.forwardResponse(response);
+            return context.requestFilterResultBuilder().shortCircuitResponse(response).completed();
         }
         // 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) {
@@ -86,13 +90,13 @@ public class ProduceValidationFilter implements ProduceRequestFilter, ProduceRes
                 topicDatum.partitionData().removeIf(partitionProduceData -> !result.isPartitionValid(topicDatum.name(), partitionProduceData.index()));
             }
             correlatedResults.put(header.correlationId(), result);
-            context.forwardRequest(header, request);
+            return 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);
-            context.forwardResponse(response);
+            return context.requestFilterResultBuilder().shortCircuitResponse(response).completed();
         }
     }
 
@@ -145,15 +149,16 @@ public class ProduceValidationFilter implements ProduceRequestFilter, ProduceRes
     }
 
     @Override
-    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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);
-            context.forwardResponse(response);
+            return context.forwardResponse(header, response);
         }
         else {
-            context.forwardResponse(response);
+            return context.forwardResponse(header, 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 21b6cbc6..a70cb447 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,6 +6,8 @@
 
 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;
@@ -15,7 +17,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 chainnwill be consulted (via their invoker) to see if any want to handle that message.
+ * filter chain will 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
@@ -72,34 +74,49 @@ 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>
-     * 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.
+     * 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.
      * </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 void onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-        filterContext.forwardRequest(header, body);
+    default CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body,
+                                                           KrpcFilterContext filterContext) {
+        return filterContext.requestFilterResultBuilder().forward(header, body).completed();
     }
 
     /**
      * <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>
-     * 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.
+     * 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.
      * </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 void onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-        filterContext.forwardResponse(header, body);
+    default CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body,
+                                                             KrpcFilterContext filterContext) {
+        return filterContext.responseFilterResultBuilder().forward(header, body).completed();
     }
 
 }
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
new file mode 100644
index 00000000..64170a13
--- /dev/null
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterResult.java
@@ -0,0 +1,45 @@
+/*
+ * 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
new file mode 100644
index 00000000..4b789e6f
--- /dev/null
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterResultBuilder.java
@@ -0,0 +1,42 @@
+/*
+ * 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 dc9c00c5..dbe5d6dc 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,6 +6,8 @@
 
 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;
@@ -15,12 +17,12 @@ record HandleNothingFilterInvoker() implements FilterInvoker {
     static final FilterInvoker INSTANCE = new HandleNothingFilterInvoker();
 
     @Override
-    public void onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         throw new IllegalStateException("I should never be invoked");
     }
 
     @Override
-    public void onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public CompletionStage<ResponseFilterResult> 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 18688b31..3cf00ea7 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,12 +38,32 @@ public interface KrpcFilterContext {
     String sniHostname();
 
     /**
-     * Send a request towards the broker, invoking upstream filters.
+     * 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()}
      *
      * @param header The header to forward to the broker.
      * @param request The request to forward to the broker.
+     * @return completed filter results.
      */
-    void forwardRequest(RequestHeaderData header, ApiMessage request);
+    CompletionStage<RequestFilterResult> forwardRequest(RequestHeaderData header, ApiMessage request);
 
     /**
      * Send a message from a filter towards the broker, invoking upstream filters
@@ -60,35 +80,32 @@ public interface KrpcFilterContext {
     <T extends ApiMessage> CompletionStage<T> sendRequest(short apiVersion, ApiMessage request);
 
     /**
-     * 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.
-     */
-    void forwardResponse(ResponseHeaderData header, ApiMessage response);
-
-    /**
-     * 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.
+     * 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.
      */
-    void forwardResponse(ApiMessage response);
+    CompletionStage<ResponseFilterResult> forwardResponse(ResponseHeaderData header, 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.
+     * 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
      */
-    void closeConnection();
+    ResponseFilterResultBuilder responseFilterResultBuilder();
 
     // 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 7dc78aea..c761a75e 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,6 +6,8 @@
 
 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;
@@ -35,13 +37,16 @@ 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.
      */
-    void onRequest(ApiKeys apiKey,
-                   RequestHeaderData header,
-                   ApiMessage body,
-                   KrpcFilterContext filterContext);
+    CompletionStage<RequestFilterResult> 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 ba1d5f5c..20f1a969 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,6 +6,8 @@
 
 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;
@@ -13,8 +15,8 @@ import org.apache.kafka.common.protocol.ApiMessage;
 record RequestFilterInvoker(RequestFilter filter) implements FilterInvoker {
 
     @Override
-    public void onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-        filter.onRequest(apiKey, header, body, filterContext);
+    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        return 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
new file mode 100644
index 00000000..5db3d065
--- /dev/null
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterResult.java
@@ -0,0 +1,24 @@
+/*
+ * 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
new file mode 100644
index 00000000..4ad85930
--- /dev/null
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterResultBuilder.java
@@ -0,0 +1,43 @@
+/*
+ * 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 1338f604..4acd982e 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,6 +6,8 @@
 
 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;
@@ -14,13 +16,13 @@ import org.apache.kafka.common.protocol.ApiMessage;
 record RequestResponseInvoker(RequestFilter requestFilter, ResponseFilter responseFilter) implements FilterInvoker {
 
     @Override
-    public void onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-        requestFilter.onRequest(apiKey, header, body, filterContext);
+    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        return requestFilter.onRequest(apiKey, header, body, filterContext);
     }
 
     @Override
-    public void onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-        responseFilter.onResponse(apiKey, header, body, filterContext);
+    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        return 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 e5ffaea9..dffd64e4 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,6 +6,8 @@
 
 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;
@@ -39,9 +41,11 @@ 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.
      */
-    void onResponse(ApiKeys apiKey,
-                    ResponseHeaderData header,
-                    ApiMessage body,
-                    KrpcFilterContext filterContext);
+    CompletionStage<ResponseFilterResult> 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 c0bc7746..1d6d16cd 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,6 +6,8 @@
 
 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;
@@ -13,8 +15,8 @@ import org.apache.kafka.common.protocol.ApiMessage;
 record ResponseFilterInvoker(ResponseFilter filter) implements FilterInvoker {
 
     @Override
-    public void onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-        filter.onResponse(apiKey, header, body, filterContext);
+    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        return 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
new file mode 100644
index 00000000..6fc44128
--- /dev/null
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterResult.java
@@ -0,0 +1,14 @@
+/*
+ * 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
new file mode 100644
index 00000000..3d8930f3
--- /dev/null
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterResultBuilder.java
@@ -0,0 +1,15 @@
+/*
+ * 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 209c4a2f..2be85d13 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,6 +6,8 @@
 
 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;
@@ -20,22 +22,22 @@ import org.apache.kafka.common.protocol.ApiMessage;
 record SafeInvoker(FilterInvoker invoker) implements FilterInvoker {
 
     @Override
-    public void onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, short apiVersion, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         if (invoker.shouldHandleRequest(apiKey, apiVersion)) {
-            invoker.onRequest(apiKey, apiVersion, header, body, filterContext);
+            return invoker.onRequest(apiKey, apiVersion, header, body, filterContext);
         }
         else {
-            filterContext.forwardRequest(header, body);
+            return filterContext.forwardRequest(header, body);
         }
     }
 
     @Override
-    public void onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, short apiVersion, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         if (invoker.shouldHandleResponse(apiKey, apiVersion)) {
-            invoker.onResponse(apiKey, apiVersion, header, body, filterContext);
+            return invoker.onResponse(apiKey, apiVersion, header, body, filterContext);
         }
         else {
-            filterContext.forwardResponse(header, body);
+            return filterContext.responseFilterResultBuilder().forward(header, body).completed();
         }
     }
 
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
new file mode 100644
index 00000000..873714db
--- /dev/null
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/filterresultbuilder/CloseOrTerminalStage.java
@@ -0,0 +1,17 @@
+/*
+ * 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
new file mode 100644
index 00000000..acaa9e07
--- /dev/null
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/filterresultbuilder/CloseStage.java
@@ -0,0 +1,23 @@
+/*
+ * 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
new file mode 100644
index 00000000..4603d99e
--- /dev/null
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/filterresultbuilder/TerminalStage.java
@@ -0,0 +1,31 @@
+/*
+ * 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 a2a20feb..e20b57f4 100644
--- a/kroxylicious-api/src/main/templates/Kproxy/Filter.ftl
+++ b/kroxylicious-api/src/main/templates/Kproxy/Filter.ftl
@@ -8,6 +8,8 @@
 <#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
 />
 /*
@@ -28,6 +30,8 @@
  */
 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;
@@ -55,7 +59,7 @@ public interface ${filterClass} extends KrpcFilter {
 
     /**
      * Handle the given {@code ${msgType}},
-     * returning the {@code ${messageSpec.name}Data} instance to be passed to the next filter.
+     * returning the {@code ${dataClass}} 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.
      *
@@ -63,7 +67,9 @@ 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.
      */
-    void on${messageSpec.name}(short apiVersion, <#if messageSpec.type?lower_case == 'response'>ResponseHeaderData<#else>RequestHeaderData</#if> header, ${dataClass} ${msgType}, KrpcFilterContext context);
+     CompletionStage<${filterResultClass}> on${messageSpec.name}(short apiVersion, ${headerClass} 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 7bfd81fb..18db9847 100644
--- a/kroxylicious-api/src/main/templates/Kproxy/FilterInvoker.ftl
+++ b/kroxylicious-api/src/main/templates/Kproxy/FilterInvoker.ftl
@@ -8,6 +8,8 @@
 <#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
 />
@@ -29,6 +31,8 @@
  */
 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'>
@@ -58,7 +62,7 @@ class ${filterInvokerClass} implements FilterInvoker {
     }
 
     @Override
-    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);
+    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);
     }
 }
\ 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 6ac5be2c..4056d7df 100644
--- a/kroxylicious-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl
+++ b/kroxylicious-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl
@@ -23,6 +23,8 @@
  */
 package ${outputPackage};
 
+import java.util.concurrent.CompletionStage;
+
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
@@ -67,14 +69,14 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
      */
     @Override
     @SuppressWarnings("SwitchStatementWithTooFewBranches")
-    public void onRequest(ApiKeys apiKey,
-                           short apiVersion,
-                           RequestHeaderData header,
-                           ApiMessage body,
-                           KrpcFilterContext filterContext) {
+    public CompletionStage<RequestFilterResult> 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
-        switch (apiKey) {
+        return switch (apiKey) {
 <#list messageSpecs as messageSpec>
     <#if messageSpec.type?lower_case == 'request'>
             case ${retrieveApiKey(messageSpec)} ->
@@ -82,7 +84,7 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
     </#if>
 </#list>
             default -> throw new IllegalStateException("Unsupported RPC " + apiKey);
-        }
+        };
 
     }
 
@@ -96,14 +98,14 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
      */
     @Override
     @SuppressWarnings("SwitchStatementWithTooFewBranches")
-    public void onResponse(ApiKeys apiKey,
-                            short apiVersion,
-                            ResponseHeaderData header,
-                            ApiMessage body,
-                            KrpcFilterContext filterContext) {
+    public CompletionStage<ResponseFilterResult> 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
-        switch (apiKey) {
+        return switch (apiKey) {
 <#list messageSpecs as messageSpec>
     <#if messageSpec.type?lower_case == 'response'>
             case ${retrieveApiKey(messageSpec)} ->
@@ -111,7 +113,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 23aac130..523cced8 100644
--- a/kroxylicious-sample/pom.xml
+++ b/kroxylicious-sample/pom.xml
@@ -61,6 +61,11 @@
             <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 4364dd64..e38b618a 100644
--- a/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleFetchResponseFilter.java
+++ b/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleFetchResponseFilter.java
@@ -6,6 +6,8 @@
 
 package io.kroxylicious.sample;
 
+import java.util.concurrent.CompletionStage;
+
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 
@@ -14,6 +16,7 @@ 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;
 
@@ -48,15 +51,17 @@ 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.
+     * @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.
      */
     @Override
-    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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
-        context.forwardResponse(header, response);
+        return 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 f80f9bf9..1b54800a 100644
--- a/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleProduceRequestFilter.java
+++ b/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleProduceRequestFilter.java
@@ -6,6 +6,8 @@
 
 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;
@@ -15,6 +17,7 @@ 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;
 
@@ -49,15 +52,18 @@ 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.
+     * @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.
      */
     @Override
-    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> 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
-        context.forwardRequest(header, request);
+
+        return 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 360b8b06..8bad87a5 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 3fef507b..7ceb28f0 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
@@ -25,10 +25,16 @@ 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;
 
 /**
@@ -93,8 +99,7 @@ class DefaultFilterContext implements KrpcFilterContext {
      * @param header The header
      * @param message The message
      */
-    @Override
-    public void forwardRequest(RequestHeaderData header, ApiMessage message) {
+    protected void forwardRequestInternal(RequestHeaderData header, ApiMessage message) {
         if (decodedFrame.body() != message) {
             throw new IllegalStateException();
         }
@@ -167,6 +172,11 @@ 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).
@@ -174,8 +184,7 @@ class DefaultFilterContext implements KrpcFilterContext {
      * @param header The header
      * @param response The message
      */
-    @Override
-    public void forwardResponse(ResponseHeaderData header, ApiMessage response) {
+    protected void forwardResponseInternal(ResponseHeaderData header, ApiMessage response) {
         // check it's a response
         String name = response.getClass().getName();
         if (!name.endsWith("ResponseData")) {
@@ -200,17 +209,21 @@ class DefaultFilterContext implements KrpcFilterContext {
     }
 
     @Override
-    public void forwardResponse(ApiMessage response) {
-        if (decodedFrame instanceof RequestFrame) {
-            this.forwardResponse(new ResponseHeaderData().setCorrelationId(decodedFrame.correlationId()), response);
-        }
-        else {
-            this.forwardResponse((ResponseHeaderData) decodedFrame.header(), response);
-        }
+    public ResponseFilterResultBuilder responseFilterResultBuilder() {
+        return new ResponseFilterResultBuilderImpl();
     }
 
     @Override
-    public void closeConnection() {
+    public CompletionStage<RequestFilterResult> forwardRequest(RequestHeaderData header, ApiMessage request) {
+        return requestFilterResultBuilder().forward(header, request).completed();
+    }
+
+    @Override
+    public RequestFilterResultBuilder requestFilterResultBuilder() {
+        return new RequestFilterResultBuilderImpl();
+    }
+
+    protected 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 b881a4b5..bed0a1f1 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
@@ -8,6 +8,8 @@ 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;
@@ -30,8 +32,7 @@ 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;
@@ -58,7 +59,49 @@ public class FilterHandler
                 LOGGER.debug("{}: Dispatching downstream {} request to filter{}: {}",
                         ctx.channel(), decodedFrame.apiKey(), filterDescriptor(), msg);
             }
-            invoker.onRequest(decodedFrame.apiKey(), decodedFrame.apiVersion(), decodedFrame.header(), decodedFrame.body(), filterContext);
+
+            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();
+                }
+
+            });
 
         }
         else {
@@ -89,7 +132,38 @@ public class FilterHandler
                     LOGGER.debug("{}: Dispatching upstream {} response to filter {}: {}",
                             ctx.channel(), decodedFrame.apiKey(), filterDescriptor(), msg);
                 }
-                invoker.onResponse(decodedFrame.apiKey(), decodedFrame.apiVersion(), decodedFrame.header(), decodedFrame.body(), filterContext);
+                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();
+                    }
+
+                });
+
             }
         }
         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 ca4957be..8b2ab21d 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,6 +7,7 @@ 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;
@@ -16,6 +17,7 @@ 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
@@ -72,8 +74,9 @@ public class ApiVersionsFilter implements ApiVersionsResponseFilter {
     }
 
     @Override
-    public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData data, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData data,
+                                                                       KrpcFilterContext context) {
         intersectApiVersions(context.channelDescriptor(), data);
-        context.forwardResponse(header, data);
+        return 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 a5e850f8..178948a1 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,6 +7,7 @@ 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;
@@ -27,6 +28,7 @@ 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;
@@ -48,29 +50,31 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
     }
 
     @Override
-    public void onMetadataResponse(short apiVersion, ResponseHeaderData header, MetadataResponseData data, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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);
         }
-        doReconcileThenForwardResponse(header, data, context, nodeMap);
+        return doReconcileThenForwardResponse(header, data, context, nodeMap);
     }
 
     @Override
-    public void onDescribeClusterResponse(short apiVersion, ResponseHeaderData header, DescribeClusterResponseData data, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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);
         }
-        doReconcileThenForwardResponse(header, data, context, nodeMap);
+        return doReconcileThenForwardResponse(header, data, context, nodeMap);
     }
 
     @Override
-    public void onFindCoordinatorResponse(short apiVersion, ResponseHeaderData header, FindCoordinatorResponseData data, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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.
@@ -83,7 +87,7 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
             apply(context, data, FindCoordinatorResponseData::nodeId, FindCoordinatorResponseData::host, FindCoordinatorResponseData::port,
                     FindCoordinatorResponseData::setHost, FindCoordinatorResponseData::setPort);
         }
-        context.forwardResponse(header, data);
+        return context.responseFilterResultBuilder().forward(header, data).completed();
     }
 
     private <T> void apply(KrpcFilterContext context, T broker, Function<T, Integer> nodeIdGetter, Function<T, String> hostGetter, ToIntFunction<T> portGetter,
@@ -99,11 +103,12 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
         portSetter.accept(broker, downstreamAddress.port());
     }
 
-    private void doReconcileThenForwardResponse(ResponseHeaderData header, ApiMessage data, KrpcFilterContext context, Map<Integer, HostPort> nodeMap) {
+    private CompletionStage<ResponseFilterResult> 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();
-        context.forwardResponse(header, data);
-        LOGGER.debug("Endpoint reconciliation complete for  virtual cluster {}", virtualCluster);
+        LOGGER.debug("Endpoint reconciliation complete for virtual cluster {}", virtualCluster);
+        return context.responseFilterResultBuilder().forward(header, data).completed();
     }
 }
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 2de82316..39182460 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,6 +6,8 @@
 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;
@@ -17,6 +19,7 @@ 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
@@ -42,9 +45,9 @@ public class EagerMetadataLearner implements RequestFilter {
     }
 
     @Override
-    public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         if (KAFKA_PRELUDE.contains(apiKey)) {
-            filterContext.forwardRequest(header, body);
+            return filterContext.requestFilterResultBuilder().forward(header, body).completed();
         }
         else {
             final short apiVersion = determineMetadataApiVersion(header);
@@ -53,19 +56,25 @@ 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 -> {
-                        if (useClientRequest) {
-                            // The client's requested matched our out-of-band request, so we may as well return the
-                            // response.
-                            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();
+                        var builder = filterContext.requestFilterResultBuilder();
+                        if (useClientRequest) {
+                            // The client's request matched our out-of-band message, so we may as well return the
+                            // response.
+                            future.complete(builder.shortCircuitResponse(metadataResponseData).withCloseConnection().build());
+                        }
+                        else {
+                            future.complete(builder.withCloseConnection().build());
+
+                        }
                         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 2788984e..18123208 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,6 +9,8 @@ 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;
@@ -31,6 +33,7 @@ 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;
 
 /**
@@ -71,7 +74,8 @@ public class FetchResponseTransformationFilter implements FetchResponseFilter {
     }
 
     @Override
-    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData fetchResponse, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData fetchResponse,
+                                                                 KrpcFilterContext context) {
         List<MetadataRequestData.MetadataRequestTopic> requestTopics = fetchResponse.responses().stream()
                 .filter(t -> t.topic().isEmpty())
                 .map(fetchableTopicResponse -> {
@@ -82,24 +86,29 @@ 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(ti -> ti.topicId(), ti -> ti.name()));
+                        Map<Uuid, String> uidToName = metadataResponse.topics().stream().collect(Collectors.toMap(MetadataResponseData.MetadataResponseTopic::topicId,
+                                MetadataResponseData.MetadataResponseTopic::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");
-                        context.forwardResponse(header, fetchResponse);
+
+                        var value = context.responseFilterResultBuilder().forward(header, fetchResponse).build();
+                        future.complete(value);
                     });
+            return future;
         }
         else {
             applyTransformation(context, fetchResponse);
-            context.forwardResponse(header, fetchResponse);
+            return 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
new file mode 100644
index 00000000..ae944727
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FilterResultBuilderImpl.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.internal.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 6309aeeb..8015a655 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,6 +9,7 @@ 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;
@@ -26,6 +27,7 @@ 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;
 
 /**
@@ -72,9 +74,9 @@ public class ProduceRequestTransformationFilter implements ProduceRequestFilter
     }
 
     @Override
-    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData data, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData data, KrpcFilterContext context) {
         applyTransformation(context, data);
-        context.forwardRequest(header, data);
+        return 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
new file mode 100644
index 00000000..044c8dca
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/RequestFilterResultBuilderImpl.java
@@ -0,0 +1,93 @@
+/*
+ * 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
new file mode 100644
index 00000000..57112c47
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ResponseFilterResultBuilderImpl.java
@@ -0,0 +1,53 @@
+/*
+ * 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();
+            }
+        };
+    }
+
+}


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 441525fd..7178dfa5 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
@@ -242,7 +242,6 @@ 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 c8261875..20fb6530 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CompositePrefixingFixedClientIdFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CompositePrefixingFixedClientIdFilter.java
@@ -7,6 +7,7 @@
 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;
@@ -27,9 +28,9 @@ public class CompositePrefixingFixedClientIdFilter implements CompositeFilter {
 
     private class PrefixingFilter implements RequestFilter {
         @Override
-        public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
             header.setClientId(config.prefix + header.clientId());
-            filterContext.forwardRequest(header, body);
+            return 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 8a2283cc..4b040a50 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CreateTopicRejectFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CreateTopicRejectFilter.java
@@ -6,6 +6,8 @@
 
 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;
@@ -16,7 +18,8 @@ public class CreateTopicRejectFilter implements CreateTopicsRequestFilter {
     public static final String ERROR_MESSAGE = "rejecting all topics";
 
     @Override
-    public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request,
+                                                                      KrpcFilterContext context) {
         CreateTopicsResponseData response = new CreateTopicsResponseData();
         CreateTopicsResponseData.CreatableTopicResultCollection topics = new CreateTopicsResponseData.CreatableTopicResultCollection();
         allocateByteBufToTestKroxyliciousReleasesIt(context);
@@ -27,7 +30,7 @@ public class CreateTopicRejectFilter implements CreateTopicsRequestFilter {
             topics.add(result);
         });
         response.setTopics(topics);
-        context.forwardResponse(response);
+        return context.requestFilterResultBuilder().shortCircuitResponse(response).completed();
     }
 
     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 d9361025..7b76df75 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/FixedClientIdFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/FixedClientIdFilter.java
@@ -6,6 +6,8 @@
 
 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;
@@ -35,23 +37,13 @@ public class FixedClientIdFilter implements RequestFilter, ResponseFilter {
     }
 
     @Override
-    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) {
+    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         header.setClientId(clientId);
-        filterContext.forwardRequest(header, body);
+        return filterContext.forwardRequest(header, body);
     }
 
     @Override
-    public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-        filterContext.forwardResponse(header, body);
+    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        return 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 d599ef74..2f00b31e 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/OutOfBandSendFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/OutOfBandSendFilter.java
@@ -7,6 +7,7 @@
 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;
@@ -55,21 +56,23 @@ public class OutOfBandSendFilter implements DescribeClusterRequestFilter, Descri
     }
 
     @Override
-    public void onDescribeClusterRequest(short apiVersion, RequestHeaderData header, DescribeClusterRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> 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(","));
         });
-        context.forwardRequest(header, request);
+        return context.forwardRequest(header, request);
     }
 
     @Override
-    public void onDescribeClusterResponse(short apiVersion, ResponseHeaderData header, DescribeClusterResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onDescribeClusterResponse(short apiVersion, ResponseHeaderData header, DescribeClusterResponseData response,
+                                                                           KrpcFilterContext context) {
         response.setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code())
                 .setErrorMessage("filterNameTaggedFieldsFromOutOfBandResponse: " + values);
-        context.forwardResponse(response);
+        return context.forwardResponse(header, 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 bffd8947..9ed8696b 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
@@ -7,6 +7,7 @@
 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;
@@ -41,19 +42,19 @@ public class RequestResponseMarkingFilter implements RequestFilter, ResponseFilt
     }
 
     @Override
-    public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         if (keysToMark.contains(apiKey)) {
             body.unknownTaggedFields().add(createTaggedField("request"));
         }
-        filterContext.forwardRequest(header, body);
+        return filterContext.forwardRequest(header, body);
     }
 
     @Override
-    public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
         if (keysToMark.contains(apiKey)) {
             body.unknownTaggedFields().add(createTaggedField("response"));
         }
-        filterContext.forwardResponse(header, body);
+        return 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 996ec724..d351cb72 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,6 +8,7 @@ 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;
 
@@ -16,11 +17,14 @@ 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.junit.jupiter.MockitoExtension;
 
 import com.flipkart.zjsonpatch.JsonDiff;
 import com.google.common.reflect.ClassPath;
@@ -29,6 +33,8 @@ 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;
 
@@ -36,15 +42,16 @@ 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.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
+import static org.mockito.Mock.Strictness.LENIENT;
 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()
@@ -62,14 +69,21 @@ class MultiTenantTransformationFilterTest {
 
     private final FilterInvoker invoker = getOnlyElement(FilterAndInvoker.build(filter)).invoker();
 
-    private final KrpcFilterContext context = mock(KrpcFilterContext.class);
+    @Mock(strictness = LENIENT)
+    private KrpcFilterContext context;
 
-    private final ArgumentCaptor<ApiMessage> apiMessageCaptor = ArgumentCaptor.forClass(ApiMessage.class);
+    @Captor
+    private ArgumentCaptor<RequestHeaderData> requestHeaderDataCaptor;
 
-    @BeforeEach
-    public void beforeEach() {
-        when(context.sniHostname()).thenReturn("tenant1.kafka.example.com");
-    }
+    @Mock(strictness = LENIENT)
+    private RequestFilterResult requestFilterResult;
+    @Captor
+    private ArgumentCaptor<ResponseHeaderData> responseHeaderDataArgumentCaptor;
+    @Mock(strictness = LENIENT)
+    private ResponseFilterResult responseFilterResult;
+
+    @Captor
+    private ArgumentCaptor<ApiMessage> apiMessageCaptor;
 
     public static Stream<Arguments> requests() throws Exception {
         return RequestResponseTestDef.requestResponseTestDefinitions(getTestResources()).filter(td -> td.request() != null)
@@ -78,16 +92,24 @@ class MultiTenantTransformationFilterTest {
 
     @ParameterizedTest(name = "{0}")
     @MethodSource(value = "requests")
-    void requestsTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef requestTestDef) {
+    void requestsTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef requestTestDef)
+            throws Exception {
         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());
 
-        invoker.onRequest(ApiKeys.forId(apiMessageType.apiKey()), header.requestApiVersion(), header, request, context);
-        verify(context).forwardRequest(any(), apiMessageCaptor.capture());
+        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));
 
-        var filtered = requestWriter.apply(apiMessageCaptor.getValue(), header.requestApiVersion());
+        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());
         assertEquals(requestTestDef.expectedPatch(), JsonDiff.asJson(marshalled, filtered));
     }
 
@@ -98,18 +120,26 @@ class MultiTenantTransformationFilterTest {
 
     @ParameterizedTest(name = "{0}")
     @MethodSource(value = "responses")
-    void responseTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef responseTestDef) {
+    void responseTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef responseTestDef)
+            throws Exception {
         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();
-        invoker.onResponse(ApiKeys.forId(apiMessageType.apiKey()), header.requestApiVersion(), headerData, response, context);
-        verify(context).forwardResponse(any(), apiMessageCaptor.capture());
+        var stage = invoker.onResponse(ApiKeys.forId(apiMessageType.apiKey()), header.requestApiVersion(), headerData, response, context);
+        assertThat(stage).isCompleted();
+        var forward = stage.toCompletableFuture().get().message();
 
-        var filtered = responseWriter.apply(apiMessageCaptor.getValue(), header.requestApiVersion());
+        var filtered = responseWriter.apply(forward, 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 e2559c6d..2152785f 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,6 +7,7 @@
 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;
@@ -40,12 +41,8 @@ class FilterInvokersTest {
     public void testCompositeFilter() {
         MultipleSpecificFilter filterA = new MultipleSpecificFilter();
         RequestResponseFilter filterB = new RequestResponseFilter();
-        RequestFilter filterC = (apiKey, header, body, filterContext) -> {
-
-        };
-        ResponseFilter filterD = (apiKey, header, body, filterContext) -> {
-
-        };
+        RequestFilter filterC = (apiKey, header, body, filterContext) -> null;
+        ResponseFilter filterD = (apiKey, header, body, filterContext) -> null;
         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();
@@ -67,15 +64,9 @@ class FilterInvokersTest {
     }
 
     public static Stream<KrpcFilter> validFilters() {
-        ApiVersionsRequestFilter singleSpecificMessageFilter = (apiVersion, header, request, context) -> {
-
-        };
-        RequestFilter requestFilter = (apiKey, header, body, filterContext) -> {
-
-        };
-        ResponseFilter responseFilter = (apiKey, header, body, filterContext) -> {
-
-        };
+        ApiVersionsRequestFilter singleSpecificMessageFilter = (apiVersion, header, request, context) -> null;
+        RequestFilter requestFilter = (apiKey, header, body, filterContext) -> null;
+        ResponseFilter responseFilter = (apiKey, header, body, filterContext) -> null;
         return Stream.of(singleSpecificMessageFilter,
                 new MultipleSpecificFilter(),
                 requestFilter,
@@ -88,13 +79,13 @@ class FilterInvokersTest {
     static class RequestResponseFilter implements RequestFilter, ResponseFilter {
 
         @Override
-        public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-
+        public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+            return null;
         }
 
         @Override
-        public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
-
+        public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+            return null;
         }
     }
 
@@ -117,13 +108,17 @@ class FilterInvokersTest {
     static class MultipleSpecificFilter implements ApiVersionsRequestFilter, ApiVersionsResponseFilter {
 
         @Override
-        public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+        public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
+                                                                         KrpcFilterContext context) {
 
+            return null;
         }
 
         @Override
-        public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
+        public CompletionStage<ResponseFilterResult> onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response,
+                                                                           KrpcFilterContext context) {
 
+            return null;
         }
     }
 
@@ -134,8 +129,9 @@ class FilterInvokersTest {
         }
 
         @Override
-        public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
 
+            return null;
         }
     }
 
@@ -152,9 +148,7 @@ 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) -> {
-
-            })));
+            return List.of((CompositeFilter) () -> List.of((CompositeFilter) () -> List.of((ApiVersionsRequestFilter) (apiVersion, header, request, context) -> null)));
         }
     }
 
@@ -165,16 +159,19 @@ class FilterInvokersTest {
         }
 
         @Override
-        public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
 
+            return null;
         }
     }
 
     static class SpecificAndCompositeFilter implements ApiVersionsRequestFilter, CompositeFilter {
 
         @Override
-        public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+        public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
+                                                                         KrpcFilterContext context) {
 
+            return null;
         }
 
         @Override
@@ -186,26 +183,32 @@ class FilterInvokersTest {
     static class SpecificAndResponseFilter implements ApiVersionsRequestFilter, ResponseFilter {
 
         @Override
-        public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+        public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
+                                                                         KrpcFilterContext context) {
 
+            return null;
         }
 
         @Override
-        public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
 
+            return null;
         }
     }
 
     static class SpecificAndRequestFilter implements ApiVersionsRequestFilter, RequestFilter {
 
         @Override
-        public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
 
+            return null;
         }
 
         @Override
-        public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+        public CompletionStage<RequestFilterResult> 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 56e34e04..611d475a 100644
--- a/kroxylicious-sample/src/test/java/io/kroxylicious/sample/SampleFetchResponseFilterTest.java
+++ b/kroxylicious-sample/src/test/java/io/kroxylicious/sample/SampleFetchResponseFilterTest.java
@@ -10,6 +10,7 @@ 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;
@@ -24,19 +25,21 @@ 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
@@ -45,12 +48,19 @@ 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;
+
     @Captor
-    private ArgumentCaptor<ApiMessage> apiMessageCaptor = ArgumentCaptor.forClass(ApiMessage.class);
+    private ArgumentCaptor<ResponseHeaderData> responseHeaderDataCaptor;
 
+    @Captor
+    private ArgumentCaptor<ApiMessage> apiMessageCaptor;
     private SampleFetchResponseFilter filter;
     private ResponseHeaderData headerData;
 
@@ -66,11 +76,12 @@ class SampleFetchResponseFilterTest {
      * Unit Test: Checks that transformation is applied when response data contains configured value.
      */
     @Test
-    public void willTransformFetchResponseTest() {
+    public void willTransformFetchResponseTest() throws Exception {
         var responseData = buildFetchResponseData(PRE_TRANSFORM_VALUE);
-        filter.onFetchResponse(API_VERSION, headerData, responseData, context);
-        verify(context).forwardResponse(any(), apiMessageCaptor.capture());
-        var unpackedResponse = unpackFetchResponseData((FetchResponseData) apiMessageCaptor.getValue());
+        var stage = filter.onFetchResponse(API_VERSION, headerData, responseData, context);
+
+        var response = stage.toCompletableFuture().get().message();
+        var unpackedResponse = unpackFetchResponseData(((FetchResponseData) response));
         // 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
@@ -84,21 +95,24 @@ class SampleFetchResponseFilterTest {
      * value.
      */
     @Test
-    public void wontTransformFetchResponseTest() {
+    public void wontTransformFetchResponseTest() throws Exception {
         var responseData = buildFetchResponseData(NO_TRANSFORM_VALUE);
-        filter.onFetchResponse(API_VERSION, headerData, responseData, context);
-        verify(context).forwardResponse(any(), apiMessageCaptor.capture());
-        var unpackedResponse = unpackFetchResponseData((FetchResponseData) apiMessageCaptor.getValue());
+        var stage = filter.onFetchResponse(API_VERSION, headerData, responseData, context);
+
+        var response = stage.toCompletableFuture().get().message();
+        var unpackedResponse = unpackFetchResponseData((FetchResponseData) response);
         // 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() {
-        context = mock(KrpcFilterContext.class);
-        // create stub for createByteBufferOutputStream method
-        ArgumentCaptor<Integer> argument = ArgumentCaptor.forClass(Integer.class);
-        when(context.createByteBufferOutputStream(argument.capture())).thenAnswer(
+        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(
                 (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 2ce3840e..4e538cef 100644
--- a/kroxylicious-sample/src/test/java/io/kroxylicious/sample/SampleProduceRequestFilterTest.java
+++ b/kroxylicious-sample/src/test/java/io/kroxylicious/sample/SampleProduceRequestFilterTest.java
@@ -10,6 +10,7 @@ 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;
@@ -24,19 +25,21 @@ 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
@@ -46,10 +49,18 @@ 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<ApiMessage> apiMessageCaptor = ArgumentCaptor.forClass(ApiMessage.class);
+    private ArgumentCaptor<Integer> bufferInitialCapacity;
+
+    @Captor
+    private ArgumentCaptor<ApiMessage> apiMessageCaptor;
+    @Captor
+    private ArgumentCaptor<RequestHeaderData> requestHeaderDataCaptor;
 
     private SampleProduceRequestFilter filter;
     private RequestHeaderData headerData;
@@ -66,11 +77,13 @@ class SampleProduceRequestFilterTest {
      * Unit Test: Checks that transformation is applied when request data contains configured value.
      */
     @Test
-    public void willTransformProduceRequestTest() {
+    public void willTransformProduceRequestTest() throws Exception {
         var requestData = buildProduceRequestData(PRE_TRANSFORM_VALUE);
-        filter.onProduceRequest(API_VERSION, headerData, requestData, context);
-        verify(context).forwardRequest(any(), apiMessageCaptor.capture());
-        var unpackedRequest = unpackProduceRequestData((ProduceRequestData) apiMessageCaptor.getValue());
+
+        var stage = filter.onProduceRequest(API_VERSION, headerData, requestData, context);
+        assertThat(stage).isCompleted();
+        var forwardedRequest = stage.toCompletableFuture().get().message();
+        var unpackedRequest = unpackProduceRequestData((ProduceRequestData) forwardedRequest);
         // 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)
@@ -83,21 +96,23 @@ class SampleProduceRequestFilterTest {
      * value.
      */
     @Test
-    public void wontTransformProduceRequestTest() {
+    public void wontTransformProduceRequestTest() throws Exception {
         var requestData = buildProduceRequestData(NO_TRANSFORM_VALUE);
-        filter.onProduceRequest(API_VERSION, headerData, requestData, context);
-        verify(context).forwardRequest(any(), apiMessageCaptor.capture());
-        var unpackedRequest = unpackProduceRequestData((ProduceRequestData) apiMessageCaptor.getValue());
+        var stage = filter.onProduceRequest(API_VERSION, headerData, requestData, context);
+        assertThat(stage).isCompleted();
+        var forwardedRequest = stage.toCompletableFuture().get().message();
+        var unpackedRequest = unpackProduceRequestData((ProduceRequestData) forwardedRequest);
         // We should see that the unpacked request value has not changed from the input value
         assertThat(unpackedRequest)
                 .containsExactly(NO_TRANSFORM_VALUE);
     }
 
     private void setupContextMock() {
-        context = mock(KrpcFilterContext.class);
-        // create stub for createByteBufferOutputStream method
-        ArgumentCaptor<Integer> argument = ArgumentCaptor.forClass(Integer.class);
-        when(context.createByteBufferOutputStream(argument.capture())).thenAnswer(
+        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(
                 (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 4d961e74..32de5067 100644
--- a/kroxylicious-unit-test/pom.xml
+++ b/kroxylicious-unit-test/pom.xml
@@ -23,6 +23,11 @@
     <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
new file mode 100644
index 00000000..3d57d3d3
--- /dev/null
+++ b/kroxylicious-unit-test/src/main/java/io/kroxylicious/test/filter/TestHarnessResponseFilterResult.java
@@ -0,0 +1,15 @@
+/*
+ * 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 f6ccb617..f761b96c 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
@@ -24,6 +24,8 @@ 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;
@@ -42,7 +44,8 @@ public class FilterHandlerTest extends FilterHarness {
 
     @Test
     public void testForwardRequest() {
-        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> context.forwardRequest(header, request);
+        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> context.requestFilterResultBuilder().forward(header, request)
+                .completed();
         buildChannel(filter);
         var frame = writeRequest(new ApiVersionsRequestData());
         var propagated = channel.readOutbound();
@@ -58,8 +61,10 @@ public class FilterHandlerTest extends FilterHarness {
             }
 
             @Override
-            public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+            public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
+                                                                             KrpcFilterContext context) {
                 fail("Should not be called");
+                return null;
             }
         };
         buildChannel(filter);
@@ -72,14 +77,19 @@ 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.forwardResponse(header, response);
+        ApiVersionsResponseFilter filter = (apiVersion, header, response, context) -> context.responseFilterResultBuilder().forward(header, response)
+                .completed();
         buildChannel(filter);
         var frame = writeResponse(new ApiVersionsResponseData());
         var propagated = channel.readInbound();
@@ -123,8 +133,10 @@ public class FilterHandlerTest extends FilterHarness {
             }
 
             @Override
-            public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
+            public CompletionStage<ResponseFilterResult> onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response,
+                                                                               KrpcFilterContext context) {
                 fail("Should not be called");
+                return null;
             }
         };
         buildChannel(filter);
@@ -136,10 +148,13 @@ public class FilterHandlerTest extends FilterHarness {
     @Test
     public void testDropResponse() {
         ApiVersionsResponseFilter filter = (apiVersion, header, response, context) -> {
-            /* don't call forwardRequest => drop the request */
+            return context.responseFilterResultBuilder().drop().completed();
         };
         buildChannel(filter);
         var frame = writeResponse(new ApiVersionsResponseData());
+        var propagated = channel.readInbound();
+        assertNull(propagated);
+
     }
 
     @Test
@@ -150,6 +165,7 @@ 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);
@@ -195,6 +211,7 @@ 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);
@@ -223,6 +240,7 @@ 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);
@@ -252,6 +270,7 @@ 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);
@@ -267,7 +286,7 @@ public class FilterHandlerTest extends FilterHarness {
         assertFalse(q.isDone(),
                 "Future should not be finished yet");
 
-        // timeout the request
+        // timeout the message
         Thread.sleep(60L);
         channel.runPendingTasks();
 
@@ -291,7 +310,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)));
-            context.forwardResponse(header, response);
+            return context.responseFilterResultBuilder().forward(header, response).completed();
         };
     }
 
@@ -303,7 +322,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)));
-            context.forwardRequest(header, request);
+            return context.requestFilterResultBuilder().forward(header, request).completed();
         };
     }
 
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 bb5c438d..ecd10c57 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaAuthnHandlerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaAuthnHandlerTest.java
@@ -11,6 +11,7 @@ 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;
@@ -49,6 +50,7 @@ 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;
@@ -208,8 +210,9 @@ public class KafkaAuthnHandlerTest {
             }
 
             @Override
-            public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+            public CompletionStage<ResponseFilterResult> 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 7a3b2283..ba059281 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,6 +8,7 @@ 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;
@@ -21,6 +22,7 @@ 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;
 
@@ -29,7 +31,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,
@@ -41,14 +43,15 @@ public class RequestDecoderTest extends AbstractCodecTest {
                         new KafkaRequestDecoder(
                                 DecodePredicate
                                         .forFilters(FilterAndInvoker.build(
-                                                (ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request)))),
+                                                (ApiVersionsRequestFilter) (version, header, request, context) -> context.requestFilterResultBuilder()
+                                                        .forward(header, request).completed()))),
                         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,
@@ -64,9 +67,10 @@ public class RequestDecoderTest extends AbstractCodecTest {
                                     }
 
                                     @Override
-                                    public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
-                                                                     KrpcFilterContext context) {
-                                        context.forwardRequest(header, request);
+                                    public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header,
+                                                                                                     ApiVersionsRequestData request,
+                                                                                                     KrpcFilterContext context) {
+                                        return context.requestFilterResultBuilder().forward(header, request).completed();
                                     }
                                 }))),
                         OpaqueRequestFrame.class),
@@ -74,7 +78,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();
@@ -87,7 +91,9 @@ public class RequestDecoderTest extends AbstractCodecTest {
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
                 DecodePredicate.forFilters(
-                        FilterAndInvoker.build((ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request))))
+                        FilterAndInvoker.build((ApiVersionsRequestFilter) (version, header, request, context) -> {
+                            return context.requestFilterResultBuilder().forward(header, request).completed();
+                        })))
                 .decode(null, byteBuf, messages);
 
         assertEquals(List.of(), messageClasses(messages));
@@ -106,7 +112,9 @@ public class RequestDecoderTest extends AbstractCodecTest {
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
                 DecodePredicate.forFilters(
-                        FilterAndInvoker.build((ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request))))
+                        FilterAndInvoker
+                                .build((ApiVersionsRequestFilter) (version, header, request, context) -> context.requestFilterResultBuilder().forward(header, request)
+                                        .completed())))
                 .decode(null, byteBuf, messages);
 
         assertEquals(List.of(), messageClasses(messages));
@@ -114,19 +122,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);
 
@@ -149,7 +157,9 @@ public class RequestDecoderTest extends AbstractCodecTest {
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
                 DecodePredicate.forFilters(
-                        FilterAndInvoker.build((ApiVersionsRequestFilter) (version, head, request, context) -> context.forwardRequest(header, request))))
+                        FilterAndInvoker
+                                .build((ApiVersionsRequestFilter) (version, head, request, context) -> context.requestFilterResultBuilder().forward(header, request)
+                                        .completed())))
                 .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 8e0451df..05b3dda2 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,14 +19,11 @@ 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;
@@ -47,7 +44,6 @@ 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;
@@ -80,9 +76,6 @@ class BrokerAddressFilterTest {
     @Mock
     private KrpcFilterContext context;
 
-    @Captor
-    private ArgumentCaptor<ApiMessage> apiMessageCaptor;
-
     private BrokerAddressFilter filter;
 
     private FilterInvoker invoker;
@@ -114,31 +107,36 @@ class BrokerAddressFilterTest {
     @ParameterizedTest(name = "{0}")
     @MethodSource(value = "nodeInfoCarryingResponses")
     void nodeInfoCarryingResponsesTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header,
-                                              ApiMessageTestDef responseTestDef) {
+                                              ApiMessageTestDef responseTestDef)
+            throws Exception {
         filterResponseAndVerify(apiMessageType, header, responseTestDef);
     }
 
     @ParameterizedTest(name = "{0}")
     @MethodSource(value = "completeClusterInfoCarryingResponses")
     void reconcileCachesUpstreamAddress(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header,
-                                        ApiMessageTestDef responseTestDef) {
+                                        ApiMessageTestDef responseTestDef)
+            throws Exception {
 
         filterResponseAndVerify(apiMessageType, header, responseTestDef);
         verify(endpointReconciler, times(1)).reconcile(Mockito.eq(virtualCluster), Mockito.anyMap());
     }
 
-    private void filterResponseAndVerify(ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef responseTestDef) {
+    private void filterResponseAndVerify(ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef responseTestDef) throws Exception {
         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();
-        invoker.onResponse(ApiKeys.forId(apiMessageType.apiKey()), header.requestApiVersion(), headerData, response, context);
-        verify(context).forwardResponse(any(), apiMessageCaptor.capture());
+        var stage = invoker.onResponse(ApiKeys.forId(apiMessageType.apiKey()), header.requestApiVersion(), headerData, response, context);
+        assertThat(stage).isCompleted();
+        var value = stage.toCompletableFuture().get().message();
 
-        var filtered = responseWriter.apply(apiMessageCaptor.getValue(), header.requestApiVersion());
+        var filtered = responseWriter.apply(value, 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 8dfc306d..084c26d0 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,11 +36,10 @@ 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;
 
@@ -66,8 +65,9 @@ class EagerMetadataLearnerTest {
     @ParameterizedTest(name = "{0}")
     @MethodSource("preludeRequests")
     public void forwardsRequestsOfKafkaPrelude(String name, ApiKeys apiKey, RequestHeaderData header, ApiMessage request) {
-        learner.onRequest(apiKey, header, request, context);
-        verify(context).forwardRequest(header, request);
+        when(context.requestFilterResultBuilder()).thenReturn(new RequestFilterResultBuilderImpl());
+        var stage = learner.onRequest(apiKey, header, request, context);
+        assertThat(stage).isCompleted();
     }
 
     public static Stream<Arguments> postPreludeRequests() {
@@ -81,22 +81,25 @@ class EagerMetadataLearnerTest {
 
     @ParameterizedTest(name = "{0}")
     @MethodSource("postPreludeRequests")
-    public void spontaneouslyEmitsMetadataRequest(String name, ApiKeys apiKey, RequestHeaderData header, ApiMessage request) {
+    public void spontaneouslyEmitsMetadataRequest(String name, ApiKeys apiKey, RequestHeaderData header, ApiMessage request) throws Exception {
         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));
-        learner.onRequest(apiKey, header, request, context);
+        var stage = learner.onRequest(apiKey, header, request, context);
+        assertThat(stage).isCompleted();
+        var result = stage.toCompletableFuture().get();
 
         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));
         }
-        verify(context, apiKey.equals(ApiKeys.METADATA) ? times(1) : never()).forwardResponse(metadataResponse);
-        verify(context).closeConnection();
+        assertThat(result.closeConnection()).isTrue();
     }
 
     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
new file mode 100644
index 00000000..90c18e4f
--- /dev/null
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/RequestFilterResultBuilderTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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
new file mode 100644
index 00000000..356c6b68
--- /dev/null
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/ResponseFilterResultBuilderTest.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.internal.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/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
index d47ca7fd..f8d4873e 100644
--- a/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
+++ b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
@@ -38,6 +38,10 @@ 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
@@ -170,28 +174,29 @@ public class InvokerDispatchBenchmark {
         }
 
         @Override
-        public void forwardRequest(RequestHeaderData header, ApiMessage request) {
-
+        public <T extends ApiMessage> CompletionStage<T> sendRequest(short apiVersion, ApiMessage request) {
+            return null;
         }
 
         @Override
-        public <T extends ApiMessage> CompletionStage<T> sendRequest(short apiVersion, ApiMessage request) {
+        public CompletionStage<ResponseFilterResult> forwardResponse(ResponseHeaderData header, ApiMessage response) {
             return null;
         }
 
         @Override
-        public void forwardResponse(ResponseHeaderData header, ApiMessage response) {
-
+        public RequestFilterResultBuilder requestFilterResultBuilder() {
+            return null;
         }
 
         @Override
-        public void forwardResponse(ApiMessage response) {
-
+        public CompletionStage<RequestFilterResult> forwardRequest(RequestHeaderData header, ApiMessage request) {
+            return null;
         }
 
         @Override
-        public void closeConnection() {
-
+        public ResponseFilterResultBuilder responseFilterResultBuilder() {
+            return null;
         }
+
     }
 }
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 5e372c9d..1023de78 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/EightInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/EightInterfaceFilter.java
@@ -6,6 +6,8 @@
 
 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;
@@ -26,45 +28,62 @@ 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 void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+        return null;
     }
 
     @Override
-    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
+                                                                   KrpcFilterContext context) {
+        return null;
     }
 
     @Override
-    public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request,
+                                                                      KrpcFilterContext context) {
 
+        return null;
     }
 
     @Override
-    public void onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response,
+                                                                        KrpcFilterContext context) {
 
+        return null;
     }
 
     @Override
-    public void onDeleteTopicsRequest(short apiVersion, RequestHeaderData header, DeleteTopicsRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onDeleteTopicsRequest(short apiVersion, RequestHeaderData header, DeleteTopicsRequestData request,
+                                                                      KrpcFilterContext context) {
 
+        return null;
     }
 
     @Override
-    public void onDeleteTopicsResponse(short apiVersion, ResponseHeaderData header, DeleteTopicsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onDeleteTopicsResponse(short apiVersion, ResponseHeaderData header, DeleteTopicsResponseData response,
+                                                                        KrpcFilterContext context) {
 
+        return null;
     }
 
     @Override
-    public void onDescribeGroupsRequest(short apiVersion, RequestHeaderData header, DescribeGroupsRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onDescribeGroupsRequest(short apiVersion, RequestHeaderData header, DescribeGroupsRequestData request,
+                                                                        KrpcFilterContext context) {
 
+        return null;
     }
 
     @Override
-    public void onDescribeGroupsResponse(short apiVersion, ResponseHeaderData header, DescribeGroupsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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 c3a22e0c..92d3dfdc 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter.java
@@ -6,6 +6,8 @@
 
 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;
@@ -18,24 +20,33 @@ 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 void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+        return null;
     }
 
     @Override
-    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
+                                                                   KrpcFilterContext context) {
+        return null;
     }
 
     @Override
-    public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request,
+                                                                      KrpcFilterContext context) {
 
+        return null;
     }
 
     @Override
-    public void onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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 db734d4e..6bd2e4b2 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter0.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter0.java
@@ -6,6 +6,8 @@
 
 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;
@@ -19,28 +21,37 @@ 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 void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
+        return null;
     }
 
     @Override
-    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
+                                                                   KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
+        return null;
     }
 
     @Override
-    public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
+                                                                     KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
+        return null;
     }
 
     @Override
-    public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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 36c690a8..46f1675a 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter1.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter1.java
@@ -6,6 +6,8 @@
 
 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;
@@ -19,27 +21,36 @@ 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 void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
+        return null;
     }
 
     @Override
-    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
+                                                                   KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
+        return null;
     }
 
     @Override
-    public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
+                                                                     KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
+        return null;
     }
 
     @Override
-    public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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 eaa7240b..a5de5eda 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter2.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter2.java
@@ -6,6 +6,8 @@
 
 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;
@@ -19,28 +21,35 @@ 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 void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
+        return null;
     }
 
     @Override
-    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
+                                                                   KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
+        return null;
     }
 
     @Override
-    public void onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
+        return null;
     }
 
     @Override
-    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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 90c65d97..8b156aba 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter3.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter3.java
@@ -6,6 +6,8 @@
 
 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;
@@ -19,28 +21,35 @@ 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 void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
+        return null;
     }
 
     @Override
-    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
+                                                                   KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
+        return null;
     }
 
     @Override
-    public void onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
         Blackhole.consumeCPU(CONSUME_TOKENS);
+        return null;
     }
 
     @Override
-    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> 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 bacb5e2f..cf727a7b 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/OneInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/OneInterfaceFilter.java
@@ -6,15 +6,20 @@
 
 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 void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
+                                                                   KrpcFilterContext context) {
+        return null;
     }
 }
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 e1c3459d..0b1c9d2c 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter.java
@@ -6,6 +6,8 @@
 
 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;
@@ -14,14 +16,19 @@ 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 void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+        return null;
     }
 
     @Override
-    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
+                                                                   KrpcFilterContext context) {
+        return null;
     }
 }
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 9c249754..5a2bcd37 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter0.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter0.java
@@ -6,6 +6,8 @@
 
 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;
@@ -14,14 +16,19 @@ 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 void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+        return null;
     }
 
     @Override
-    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
+                                                                   KrpcFilterContext context) {
+        return null;
     }
 }
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 a58cd4b7..05a06e6b 100644
--- a/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter1.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter1.java
@@ -6,6 +6,8 @@
 
 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;
@@ -14,14 +16,19 @@ 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 void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+        return null;
     }
 
     @Override
-    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
+                                                                   KrpcFilterContext context) {
+        return null;
     }
 }
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 f7275cca..e02940f9 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,6 +9,7 @@ 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;
@@ -444,38 +445,42 @@ 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 void onRequest(ApiKeys apiKey,
-                          short apiVersion,
-                          RequestHeaderData header,
-                          ApiMessage body,
-                          KrpcFilterContext filterContext) {
+    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey,
+                                                          short apiVersion,
+                                                          RequestHeaderData header,
+                                                          ApiMessage body,
+                                                          KrpcFilterContext filterContext) {
         FilterInvoker invoker = requestInvokers[apiKey.id];
-        invoker.onRequest(apiKey, apiVersion, header, body, filterContext);
+        return 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 void onResponse(ApiKeys apiKey,
-                           short apiVersion,
-                           ResponseHeaderData header,
-                           ApiMessage body,
-                           KrpcFilterContext filterContext) {
+    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey,
+                                                            short apiVersion,
+                                                            ResponseHeaderData header,
+                                                            ApiMessage body,
+                                                            KrpcFilterContext filterContext) {
         FilterInvoker invoker = responseInvokers[apiKey.id];
-        invoker.onResponse(apiKey, apiVersion, header, body, filterContext);
+        return 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 d3d586dd..d986796f 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,6 +6,8 @@
 
 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;
@@ -159,21 +161,23 @@ 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 void onRequest(ApiKeys apiKey,
-                          short apiVersion,
-                          RequestHeaderData header,
-                          ApiMessage body,
-                          KrpcFilterContext filterContext) {
-        switch (apiKey) {
+    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey,
+                                                          short apiVersion,
+                                                          RequestHeaderData header,
+                                                          ApiMessage body,
+                                                          KrpcFilterContext filterContext) {
+        return switch (apiKey) {
             case ADD_OFFSETS_TO_TXN ->
                 ((AddOffsetsToTxnRequestFilter) filter).onAddOffsetsToTxnRequest(apiVersion, header, (AddOffsetsToTxnRequestData) body, filterContext);
             case ADD_PARTITIONS_TO_TXN ->
@@ -278,24 +282,26 @@ 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 void onResponse(ApiKeys apiKey,
-                           short apiVersion,
-                           ResponseHeaderData header,
-                           ApiMessage body,
-                           KrpcFilterContext filterContext) {
-        switch (apiKey) {
+    public CompletionStage<ResponseFilterResult> onResponse(ApiKeys apiKey,
+                                                            short apiVersion,
+                                                            ResponseHeaderData header,
+                                                            ApiMessage body,
+                                                            KrpcFilterContext filterContext) {
+        return switch (apiKey) {
             case ADD_OFFSETS_TO_TXN ->
                 ((AddOffsetsToTxnResponseFilter) filter).onAddOffsetsToTxnResponse(apiVersion, header, (AddOffsetsToTxnResponseData) body, filterContext);
             case ADD_PARTITIONS_TO_TXN ->
@@ -408,7 +414,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);
-        }
+        };
     }
 
     /**