Repository: kroxylicious/kroxylicious
Pull request #338 (2023-05-25 04:18:34)

Add apiVersion to specific message filter onXyz[Request|Response] methods
### Type of change

- Enhancement / new feature

### Description

Adds apiVersion to specific message filter onXyz[Request|Response] methods.

Why:
Currently we do not have a way to access the apiVersion for Responses. The apiVersion is in the request headers but so far we have not made it available to response filters. This is a problem when we want to construct response data in the proxy, we need to know which fields can be serialized at the apiVersion used by the client. Also we have started adding code which is trying to sniff the version by checking for non-null fields, sometimes it might be cleaner to switch on the apiVersion instead.

I opted to add it to the signatures for requests and responses for symmetry despite apiVersion also being present in the request header data.

edit: alternatively it could be contextual data delivered via KrpcContext?

SamBarker: My initial reaction was we should include it on both the request and response methods in the API. 

However this comment:
>edit: alternatively it could be contextual data delivered via KrpcContext?

Triggered another thought: Should the `onResponse` methods have access to the request or the request Headers in general?

It moves us away from providing "pure" transformation filters into offering Request Response aware filters. I'm not really clear what the implications of that are or what use-cases that helps us support.

To answer the actual question though, No I don't think it should be part of the context as relates to a specific request response pair not general interactions with the proxy itself.  
k-wall: > 
> Triggered another thought: Should the `onResponse` methods have access to the request or the request Headers in general?

Filter authors can achieve this with the existing API. We already provide the guarantee that a filter instance is not re-used across channels. The filter author could implement a filter implements `XRequestFilter` and `XResponseFilter`.  On the request side, they cache whatever details they need from the request as filter instance state.   The response side then uses the information.   Admittedly, that is more work for the filter author.

If `onResponse` API were to offer access to the *full request*, that would mean that any XResponseFilter implementation would always impose the cost the decoding the request.  That would disadvantage use-cases that don't need the request (performance hit). 

If `onResponse` API were to offer access to the *request header only*, there will would be a `KafkaRequestDecoder#readHeader` performance hit imposed on those use-cases.   Is it worth it?  The only thing it would give the response filter author beyond the `apiVersion` would be the `clientId` and the request's `_unknownTaggedFields`.









> 
> It moves us away from providing "pure" transformation filters into offering Request Response aware filters. I'm not really clear what the implications of that are or what use-cases that helps us support.
> 
> To answer the actual question though, No I don't think it should be part of the context as relates to a specific request response pair not general interactions with the proxy itself.


SamBarker: > Filter authors can achieve this with the existing API. We already provide the guarantee that a filter instance is not re-used across channels. The filter author could implement a filter implements XRequestFilter and XResponseFilter. On the request side, they cache whatever details they need from the request as filter instance state. The response side then uses the information. Admittedly, that is more work for the filter author.

Yeah it's more work and potentially often repeated. 

Note I'm thinking out-loud rather than advocating for it at this point. I'm trying to envisage the most helpful api for filter authors which makes the most common tasks simple.

Admittedly the sample size is small and potentially biased but most of the filters we have talked about so far are symmetric in that they apply to both the request and response. So I'm thinking about optimising for those cases.

> If onResponse API were to offer access to the full request, that would mean that any XResponseFilter implementation would always impose the cost the decoding the request. That would disadvantage use-cases that don't need the request (performance hit).

That's a good point, but if we passed it in with a wrapper we can lazily decode as required (or return the pre-decoded value from the request) I think we have the state in the correlation manager anyway. Having said that what version of the request to provide to the response? The version the client sent or the version forwarded on? 

> The only thing it would give the response filter author beyond the apiVersion would be the clientId and the request's _unknownTaggedFields.

I think there is a fair chance that those would be interesting to a subset of filter authors. IIRC the schema registry uses that to mark schema coordinates and that's potentially useful in the response. But maybe we are talking about rare enough cases that caching the interesting data in onRequest is good enough.
robobario: > That's a good point, but if we passed it in with a wrapper we can lazily decode as required (or return the pre-decoded value from the request) I think we have the state in the correlation manager anyway. Having said that what version of the request to provide to the response? The version the client sent or the version forwarded on?

Currently we are constrained in the Filters to only forward requests and responses at the `apiVersion` requested by the client (after it discovers the supported apiVersions from the broker+kroxy). So `apiVersion` in the `onXyzRequest|Response` signatures means the `apiVersion` from the client. If you tried to set fields incompatible with that version on the data object it will fail at encode time.

An exception is `sendRequest`, we should make the apiVersion configurable here since the request is initiated by the proxy.

I wonder if these are acceptable limitations. It's conceivable that we could, at the proxy, want to do things like forward a request to the broker at a higher apiVersion than the client supports and then downgrade again at the proxy.

Patch:
diff --git a/api/kroxylicious-filter-api/src/main/templates/Kproxy/Filter.ftl b/api/kroxylicious-filter-api/src/main/templates/Kproxy/Filter.ftl
index 4aa2919c..a2a20feb 100644
--- a/api/kroxylicious-filter-api/src/main/templates/Kproxy/Filter.ftl
+++ b/api/kroxylicious-filter-api/src/main/templates/Kproxy/Filter.ftl
@@ -59,10 +59,11 @@ public interface ${filterClass} extends KrpcFilter {
      * The implementation may modify the given {@code data} in-place and return it,
      * or instantiate a new one.
      *
+     * @param apiVersion the apiVersion of the ${messageSpec.type?lower_case}
      * @param header <#if messageSpec.type?lower_case == 'response'>response<#else>request</#if> header.
      * @param ${msgType} The KRPC message to handle.
      * @param context The context.
      */
-    void on${messageSpec.name}(<#if messageSpec.type?lower_case == 'response'>ResponseHeaderData<#else>RequestHeaderData</#if> header, ${dataClass} ${msgType}, KrpcFilterContext context);
+    void on${messageSpec.name}(short apiVersion, <#if messageSpec.type?lower_case == 'response'>ResponseHeaderData<#else>RequestHeaderData</#if> header, ${dataClass} ${msgType}, KrpcFilterContext context);
 
 }
\ No newline at end of file
diff --git a/api/kroxylicious-filter-api/src/main/templates/Kproxy/FilterInvoker.ftl b/api/kroxylicious-filter-api/src/main/templates/Kproxy/FilterInvoker.ftl
index 5ee849da..7bfd81fb 100644
--- a/api/kroxylicious-filter-api/src/main/templates/Kproxy/FilterInvoker.ftl
+++ b/api/kroxylicious-filter-api/src/main/templates/Kproxy/FilterInvoker.ftl
@@ -59,6 +59,6 @@ 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}(header, (${messageSpec.name}Data) body, filterContext);
+        filter.on${messageSpec.name}(apiVersion, header, (${messageSpec.name}Data) body, filterContext);
     }
 }
\ No newline at end of file
diff --git a/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java b/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
index 6fad2a8e..be91751b 100644
--- a/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
+++ b/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
@@ -105,31 +105,31 @@ public class MultiTenantTransformationFilter
     private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantTransformationFilter.class);
 
     @Override
-    public void onCreateTopicsRequest(RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
+    public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onCreateTopicsResponse(ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
+    public void onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardResponse(header, response);
     }
 
     @Override
-    public void onDeleteTopicsRequest(RequestHeaderData header, DeleteTopicsRequestData request, KrpcFilterContext context) {
+    public void onDeleteTopicsRequest(short apiVersion, RequestHeaderData header, DeleteTopicsRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, topic.topicId() != null));
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onDeleteTopicsResponse(ResponseHeaderData header, DeleteTopicsResponseData response, KrpcFilterContext context) {
+    public void onDeleteTopicsResponse(short apiVersion, ResponseHeaderData header, DeleteTopicsResponseData response, KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardResponse(header, response);
     }
 
     @Override
-    public void onMetadataRequest(RequestHeaderData header, MetadataRequestData request, KrpcFilterContext context) {
+    public void onMetadataRequest(short apiVersion, RequestHeaderData header, MetadataRequestData request, KrpcFilterContext context) {
         if (request.topics() != null) {
             // n.b. request.topics() == null used to query all the topics.
             request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
@@ -139,7 +139,7 @@ public class MultiTenantTransformationFilter
     }
 
     @Override
-    public void onMetadataResponse(ResponseHeaderData header, MetadataResponseData response, KrpcFilterContext context) {
+    public void onMetadataResponse(short apiVersion, ResponseHeaderData header, MetadataResponseData response, KrpcFilterContext context) {
         String tenantPrefix = getTenantPrefix(context);
         response.topics().removeIf(topic -> !topic.name().startsWith(tenantPrefix)); // TODO: allow kafka internal topics to be returned?
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
@@ -147,33 +147,33 @@ public class MultiTenantTransformationFilter
     }
 
     @Override
-    public void onProduceRequest(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         request.topicData().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardRequest(header, request);
 
     }
 
     @Override
-    public void onProduceResponse(ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardResponse(header, response);
 
     }
 
     @Override
-    public void onListOffsetsRequest(RequestHeaderData header, ListOffsetsRequestData request, KrpcFilterContext context) {
+    public void onListOffsetsRequest(short apiVersion, RequestHeaderData header, ListOffsetsRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onListOffsetsResponse(ResponseHeaderData header, ListOffsetsResponseData response, KrpcFilterContext context) {
+    public void onListOffsetsResponse(short apiVersion, ResponseHeaderData header, ListOffsetsResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardResponse(header, response);
     }
 
     @Override
-    public void onOffsetFetchRequest(RequestHeaderData header, OffsetFetchRequestData request, KrpcFilterContext context) {
+    public void onOffsetFetchRequest(short apiVersion, RequestHeaderData header, OffsetFetchRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         request.groups().forEach(requestGroup -> {
             applyTenantPrefix(context, requestGroup::groupId, requestGroup::setGroupId, false);
@@ -185,7 +185,7 @@ public class MultiTenantTransformationFilter
     }
 
     @Override
-    public void onOffsetFetchResponse(ResponseHeaderData header, OffsetFetchResponseData response, KrpcFilterContext context) {
+    public void onOffsetFetchResponse(short apiVersion, ResponseHeaderData header, OffsetFetchResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         response.groups().forEach(responseGroup -> {
             removeTenantPrefix(context, responseGroup::groupId, responseGroup::setGroupId, false);
@@ -195,69 +195,69 @@ public class MultiTenantTransformationFilter
     }
 
     @Override
-    public void onOffsetForLeaderEpochRequest(RequestHeaderData header, OffsetForLeaderEpochRequestData request, KrpcFilterContext context) {
+    public void onOffsetForLeaderEpochRequest(short apiVersion, RequestHeaderData header, OffsetForLeaderEpochRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::topic, topic::setTopic, false));
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onOffsetForLeaderEpochResponse(ResponseHeaderData header, OffsetForLeaderEpochResponseData response, KrpcFilterContext context) {
+    public void onOffsetForLeaderEpochResponse(short apiVersion, ResponseHeaderData header, OffsetForLeaderEpochResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::topic, topic::setTopic, false));
         context.forwardResponse(header, response);
     }
 
     @Override
-    public void onOffsetCommitRequest(RequestHeaderData header, OffsetCommitRequestData request, KrpcFilterContext context) {
+    public void onOffsetCommitRequest(short apiVersion, RequestHeaderData header, OffsetCommitRequestData request, KrpcFilterContext context) {
         applyTenantPrefix(context, request::groupId, request::setGroupId, false);
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onOffsetCommitResponse(ResponseHeaderData header, OffsetCommitResponseData response, KrpcFilterContext context) {
+    public void onOffsetCommitResponse(short apiVersion, ResponseHeaderData header, OffsetCommitResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardResponse(header, response);
     }
 
     @Override
-    public void onOffsetDeleteRequest(RequestHeaderData header, OffsetDeleteRequestData request, KrpcFilterContext context) {
+    public void onOffsetDeleteRequest(short apiVersion, RequestHeaderData header, OffsetDeleteRequestData request, KrpcFilterContext context) {
         applyTenantPrefix(context, request::groupId, request::setGroupId, false);
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onOffsetDeleteResponse(ResponseHeaderData header, OffsetDeleteResponseData response, KrpcFilterContext context) {
+    public void onOffsetDeleteResponse(short apiVersion, ResponseHeaderData header, OffsetDeleteResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardResponse(header, response);
     }
 
     @Override
-    public void onFetchRequest(RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
+    public void onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::topic, topic::setTopic, topic.topicId() != null));
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onFetchResponse(ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
+    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::topic, topic::setTopic, topic.topicId() != null));
         context.forwardResponse(header, response);
     }
 
     @Override
-    public void onFindCoordinatorRequest(RequestHeaderData header, FindCoordinatorRequestData request, KrpcFilterContext context) {
+    public void onFindCoordinatorRequest(short apiVersion, RequestHeaderData header, FindCoordinatorRequestData request, KrpcFilterContext context) {
         request.setCoordinatorKeys(request.coordinatorKeys().stream().map(key -> applyTenantPrefix(context, key)).toList());
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onFindCoordinatorResponse(ResponseHeaderData header, FindCoordinatorResponseData response, KrpcFilterContext context) {
+    public void onFindCoordinatorResponse(short apiVersion, ResponseHeaderData header, FindCoordinatorResponseData response, KrpcFilterContext context) {
         response.coordinators().forEach(coordinator -> removeTenantPrefix(context, coordinator::key, coordinator::setKey, false));
         context.forwardResponse(header, response);
     }
 
     @Override
-    public void onListGroupsResponse(ResponseHeaderData header, ListGroupsResponseData response, KrpcFilterContext context) {
+    public void onListGroupsResponse(short apiVersion, ResponseHeaderData header, ListGroupsResponseData response, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         var filteredGroups = response.groups().stream().filter(listedGroup -> listedGroup.groupId().startsWith(tenantPrefix)).toList();
         filteredGroups.forEach(listedGroup -> removeTenantPrefix(context, listedGroup::groupId, listedGroup::setGroupId, false));
@@ -266,41 +266,41 @@ public class MultiTenantTransformationFilter
     }
 
     @Override
-    public void onJoinGroupRequest(RequestHeaderData header, JoinGroupRequestData request, KrpcFilterContext context) {
+    public void onJoinGroupRequest(short apiVersion, RequestHeaderData header, JoinGroupRequestData request, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setGroupId(tenantPrefix + request.groupId());
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onSyncGroupRequest(RequestHeaderData header, SyncGroupRequestData request, KrpcFilterContext context) {
+    public void onSyncGroupRequest(short apiVersion, RequestHeaderData header, SyncGroupRequestData request, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setGroupId(tenantPrefix + request.groupId());
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onLeaveGroupRequest(RequestHeaderData header, LeaveGroupRequestData request, KrpcFilterContext context) {
+    public void onLeaveGroupRequest(short apiVersion, RequestHeaderData header, LeaveGroupRequestData request, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setGroupId(tenantPrefix + request.groupId());
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onHeartbeatRequest(RequestHeaderData header, HeartbeatRequestData request, KrpcFilterContext context) {
+    public void onHeartbeatRequest(short apiVersion, RequestHeaderData header, HeartbeatRequestData request, KrpcFilterContext context) {
         var tenantPrefix = getTenantPrefix(context);
         request.setGroupId(tenantPrefix + request.groupId());
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onDescribeGroupsRequest(RequestHeaderData header, DescribeGroupsRequestData request, KrpcFilterContext context) {
+    public void onDescribeGroupsRequest(short apiVersion, RequestHeaderData header, DescribeGroupsRequestData request, KrpcFilterContext context) {
         request.setGroups(request.groups().stream().map(group -> applyTenantPrefix(context, group)).toList());
         context.forwardRequest(header, request);
     }
 
     @Override
-    public void onDescribeGroupsResponse(ResponseHeaderData header, DescribeGroupsResponseData response, KrpcFilterContext context) {
+    public void onDescribeGroupsResponse(short apiVersion, ResponseHeaderData header, DescribeGroupsResponseData response, KrpcFilterContext context) {
         response.groups().forEach(group -> removeTenantPrefix(context, group::groupId, group::setGroupId, false));
         context.forwardResponse(header, response);
     }
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilter.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilter.java
index a78cb4fb..e98b4ac9 100644
--- a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilter.java
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilter.java
@@ -62,7 +62,7 @@ public class ProduceValidationFilter implements ProduceRequestFilter, ProduceRes
     }
 
     @Override
-    public void onProduceRequest(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         ProduceRequestValidationResult result = validator.validateRequest(request);
         if (result.isAnyTopicPartitionInvalid()) {
             handleInvalidTopicPartitions(header, request, context, result);
@@ -145,7 +145,7 @@ public class ProduceValidationFilter implements ProduceRequestFilter, ProduceRes
     }
 
     @Override
-    public void onProduceResponse(ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
         ProduceRequestValidationResult produceRequestValidationResult = correlatedResults.remove(header.correlationId());
         if (produceRequestValidationResult != null) {
             LOGGER.debug("augmenting invalid topic-partition details into response: {}", produceRequestValidationResult);
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/example/authn/SaslAuthnObserver.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/example/authn/SaslAuthnObserver.java
index e77d353c..cdd695cf 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/example/authn/SaslAuthnObserver.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/example/authn/SaslAuthnObserver.java
@@ -31,7 +31,8 @@ public class SaslAuthnObserver
     private long sessionLifetimeMs;
 
     @Override
-    public void onSaslHandshakeRequest(RequestHeaderData header,
+    public void onSaslHandshakeRequest(short apiVersion,
+                                       RequestHeaderData header,
                                        SaslHandshakeRequestData request,
                                        KrpcFilterContext context) {
         this.mechanism = request.mechanism();
@@ -39,7 +40,7 @@ public class SaslAuthnObserver
     }
 
     @Override
-    public void onSaslHandshakeResponse(ResponseHeaderData header, SaslHandshakeResponseData response,
+    public void onSaslHandshakeResponse(short apiVersion, ResponseHeaderData header, SaslHandshakeResponseData response,
                                         KrpcFilterContext context) {
         if (response.errorCode() != Errors.NONE.code()) {
             this.mechanism = null;
@@ -48,7 +49,8 @@ public class SaslAuthnObserver
     }
 
     @Override
-    public void onSaslAuthenticateRequest(RequestHeaderData header,
+    public void onSaslAuthenticateRequest(short apiVersion,
+                                          RequestHeaderData header,
                                           SaslAuthenticateRequestData request,
                                           KrpcFilterContext context) {
         byte[] bytes = request.authBytes();
@@ -67,7 +69,7 @@ public class SaslAuthnObserver
     }
 
     @Override
-    public void onSaslAuthenticateResponse(ResponseHeaderData header, SaslAuthenticateResponseData response,
+    public void onSaslAuthenticateResponse(short apiVersion, ResponseHeaderData header, SaslAuthenticateResponseData response,
                                            KrpcFilterContext context) {
         if (response.errorCode() == Errors.NONE.code()) {
             authenticated = true;
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/example/topicencryption/TopicEncryption.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/example/topicencryption/TopicEncryption.java
index fb277b63..1bdb5336 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/example/topicencryption/TopicEncryption.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/example/topicencryption/TopicEncryption.java
@@ -21,7 +21,7 @@ public class TopicEncryption implements ProduceRequestFilter, FetchResponseFilte
     // but other filters will be interested in keeping track of metadata
 
     @Override
-    public void onProduceRequest(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         boolean fragmented = false;
         if (fragmented) {
             // TODO forward the fragments
@@ -35,7 +35,7 @@ public class TopicEncryption implements ProduceRequestFilter, FetchResponseFilte
     }
 
     @Override
-    public void onFetchResponse(ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
+    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
         for (var topicResponse : response.responses()) {
             String topicName = topicResponse.topic();
             if (topicName == null) {
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/example/topicname/TopicNameFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/example/topicname/TopicNameFilter.java
index acd238cc..50cc4b90 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/example/topicname/TopicNameFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/example/topicname/TopicNameFilter.java
@@ -28,7 +28,7 @@ public class TopicNameFilter
     private final Map<Uuid, String> topicNames = new HashMap<>();
 
     @Override
-    public void onMetadataResponse(ResponseHeaderData header, MetadataResponseData response, KrpcFilterContext context) {
+    public void onMetadataResponse(short apiVersion, ResponseHeaderData header, MetadataResponseData response, KrpcFilterContext context) {
         if (response.topics() != null) {
             for (var topic : response.topics()) {
                 topicNames.put(topic.topicId(), topic.name());
@@ -42,7 +42,7 @@ public class TopicNameFilter
     // We don't implement DeleteTopicsRequestFilter because we don't know whether
     // a delete topics request will succeed.
     @Override
-    public void onDeleteTopicsResponse(ResponseHeaderData header, DeleteTopicsResponseData response, KrpcFilterContext context) {
+    public void onDeleteTopicsResponse(short apiVersion, ResponseHeaderData header, DeleteTopicsResponseData response, KrpcFilterContext context) {
         for (var resp : response.responses()) {
             topicNames.remove(resp.topicId());
         }
@@ -50,7 +50,7 @@ public class TopicNameFilter
     }
 
     @Override
-    public void onCreateTopicsResponse(ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
+    public void onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
         for (var topic : response.topics()) {
             topicNames.put(topic.topicId(), topic.name());
         }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
index 086b0a40..5d7e32a5 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
@@ -166,7 +166,7 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
             var filters = new ArrayList<>(filterChainFactory.createFilters());
 
             // Add a filter to the *end of the chain* that gathers the true nodeId/upstream broker mapping.
-            filters.add((MetadataResponseFilter) (header, response, filterContext) -> {
+            filters.add((MetadataResponseFilter) (apiVersion, header, response, filterContext) -> {
                 response.brokers().forEach(b -> {
                     var replacement = new HostPort(b.host(), b.port());
                     var existing = virtualCluster.updateUpstreamClusterAddressForNode(b.nodeId(), replacement);
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 478ba4a4..289b7d7d 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
@@ -64,7 +64,7 @@ public class ApiVersionsFilter implements ApiVersionsResponseFilter {
     }
 
     @Override
-    public void onApiVersionsResponse(ResponseHeaderData header, ApiVersionsResponseData data, KrpcFilterContext context) {
+    public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData data, KrpcFilterContext context) {
         intersectApiVersions(context.channelDescriptor(), data);
         context.forwardResponse(header, data);
     }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java
index baf66c6a..a76c646f 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
@@ -40,7 +40,7 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
     }
 
     @Override
-    public void onMetadataResponse(ResponseHeaderData header, MetadataResponseData data, KrpcFilterContext context) {
+    public void onMetadataResponse(short apiVersion, ResponseHeaderData header, MetadataResponseData data, KrpcFilterContext context) {
         for (MetadataResponseBroker broker : data.brokers()) {
             apply(context, broker, MetadataResponseBroker::nodeId, MetadataResponseBroker::host, MetadataResponseBroker::port, MetadataResponseBroker::setHost,
                     MetadataResponseBroker::setPort);
@@ -49,7 +49,7 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
     }
 
     @Override
-    public void onDescribeClusterResponse(ResponseHeaderData header, DescribeClusterResponseData data, KrpcFilterContext context) {
+    public void onDescribeClusterResponse(short apiVersion, ResponseHeaderData header, DescribeClusterResponseData data, KrpcFilterContext context) {
         for (DescribeClusterBroker broker : data.brokers()) {
             apply(context, broker, DescribeClusterBroker::brokerId, DescribeClusterBroker::host, DescribeClusterBroker::port, DescribeClusterBroker::setHost,
                     DescribeClusterBroker::setPort);
@@ -58,7 +58,7 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
     }
 
     @Override
-    public void onFindCoordinatorResponse(ResponseHeaderData header, FindCoordinatorResponseData data, KrpcFilterContext context) {
+    public void onFindCoordinatorResponse(short apiVersion, ResponseHeaderData header, FindCoordinatorResponseData data, KrpcFilterContext context) {
         for (Coordinator coordinator : data.coordinators()) {
             // If the coordinator is not yet available, the server returns a nodeId of -1.
             if (coordinator.nodeId() >= 0) {
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 041da3b0..2788984e 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java
@@ -71,7 +71,7 @@ public class FetchResponseTransformationFilter implements FetchResponseFilter {
     }
 
     @Override
-    public void onFetchResponse(ResponseHeaderData header, FetchResponseData fetchResponse, KrpcFilterContext context) {
+    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData fetchResponse, KrpcFilterContext context) {
         List<MetadataRequestData.MetadataRequestTopic> requestTopics = fetchResponse.responses().stream()
                 .filter(t -> t.topic().isEmpty())
                 .map(fetchableTopicResponse -> {
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 fd604592..8e2b44f5 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
@@ -70,7 +70,7 @@ public class ProduceRequestTransformationFilter implements ProduceRequestFilter
     }
 
     @Override
-    public void onProduceRequest(RequestHeaderData header, ProduceRequestData data, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData data, KrpcFilterContext context) {
         applyTransformation(context, data);
         context.forwardRequest(header, data);
     }
diff --git a/pom.xml b/pom.xml
index 24f26d91..23291c37 100644
--- a/pom.xml
+++ b/pom.xml
@@ -43,7 +43,6 @@
         <netty.epoll.classifier>linux-x86_64</netty.epoll.classifier>
         <netty.io_uring.classifier>linux-x86_64</netty.io_uring.classifier>
         <netty.kqueue.classifier>osx-x86_64</netty.kqueue.classifier>
-        <lombok.version>1.18.26</lombok.version>
         <assertj-core.version>3.24.2</assertj-core.version>
         <mockito.version>5.3.1</mockito.version>
         <slf4j.version>2.0.7</slf4j.version>
@@ -176,12 +175,6 @@
                 <artifactId>slf4j-simple</artifactId>
                 <version>${slf4j.version}</version>
             </dependency>
-            <dependency>
-                <groupId>org.projectlombok</groupId>
-                <artifactId>lombok</artifactId>
-                <version>${lombok.version}</version>
-                <scope>provided</scope>
-            </dependency>
             <dependency>
                 <groupId>org.assertj</groupId>
                 <artifactId>assertj-core</artifactId>


Test patch:
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 591ab8ac..662fc33c 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CreateTopicRejectFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CreateTopicRejectFilter.java
@@ -16,7 +16,7 @@ public class CreateTopicRejectFilter implements CreateTopicsRequestFilter {
     public static final String ERROR_MESSAGE = "rejecting all topics";
 
     @Override
-    public void onCreateTopicsRequest(RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
+    public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
         CreateTopicsResponseData response = new CreateTopicsResponseData();
         CreateTopicsResponseData.CreatableTopicResultCollection topics = new CreateTopicsResponseData.CreatableTopicResultCollection();
         request.topics().forEach(creatableTopic -> {
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 13c8e41a..15e28278 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
@@ -35,7 +35,7 @@ public class FilterHandlerTest extends FilterHarness {
 
     @Test
     public void testForwardRequest() {
-        ApiVersionsRequestFilter filter = (header, request, context) -> context.forwardRequest(header, request);
+        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> context.forwardRequest(header, request);
         buildChannel(filter);
         var frame = writeRequest(new ApiVersionsRequestData());
         var propagated = channel.readOutbound();
@@ -51,7 +51,7 @@ public class FilterHandlerTest extends FilterHarness {
             }
 
             @Override
-            public void onApiVersionsRequest(RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+            public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
                 fail("Should not be called");
             }
         };
@@ -63,7 +63,7 @@ public class FilterHandlerTest extends FilterHarness {
 
     @Test
     public void testDropRequest() {
-        ApiVersionsRequestFilter filter = (header, request, context) -> {
+        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> {
             /* don't call forwardRequest => drop the request */
         };
         buildChannel(filter);
@@ -72,7 +72,7 @@ public class FilterHandlerTest extends FilterHarness {
 
     @Test
     public void testForwardResponse() {
-        ApiVersionsResponseFilter filter = (header, response, context) -> context.forwardResponse(header, response);
+        ApiVersionsResponseFilter filter = (apiVersion, header, response, context) -> context.forwardResponse(header, response);
         buildChannel(filter);
         var frame = writeResponse(new ApiVersionsResponseData());
         var propagated = channel.readInbound();
@@ -88,7 +88,7 @@ public class FilterHandlerTest extends FilterHarness {
             }
 
             @Override
-            public void onApiVersionsResponse(ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
+            public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
                 fail("Should not be called");
             }
         };
@@ -100,7 +100,7 @@ public class FilterHandlerTest extends FilterHarness {
 
     @Test
     public void testDropResponse() {
-        ApiVersionsResponseFilter filter = (header, response, context) -> {
+        ApiVersionsResponseFilter filter = (apiVersion, header, response, context) -> {
             /* don't call forwardRequest => drop the request */
         };
         buildChannel(filter);
@@ -111,7 +111,7 @@ public class FilterHandlerTest extends FilterHarness {
     public void testSendRequest() {
         FetchRequestData body = new FetchRequestData();
         InternalCompletionStage<ApiMessage>[] fut = new InternalCompletionStage[]{ null };
-        ApiVersionsRequestFilter filter = (header, request, context) -> {
+        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> {
             assertNull(fut[0],
                     "Expected to only be called once");
             fut[0] = (InternalCompletionStage<ApiMessage>) context.sendRequest((short) 3, body);
@@ -156,7 +156,7 @@ public class FilterHandlerTest extends FilterHarness {
     public void testSendRequestCompletionStageCannotBeConvertedToFuture() {
         FetchRequestData body = new FetchRequestData();
         CompletionStage<?>[] fut = { null };
-        ApiVersionsRequestFilter filter = (header, request, context) -> {
+        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> {
             assertNull(fut[0],
                     "Expected to only be called once");
             fut[0] = context.sendRequest((short) 3, body);
@@ -184,7 +184,7 @@ public class FilterHandlerTest extends FilterHarness {
     public void testSendAcklessProduceRequest() throws ExecutionException, InterruptedException {
         ProduceRequestData body = new ProduceRequestData().setAcks((short) 0);
         CompletionStage<ApiMessage>[] fut = new CompletionStage[]{ null };
-        ApiVersionsRequestFilter filter = (header, request, context) -> {
+        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> {
             assertNull(fut[0],
                     "Expected to only be called once");
             fut[0] = context.sendRequest((short) 3, body);
@@ -213,7 +213,7 @@ public class FilterHandlerTest extends FilterHarness {
     public void testSendRequestTimeout() throws InterruptedException {
         FetchRequestData body = new FetchRequestData();
         CompletionStage<ApiMessage>[] fut = new CompletionStage[]{ null };
-        ApiVersionsRequestFilter filter = (header, request, context) -> {
+        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> {
             assertNull(fut[0],
                     "Expected to only be called once");
             fut[0] = context.sendRequest((short) 3, body);
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 6b11bcd5..f1cb3899 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
@@ -38,7 +38,8 @@ public class RequestDecoderTest extends AbstractCodecTest {
                         AbstractCodecTest::deserializeRequestHeaderUsingKafkaApis,
                         AbstractCodecTest::deserializeApiVersionsRequestUsingKafkaApis,
                         new KafkaRequestDecoder(
-                                DecodePredicate.forFilters(List.of((ApiVersionsRequestFilter) (header, request, context) -> context.forwardRequest(header, request)))),
+                                DecodePredicate
+                                        .forFilters(List.of((ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request)))),
                         DecodedRequestFrame.class,
                         (RequestHeaderData header) -> header),
                 "Unexpected correlation id");
@@ -61,7 +62,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
                                     }
 
                                     @Override
-                                    public void onApiVersionsRequest(RequestHeaderData header, ApiVersionsRequestData request,
+                                    public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request,
                                                                      KrpcFilterContext context) {
                                         context.forwardRequest(header, request);
                                     }
@@ -83,7 +84,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
 
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
-                DecodePredicate.forFilters(List.of((ApiVersionsRequestFilter) (header, request, context) -> context.forwardRequest(header, request))))
+                DecodePredicate.forFilters(List.of((ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request))))
                 .decode(null, byteBuf, messages);
 
         assertEquals(List.of(), messageClasses(messages));
@@ -101,7 +102,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
 
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
-                DecodePredicate.forFilters(List.of((ApiVersionsRequestFilter) (header, request, context) -> context.forwardRequest(header, request))))
+                DecodePredicate.forFilters(List.of((ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request))))
                 .decode(null, byteBuf, messages);
 
         assertEquals(List.of(), messageClasses(messages));
@@ -143,7 +144,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
 
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
-                DecodePredicate.forFilters(List.of((ApiVersionsRequestFilter) (head, request, context) -> context.forwardRequest(header, request))))
+                DecodePredicate.forFilters(List.of((ApiVersionsRequestFilter) (version, head, request, context) -> context.forwardRequest(header, request))))
                 .decode(null, byteBuf, messages);
 
         assertEquals(List.of(DecodedRequestFrame.class, DecodedRequestFrame.class), messageClasses(messages));
diff --git a/performance-tests/src/main/java/io/kroxylicious/EightInterfaceFilter.java b/performance-tests/src/main/java/io/kroxylicious/EightInterfaceFilter.java
index 5c0a8174..dc67dc8a 100644
--- a/performance-tests/src/main/java/io/kroxylicious/EightInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/EightInterfaceFilter.java
@@ -31,40 +31,40 @@ public class EightInterfaceFilter implements ProduceResponseFilter, ProduceReque
         DeleteTopicsRequestFilter, DeleteTopicsResponseFilter, DescribeGroupsRequestFilter, DescribeGroupsResponseFilter {
 
     @Override
-    public void onProduceRequest(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
     }
 
     @Override
-    public void onProduceResponse(ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
     }
 
     @Override
-    public void onCreateTopicsRequest(RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
+    public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
 
     }
 
     @Override
-    public void onCreateTopicsResponse(ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
+    public void onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
 
     }
 
     @Override
-    public void onDeleteTopicsRequest(RequestHeaderData header, DeleteTopicsRequestData request, KrpcFilterContext context) {
+    public void onDeleteTopicsRequest(short apiVersion, RequestHeaderData header, DeleteTopicsRequestData request, KrpcFilterContext context) {
 
     }
 
     @Override
-    public void onDeleteTopicsResponse(ResponseHeaderData header, DeleteTopicsResponseData response, KrpcFilterContext context) {
+    public void onDeleteTopicsResponse(short apiVersion, ResponseHeaderData header, DeleteTopicsResponseData response, KrpcFilterContext context) {
 
     }
 
     @Override
-    public void onDescribeGroupsRequest(RequestHeaderData header, DescribeGroupsRequestData request, KrpcFilterContext context) {
+    public void onDescribeGroupsRequest(short apiVersion, RequestHeaderData header, DescribeGroupsRequestData request, KrpcFilterContext context) {
 
     }
 
     @Override
-    public void onDescribeGroupsResponse(ResponseHeaderData header, DescribeGroupsResponseData response, KrpcFilterContext context) {
+    public void onDescribeGroupsResponse(short apiVersion, ResponseHeaderData header, DescribeGroupsResponseData response, KrpcFilterContext context) {
 
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/FourInterfaceFilter.java b/performance-tests/src/main/java/io/kroxylicious/FourInterfaceFilter.java
index addb19a4..9d9db53e 100644
--- a/performance-tests/src/main/java/io/kroxylicious/FourInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/FourInterfaceFilter.java
@@ -22,20 +22,20 @@ import io.kroxylicious.proxy.filter.ProduceResponseFilter;
 public class FourInterfaceFilter implements ProduceResponseFilter, ProduceRequestFilter, CreateTopicsRequestFilter, CreateTopicsResponseFilter {
 
     @Override
-    public void onProduceRequest(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
     }
 
     @Override
-    public void onProduceResponse(ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
     }
 
     @Override
-    public void onCreateTopicsRequest(RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
+    public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
 
     }
 
     @Override
-    public void onCreateTopicsResponse(ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
+    public void onCreateTopicsResponse(short apiVersion, ResponseHeaderData header, CreateTopicsResponseData response, KrpcFilterContext context) {
 
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/OneInterfaceFilter.java b/performance-tests/src/main/java/io/kroxylicious/OneInterfaceFilter.java
index 259b0df4..1ed6dfbb 100644
--- a/performance-tests/src/main/java/io/kroxylicious/OneInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/OneInterfaceFilter.java
@@ -15,6 +15,6 @@ import io.kroxylicious.proxy.filter.ProduceResponseFilter;
 public class OneInterfaceFilter implements ProduceResponseFilter {
 
     @Override
-    public void onProduceResponse(ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/SpcificFilterInvoker.java b/performance-tests/src/main/java/io/kroxylicious/SpcificFilterInvoker.java
index 359d956f..1061bc1c 100644
--- a/performance-tests/src/main/java/io/kroxylicious/SpcificFilterInvoker.java
+++ b/performance-tests/src/main/java/io/kroxylicious/SpcificFilterInvoker.java
@@ -314,91 +314,109 @@ class SpecificFilterInvoker implements FilterInvoker {
                           ApiMessage body,
                           KrpcFilterContext filterContext) {
         switch (apiKey) {
-            case ADD_OFFSETS_TO_TXN -> ((AddOffsetsToTxnRequestFilter) filter).onAddOffsetsToTxnRequest(header, (AddOffsetsToTxnRequestData) body, filterContext);
+            case ADD_OFFSETS_TO_TXN ->
+                ((AddOffsetsToTxnRequestFilter) filter).onAddOffsetsToTxnRequest(apiVersion, header, (AddOffsetsToTxnRequestData) body, filterContext);
             case ADD_PARTITIONS_TO_TXN ->
-                ((AddPartitionsToTxnRequestFilter) filter).onAddPartitionsToTxnRequest(header, (AddPartitionsToTxnRequestData) body, filterContext);
+                ((AddPartitionsToTxnRequestFilter) filter).onAddPartitionsToTxnRequest(apiVersion, header, (AddPartitionsToTxnRequestData) body, filterContext);
             case ALLOCATE_PRODUCER_IDS ->
-                ((AllocateProducerIdsRequestFilter) filter).onAllocateProducerIdsRequest(header, (AllocateProducerIdsRequestData) body, filterContext);
-            case ALTER_CLIENT_QUOTAS -> ((AlterClientQuotasRequestFilter) filter).onAlterClientQuotasRequest(header, (AlterClientQuotasRequestData) body, filterContext);
-            case ALTER_CONFIGS -> ((AlterConfigsRequestFilter) filter).onAlterConfigsRequest(header, (AlterConfigsRequestData) body, filterContext);
-            case ALTER_PARTITION_REASSIGNMENTS -> ((AlterPartitionReassignmentsRequestFilter) filter).onAlterPartitionReassignmentsRequest(header,
+                ((AllocateProducerIdsRequestFilter) filter).onAllocateProducerIdsRequest(apiVersion, header, (AllocateProducerIdsRequestData) body, filterContext);
+            case ALTER_CLIENT_QUOTAS ->
+                ((AlterClientQuotasRequestFilter) filter).onAlterClientQuotasRequest(apiVersion, header, (AlterClientQuotasRequestData) body, filterContext);
+            case ALTER_CONFIGS -> ((AlterConfigsRequestFilter) filter).onAlterConfigsRequest(apiVersion, header, (AlterConfigsRequestData) body, filterContext);
+            case ALTER_PARTITION_REASSIGNMENTS -> ((AlterPartitionReassignmentsRequestFilter) filter).onAlterPartitionReassignmentsRequest(apiVersion, header,
                     (AlterPartitionReassignmentsRequestData) body, filterContext);
-            case ALTER_PARTITION -> ((AlterPartitionRequestFilter) filter).onAlterPartitionRequest(header, (AlterPartitionRequestData) body, filterContext);
+            case ALTER_PARTITION -> ((AlterPartitionRequestFilter) filter).onAlterPartitionRequest(apiVersion, header, (AlterPartitionRequestData) body, filterContext);
             case ALTER_REPLICA_LOG_DIRS ->
-                ((AlterReplicaLogDirsRequestFilter) filter).onAlterReplicaLogDirsRequest(header, (AlterReplicaLogDirsRequestData) body, filterContext);
+                ((AlterReplicaLogDirsRequestFilter) filter).onAlterReplicaLogDirsRequest(apiVersion, header, (AlterReplicaLogDirsRequestData) body, filterContext);
             case ALTER_USER_SCRAM_CREDENTIALS ->
-                ((AlterUserScramCredentialsRequestFilter) filter).onAlterUserScramCredentialsRequest(header, (AlterUserScramCredentialsRequestData) body, filterContext);
-            case API_VERSIONS -> ((ApiVersionsRequestFilter) filter).onApiVersionsRequest(header, (ApiVersionsRequestData) body, filterContext);
-            case BEGIN_QUORUM_EPOCH -> ((BeginQuorumEpochRequestFilter) filter).onBeginQuorumEpochRequest(header, (BeginQuorumEpochRequestData) body, filterContext);
-            case BROKER_HEARTBEAT -> ((BrokerHeartbeatRequestFilter) filter).onBrokerHeartbeatRequest(header, (BrokerHeartbeatRequestData) body, filterContext);
+                ((AlterUserScramCredentialsRequestFilter) filter).onAlterUserScramCredentialsRequest(apiVersion, header, (AlterUserScramCredentialsRequestData) body,
+                        filterContext);
+            case API_VERSIONS -> ((ApiVersionsRequestFilter) filter).onApiVersionsRequest(apiVersion, header, (ApiVersionsRequestData) body, filterContext);
+            case BEGIN_QUORUM_EPOCH ->
+                ((BeginQuorumEpochRequestFilter) filter).onBeginQuorumEpochRequest(apiVersion, header, (BeginQuorumEpochRequestData) body, filterContext);
+            case BROKER_HEARTBEAT ->
+                ((BrokerHeartbeatRequestFilter) filter).onBrokerHeartbeatRequest(apiVersion, header, (BrokerHeartbeatRequestData) body, filterContext);
             case BROKER_REGISTRATION ->
-                ((BrokerRegistrationRequestFilter) filter).onBrokerRegistrationRequest(header, (BrokerRegistrationRequestData) body, filterContext);
+                ((BrokerRegistrationRequestFilter) filter).onBrokerRegistrationRequest(apiVersion, header, (BrokerRegistrationRequestData) body, filterContext);
             case CONTROLLED_SHUTDOWN ->
-                ((ControlledShutdownRequestFilter) filter).onControlledShutdownRequest(header, (ControlledShutdownRequestData) body, filterContext);
-            case CREATE_ACLS -> ((CreateAclsRequestFilter) filter).onCreateAclsRequest(header, (CreateAclsRequestData) body, filterContext);
+                ((ControlledShutdownRequestFilter) filter).onControlledShutdownRequest(apiVersion, header, (ControlledShutdownRequestData) body, filterContext);
+            case CREATE_ACLS -> ((CreateAclsRequestFilter) filter).onCreateAclsRequest(apiVersion, header, (CreateAclsRequestData) body, filterContext);
             case CREATE_DELEGATION_TOKEN ->
-                ((CreateDelegationTokenRequestFilter) filter).onCreateDelegationTokenRequest(header, (CreateDelegationTokenRequestData) body, filterContext);
-            case CREATE_PARTITIONS -> ((CreatePartitionsRequestFilter) filter).onCreatePartitionsRequest(header, (CreatePartitionsRequestData) body, filterContext);
-            case CREATE_TOPICS -> ((CreateTopicsRequestFilter) filter).onCreateTopicsRequest(header, (CreateTopicsRequestData) body, filterContext);
-            case DELETE_ACLS -> ((DeleteAclsRequestFilter) filter).onDeleteAclsRequest(header, (DeleteAclsRequestData) body, filterContext);
-            case DELETE_GROUPS -> ((DeleteGroupsRequestFilter) filter).onDeleteGroupsRequest(header, (DeleteGroupsRequestData) body, filterContext);
-            case DELETE_RECORDS -> ((DeleteRecordsRequestFilter) filter).onDeleteRecordsRequest(header, (DeleteRecordsRequestData) body, filterContext);
-            case DELETE_TOPICS -> ((DeleteTopicsRequestFilter) filter).onDeleteTopicsRequest(header, (DeleteTopicsRequestData) body, filterContext);
-            case DESCRIBE_ACLS -> ((DescribeAclsRequestFilter) filter).onDescribeAclsRequest(header, (DescribeAclsRequestData) body, filterContext);
+                ((CreateDelegationTokenRequestFilter) filter).onCreateDelegationTokenRequest(apiVersion, header, (CreateDelegationTokenRequestData) body, filterContext);
+            case CREATE_PARTITIONS ->
+                ((CreatePartitionsRequestFilter) filter).onCreatePartitionsRequest(apiVersion, header, (CreatePartitionsRequestData) body, filterContext);
+            case CREATE_TOPICS -> ((CreateTopicsRequestFilter) filter).onCreateTopicsRequest(apiVersion, header, (CreateTopicsRequestData) body, filterContext);
+            case DELETE_ACLS -> ((DeleteAclsRequestFilter) filter).onDeleteAclsRequest(apiVersion, header, (DeleteAclsRequestData) body, filterContext);
+            case DELETE_GROUPS -> ((DeleteGroupsRequestFilter) filter).onDeleteGroupsRequest(apiVersion, header, (DeleteGroupsRequestData) body, filterContext);
+            case DELETE_RECORDS -> ((DeleteRecordsRequestFilter) filter).onDeleteRecordsRequest(apiVersion, header, (DeleteRecordsRequestData) body, filterContext);
+            case DELETE_TOPICS -> ((DeleteTopicsRequestFilter) filter).onDeleteTopicsRequest(apiVersion, header, (DeleteTopicsRequestData) body, filterContext);
+            case DESCRIBE_ACLS -> ((DescribeAclsRequestFilter) filter).onDescribeAclsRequest(apiVersion, header, (DescribeAclsRequestData) body, filterContext);
             case DESCRIBE_CLIENT_QUOTAS ->
-                ((DescribeClientQuotasRequestFilter) filter).onDescribeClientQuotasRequest(header, (DescribeClientQuotasRequestData) body, filterContext);
-            case DESCRIBE_CLUSTER -> ((DescribeClusterRequestFilter) filter).onDescribeClusterRequest(header, (DescribeClusterRequestData) body, filterContext);
-            case DESCRIBE_CONFIGS -> ((DescribeConfigsRequestFilter) filter).onDescribeConfigsRequest(header, (DescribeConfigsRequestData) body, filterContext);
+                ((DescribeClientQuotasRequestFilter) filter).onDescribeClientQuotasRequest(apiVersion, header, (DescribeClientQuotasRequestData) body, filterContext);
+            case DESCRIBE_CLUSTER ->
+                ((DescribeClusterRequestFilter) filter).onDescribeClusterRequest(apiVersion, header, (DescribeClusterRequestData) body, filterContext);
+            case DESCRIBE_CONFIGS ->
+                ((DescribeConfigsRequestFilter) filter).onDescribeConfigsRequest(apiVersion, header, (DescribeConfigsRequestData) body, filterContext);
             case DESCRIBE_DELEGATION_TOKEN ->
-                ((DescribeDelegationTokenRequestFilter) filter).onDescribeDelegationTokenRequest(header, (DescribeDelegationTokenRequestData) body, filterContext);
-            case DESCRIBE_GROUPS -> ((DescribeGroupsRequestFilter) filter).onDescribeGroupsRequest(header, (DescribeGroupsRequestData) body, filterContext);
-            case DESCRIBE_LOG_DIRS -> ((DescribeLogDirsRequestFilter) filter).onDescribeLogDirsRequest(header, (DescribeLogDirsRequestData) body, filterContext);
-            case DESCRIBE_PRODUCERS -> ((DescribeProducersRequestFilter) filter).onDescribeProducersRequest(header, (DescribeProducersRequestData) body, filterContext);
-            case DESCRIBE_QUORUM -> ((DescribeQuorumRequestFilter) filter).onDescribeQuorumRequest(header, (DescribeQuorumRequestData) body, filterContext);
+                ((DescribeDelegationTokenRequestFilter) filter).onDescribeDelegationTokenRequest(apiVersion, header, (DescribeDelegationTokenRequestData) body,
+                        filterContext);
+            case DESCRIBE_GROUPS -> ((DescribeGroupsRequestFilter) filter).onDescribeGroupsRequest(apiVersion, header, (DescribeGroupsRequestData) body, filterContext);
+            case DESCRIBE_LOG_DIRS ->
+                ((DescribeLogDirsRequestFilter) filter).onDescribeLogDirsRequest(apiVersion, header, (DescribeLogDirsRequestData) body, filterContext);
+            case DESCRIBE_PRODUCERS ->
+                ((DescribeProducersRequestFilter) filter).onDescribeProducersRequest(apiVersion, header, (DescribeProducersRequestData) body, filterContext);
+            case DESCRIBE_QUORUM -> ((DescribeQuorumRequestFilter) filter).onDescribeQuorumRequest(apiVersion, header, (DescribeQuorumRequestData) body, filterContext);
             case DESCRIBE_TRANSACTIONS ->
-                ((DescribeTransactionsRequestFilter) filter).onDescribeTransactionsRequest(header, (DescribeTransactionsRequestData) body, filterContext);
-            case DESCRIBE_USER_SCRAM_CREDENTIALS -> ((DescribeUserScramCredentialsRequestFilter) filter).onDescribeUserScramCredentialsRequest(header,
+                ((DescribeTransactionsRequestFilter) filter).onDescribeTransactionsRequest(apiVersion, header, (DescribeTransactionsRequestData) body, filterContext);
+            case DESCRIBE_USER_SCRAM_CREDENTIALS -> ((DescribeUserScramCredentialsRequestFilter) filter).onDescribeUserScramCredentialsRequest(apiVersion, header,
                     (DescribeUserScramCredentialsRequestData) body, filterContext);
-            case ELECT_LEADERS -> ((ElectLeadersRequestFilter) filter).onElectLeadersRequest(header, (ElectLeadersRequestData) body, filterContext);
-            case END_QUORUM_EPOCH -> ((EndQuorumEpochRequestFilter) filter).onEndQuorumEpochRequest(header, (EndQuorumEpochRequestData) body, filterContext);
-            case END_TXN -> ((EndTxnRequestFilter) filter).onEndTxnRequest(header, (EndTxnRequestData) body, filterContext);
-            case ENVELOPE -> ((EnvelopeRequestFilter) filter).onEnvelopeRequest(header, (EnvelopeRequestData) body, filterContext);
+            case ELECT_LEADERS -> ((ElectLeadersRequestFilter) filter).onElectLeadersRequest(apiVersion, header, (ElectLeadersRequestData) body, filterContext);
+            case END_QUORUM_EPOCH -> ((EndQuorumEpochRequestFilter) filter).onEndQuorumEpochRequest(apiVersion, header, (EndQuorumEpochRequestData) body, filterContext);
+            case END_TXN -> ((EndTxnRequestFilter) filter).onEndTxnRequest(apiVersion, header, (EndTxnRequestData) body, filterContext);
+            case ENVELOPE -> ((EnvelopeRequestFilter) filter).onEnvelopeRequest(apiVersion, header, (EnvelopeRequestData) body, filterContext);
             case EXPIRE_DELEGATION_TOKEN ->
-                ((ExpireDelegationTokenRequestFilter) filter).onExpireDelegationTokenRequest(header, (ExpireDelegationTokenRequestData) body, filterContext);
-            case FETCH -> ((FetchRequestFilter) filter).onFetchRequest(header, (FetchRequestData) body, filterContext);
-            case FETCH_SNAPSHOT -> ((FetchSnapshotRequestFilter) filter).onFetchSnapshotRequest(header, (FetchSnapshotRequestData) body, filterContext);
-            case FIND_COORDINATOR -> ((FindCoordinatorRequestFilter) filter).onFindCoordinatorRequest(header, (FindCoordinatorRequestData) body, filterContext);
-            case HEARTBEAT -> ((HeartbeatRequestFilter) filter).onHeartbeatRequest(header, (HeartbeatRequestData) body, filterContext);
+                ((ExpireDelegationTokenRequestFilter) filter).onExpireDelegationTokenRequest(apiVersion, header, (ExpireDelegationTokenRequestData) body, filterContext);
+            case FETCH -> ((FetchRequestFilter) filter).onFetchRequest(apiVersion, header, (FetchRequestData) body, filterContext);
+            case FETCH_SNAPSHOT -> ((FetchSnapshotRequestFilter) filter).onFetchSnapshotRequest(apiVersion, header, (FetchSnapshotRequestData) body, filterContext);
+            case FIND_COORDINATOR ->
+                ((FindCoordinatorRequestFilter) filter).onFindCoordinatorRequest(apiVersion, header, (FindCoordinatorRequestData) body, filterContext);
+            case HEARTBEAT -> ((HeartbeatRequestFilter) filter).onHeartbeatRequest(apiVersion, header, (HeartbeatRequestData) body, filterContext);
             case INCREMENTAL_ALTER_CONFIGS ->
-                ((IncrementalAlterConfigsRequestFilter) filter).onIncrementalAlterConfigsRequest(header, (IncrementalAlterConfigsRequestData) body, filterContext);
-            case INIT_PRODUCER_ID -> ((InitProducerIdRequestFilter) filter).onInitProducerIdRequest(header, (InitProducerIdRequestData) body, filterContext);
-            case JOIN_GROUP -> ((JoinGroupRequestFilter) filter).onJoinGroupRequest(header, (JoinGroupRequestData) body, filterContext);
-            case LEADER_AND_ISR -> ((LeaderAndIsrRequestFilter) filter).onLeaderAndIsrRequest(header, (LeaderAndIsrRequestData) body, filterContext);
-            case LEAVE_GROUP -> ((LeaveGroupRequestFilter) filter).onLeaveGroupRequest(header, (LeaveGroupRequestData) body, filterContext);
-            case LIST_GROUPS -> ((ListGroupsRequestFilter) filter).onListGroupsRequest(header, (ListGroupsRequestData) body, filterContext);
-            case LIST_OFFSETS -> ((ListOffsetsRequestFilter) filter).onListOffsetsRequest(header, (ListOffsetsRequestData) body, filterContext);
-            case LIST_PARTITION_REASSIGNMENTS -> ((ListPartitionReassignmentsRequestFilter) filter).onListPartitionReassignmentsRequest(header,
+                ((IncrementalAlterConfigsRequestFilter) filter).onIncrementalAlterConfigsRequest(apiVersion, header, (IncrementalAlterConfigsRequestData) body,
+                        filterContext);
+            case INIT_PRODUCER_ID -> ((InitProducerIdRequestFilter) filter).onInitProducerIdRequest(apiVersion, header, (InitProducerIdRequestData) body, filterContext);
+            case JOIN_GROUP -> ((JoinGroupRequestFilter) filter).onJoinGroupRequest(apiVersion, header, (JoinGroupRequestData) body, filterContext);
+            case LEADER_AND_ISR -> ((LeaderAndIsrRequestFilter) filter).onLeaderAndIsrRequest(apiVersion, header, (LeaderAndIsrRequestData) body, filterContext);
+            case LEAVE_GROUP -> ((LeaveGroupRequestFilter) filter).onLeaveGroupRequest(apiVersion, header, (LeaveGroupRequestData) body, filterContext);
+            case LIST_GROUPS -> ((ListGroupsRequestFilter) filter).onListGroupsRequest(apiVersion, header, (ListGroupsRequestData) body, filterContext);
+            case LIST_OFFSETS -> ((ListOffsetsRequestFilter) filter).onListOffsetsRequest(apiVersion, header, (ListOffsetsRequestData) body, filterContext);
+            case LIST_PARTITION_REASSIGNMENTS -> ((ListPartitionReassignmentsRequestFilter) filter).onListPartitionReassignmentsRequest(apiVersion, header,
                     (ListPartitionReassignmentsRequestData) body, filterContext);
-            case LIST_TRANSACTIONS -> ((ListTransactionsRequestFilter) filter).onListTransactionsRequest(header, (ListTransactionsRequestData) body, filterContext);
-            case METADATA -> ((MetadataRequestFilter) filter).onMetadataRequest(header, (MetadataRequestData) body, filterContext);
-            case OFFSET_COMMIT -> ((OffsetCommitRequestFilter) filter).onOffsetCommitRequest(header, (OffsetCommitRequestData) body, filterContext);
-            case OFFSET_DELETE -> ((OffsetDeleteRequestFilter) filter).onOffsetDeleteRequest(header, (OffsetDeleteRequestData) body, filterContext);
-            case OFFSET_FETCH -> ((OffsetFetchRequestFilter) filter).onOffsetFetchRequest(header, (OffsetFetchRequestData) body, filterContext);
+            case LIST_TRANSACTIONS ->
+                ((ListTransactionsRequestFilter) filter).onListTransactionsRequest(apiVersion, header, (ListTransactionsRequestData) body, filterContext);
+            case METADATA -> ((MetadataRequestFilter) filter).onMetadataRequest(apiVersion, header, (MetadataRequestData) body, filterContext);
+            case OFFSET_COMMIT -> ((OffsetCommitRequestFilter) filter).onOffsetCommitRequest(apiVersion, header, (OffsetCommitRequestData) body, filterContext);
+            case OFFSET_DELETE -> ((OffsetDeleteRequestFilter) filter).onOffsetDeleteRequest(apiVersion, header, (OffsetDeleteRequestData) body, filterContext);
+            case OFFSET_FETCH -> ((OffsetFetchRequestFilter) filter).onOffsetFetchRequest(apiVersion, header, (OffsetFetchRequestData) body, filterContext);
             case OFFSET_FOR_LEADER_EPOCH ->
-                ((OffsetForLeaderEpochRequestFilter) filter).onOffsetForLeaderEpochRequest(header, (OffsetForLeaderEpochRequestData) body, filterContext);
-            case PRODUCE -> ((ProduceRequestFilter) filter).onProduceRequest(header, (ProduceRequestData) body, filterContext);
+                ((OffsetForLeaderEpochRequestFilter) filter).onOffsetForLeaderEpochRequest(apiVersion, header, (OffsetForLeaderEpochRequestData) body, filterContext);
+            case PRODUCE -> ((ProduceRequestFilter) filter).onProduceRequest(apiVersion, header, (ProduceRequestData) body, filterContext);
             case RENEW_DELEGATION_TOKEN ->
-                ((RenewDelegationTokenRequestFilter) filter).onRenewDelegationTokenRequest(header, (RenewDelegationTokenRequestData) body, filterContext);
-            case SASL_AUTHENTICATE -> ((SaslAuthenticateRequestFilter) filter).onSaslAuthenticateRequest(header, (SaslAuthenticateRequestData) body, filterContext);
-            case SASL_HANDSHAKE -> ((SaslHandshakeRequestFilter) filter).onSaslHandshakeRequest(header, (SaslHandshakeRequestData) body, filterContext);
-            case STOP_REPLICA -> ((StopReplicaRequestFilter) filter).onStopReplicaRequest(header, (StopReplicaRequestData) body, filterContext);
-            case SYNC_GROUP -> ((SyncGroupRequestFilter) filter).onSyncGroupRequest(header, (SyncGroupRequestData) body, filterContext);
-            case TXN_OFFSET_COMMIT -> ((TxnOffsetCommitRequestFilter) filter).onTxnOffsetCommitRequest(header, (TxnOffsetCommitRequestData) body, filterContext);
-            case UNREGISTER_BROKER -> ((UnregisterBrokerRequestFilter) filter).onUnregisterBrokerRequest(header, (UnregisterBrokerRequestData) body, filterContext);
-            case UPDATE_FEATURES -> ((UpdateFeaturesRequestFilter) filter).onUpdateFeaturesRequest(header, (UpdateFeaturesRequestData) body, filterContext);
-            case UPDATE_METADATA -> ((UpdateMetadataRequestFilter) filter).onUpdateMetadataRequest(header, (UpdateMetadataRequestData) body, filterContext);
-            case VOTE -> ((VoteRequestFilter) filter).onVoteRequest(header, (VoteRequestData) body, filterContext);
-            case WRITE_TXN_MARKERS -> ((WriteTxnMarkersRequestFilter) filter).onWriteTxnMarkersRequest(header, (WriteTxnMarkersRequestData) body, filterContext);
+                ((RenewDelegationTokenRequestFilter) filter).onRenewDelegationTokenRequest(apiVersion, header, (RenewDelegationTokenRequestData) body, filterContext);
+            case SASL_AUTHENTICATE ->
+                ((SaslAuthenticateRequestFilter) filter).onSaslAuthenticateRequest(apiVersion, header, (SaslAuthenticateRequestData) body, filterContext);
+            case SASL_HANDSHAKE -> ((SaslHandshakeRequestFilter) filter).onSaslHandshakeRequest(apiVersion, header, (SaslHandshakeRequestData) body, filterContext);
+            case STOP_REPLICA -> ((StopReplicaRequestFilter) filter).onStopReplicaRequest(apiVersion, header, (StopReplicaRequestData) body, filterContext);
+            case SYNC_GROUP -> ((SyncGroupRequestFilter) filter).onSyncGroupRequest(apiVersion, header, (SyncGroupRequestData) body, filterContext);
+            case TXN_OFFSET_COMMIT ->
+                ((TxnOffsetCommitRequestFilter) filter).onTxnOffsetCommitRequest(apiVersion, header, (TxnOffsetCommitRequestData) body, filterContext);
+            case UNREGISTER_BROKER ->
+                ((UnregisterBrokerRequestFilter) filter).onUnregisterBrokerRequest(apiVersion, header, (UnregisterBrokerRequestData) body, filterContext);
+            case UPDATE_FEATURES -> ((UpdateFeaturesRequestFilter) filter).onUpdateFeaturesRequest(apiVersion, header, (UpdateFeaturesRequestData) body, filterContext);
+            case UPDATE_METADATA -> ((UpdateMetadataRequestFilter) filter).onUpdateMetadataRequest(apiVersion, header, (UpdateMetadataRequestData) body, filterContext);
+            case VOTE -> ((VoteRequestFilter) filter).onVoteRequest(apiVersion, header, (VoteRequestData) body, filterContext);
+            case WRITE_TXN_MARKERS ->
+                ((WriteTxnMarkersRequestFilter) filter).onWriteTxnMarkersRequest(apiVersion, header, (WriteTxnMarkersRequestData) body, filterContext);
             default -> throw new IllegalStateException("Unsupported RPC " + apiKey);
         }
     }
@@ -418,93 +436,117 @@ class SpecificFilterInvoker implements FilterInvoker {
                            ApiMessage body,
                            KrpcFilterContext filterContext) {
         switch (apiKey) {
-            case ADD_OFFSETS_TO_TXN -> ((AddOffsetsToTxnResponseFilter) filter).onAddOffsetsToTxnResponse(header, (AddOffsetsToTxnResponseData) body, filterContext);
+            case ADD_OFFSETS_TO_TXN ->
+                ((AddOffsetsToTxnResponseFilter) filter).onAddOffsetsToTxnResponse(apiVersion, header, (AddOffsetsToTxnResponseData) body, filterContext);
             case ADD_PARTITIONS_TO_TXN ->
-                ((AddPartitionsToTxnResponseFilter) filter).onAddPartitionsToTxnResponse(header, (AddPartitionsToTxnResponseData) body, filterContext);
+                ((AddPartitionsToTxnResponseFilter) filter).onAddPartitionsToTxnResponse(apiVersion, header, (AddPartitionsToTxnResponseData) body, filterContext);
             case ALLOCATE_PRODUCER_IDS ->
-                ((AllocateProducerIdsResponseFilter) filter).onAllocateProducerIdsResponse(header, (AllocateProducerIdsResponseData) body, filterContext);
+                ((AllocateProducerIdsResponseFilter) filter).onAllocateProducerIdsResponse(apiVersion, header, (AllocateProducerIdsResponseData) body, filterContext);
             case ALTER_CLIENT_QUOTAS ->
-                ((AlterClientQuotasResponseFilter) filter).onAlterClientQuotasResponse(header, (AlterClientQuotasResponseData) body, filterContext);
-            case ALTER_CONFIGS -> ((AlterConfigsResponseFilter) filter).onAlterConfigsResponse(header, (AlterConfigsResponseData) body, filterContext);
-            case ALTER_PARTITION_REASSIGNMENTS -> ((AlterPartitionReassignmentsResponseFilter) filter).onAlterPartitionReassignmentsResponse(header,
+                ((AlterClientQuotasResponseFilter) filter).onAlterClientQuotasResponse(apiVersion, header, (AlterClientQuotasResponseData) body, filterContext);
+            case ALTER_CONFIGS -> ((AlterConfigsResponseFilter) filter).onAlterConfigsResponse(apiVersion, header, (AlterConfigsResponseData) body, filterContext);
+            case ALTER_PARTITION_REASSIGNMENTS -> ((AlterPartitionReassignmentsResponseFilter) filter).onAlterPartitionReassignmentsResponse(apiVersion, header,
                     (AlterPartitionReassignmentsResponseData) body, filterContext);
-            case ALTER_PARTITION -> ((AlterPartitionResponseFilter) filter).onAlterPartitionResponse(header, (AlterPartitionResponseData) body, filterContext);
+            case ALTER_PARTITION ->
+                ((AlterPartitionResponseFilter) filter).onAlterPartitionResponse(apiVersion, header, (AlterPartitionResponseData) body, filterContext);
             case ALTER_REPLICA_LOG_DIRS ->
-                ((AlterReplicaLogDirsResponseFilter) filter).onAlterReplicaLogDirsResponse(header, (AlterReplicaLogDirsResponseData) body, filterContext);
-            case ALTER_USER_SCRAM_CREDENTIALS -> ((AlterUserScramCredentialsResponseFilter) filter).onAlterUserScramCredentialsResponse(header,
+                ((AlterReplicaLogDirsResponseFilter) filter).onAlterReplicaLogDirsResponse(apiVersion, header, (AlterReplicaLogDirsResponseData) body, filterContext);
+            case ALTER_USER_SCRAM_CREDENTIALS -> ((AlterUserScramCredentialsResponseFilter) filter).onAlterUserScramCredentialsResponse(apiVersion, header,
                     (AlterUserScramCredentialsResponseData) body, filterContext);
-            case API_VERSIONS -> ((ApiVersionsResponseFilter) filter).onApiVersionsResponse(header, (ApiVersionsResponseData) body, filterContext);
-            case BEGIN_QUORUM_EPOCH -> ((BeginQuorumEpochResponseFilter) filter).onBeginQuorumEpochResponse(header, (BeginQuorumEpochResponseData) body, filterContext);
-            case BROKER_HEARTBEAT -> ((BrokerHeartbeatResponseFilter) filter).onBrokerHeartbeatResponse(header, (BrokerHeartbeatResponseData) body, filterContext);
+            case API_VERSIONS -> ((ApiVersionsResponseFilter) filter).onApiVersionsResponse(apiVersion, header, (ApiVersionsResponseData) body, filterContext);
+            case BEGIN_QUORUM_EPOCH ->
+                ((BeginQuorumEpochResponseFilter) filter).onBeginQuorumEpochResponse(apiVersion, header, (BeginQuorumEpochResponseData) body, filterContext);
+            case BROKER_HEARTBEAT ->
+                ((BrokerHeartbeatResponseFilter) filter).onBrokerHeartbeatResponse(apiVersion, header, (BrokerHeartbeatResponseData) body, filterContext);
             case BROKER_REGISTRATION ->
-                ((BrokerRegistrationResponseFilter) filter).onBrokerRegistrationResponse(header, (BrokerRegistrationResponseData) body, filterContext);
+                ((BrokerRegistrationResponseFilter) filter).onBrokerRegistrationResponse(apiVersion, header, (BrokerRegistrationResponseData) body, filterContext);
             case CONTROLLED_SHUTDOWN ->
-                ((ControlledShutdownResponseFilter) filter).onControlledShutdownResponse(header, (ControlledShutdownResponseData) body, filterContext);
-            case CREATE_ACLS -> ((CreateAclsResponseFilter) filter).onCreateAclsResponse(header, (CreateAclsResponseData) body, filterContext);
+                ((ControlledShutdownResponseFilter) filter).onControlledShutdownResponse(apiVersion, header, (ControlledShutdownResponseData) body, filterContext);
+            case CREATE_ACLS -> ((CreateAclsResponseFilter) filter).onCreateAclsResponse(apiVersion, header, (CreateAclsResponseData) body, filterContext);
             case CREATE_DELEGATION_TOKEN ->
-                ((CreateDelegationTokenResponseFilter) filter).onCreateDelegationTokenResponse(header, (CreateDelegationTokenResponseData) body, filterContext);
-            case CREATE_PARTITIONS -> ((CreatePartitionsResponseFilter) filter).onCreatePartitionsResponse(header, (CreatePartitionsResponseData) body, filterContext);
-            case CREATE_TOPICS -> ((CreateTopicsResponseFilter) filter).onCreateTopicsResponse(header, (CreateTopicsResponseData) body, filterContext);
-            case DELETE_ACLS -> ((DeleteAclsResponseFilter) filter).onDeleteAclsResponse(header, (DeleteAclsResponseData) body, filterContext);
-            case DELETE_GROUPS -> ((DeleteGroupsResponseFilter) filter).onDeleteGroupsResponse(header, (DeleteGroupsResponseData) body, filterContext);
-            case DELETE_RECORDS -> ((DeleteRecordsResponseFilter) filter).onDeleteRecordsResponse(header, (DeleteRecordsResponseData) body, filterContext);
-            case DELETE_TOPICS -> ((DeleteTopicsResponseFilter) filter).onDeleteTopicsResponse(header, (DeleteTopicsResponseData) body, filterContext);
-            case DESCRIBE_ACLS -> ((DescribeAclsResponseFilter) filter).onDescribeAclsResponse(header, (DescribeAclsResponseData) body, filterContext);
+                ((CreateDelegationTokenResponseFilter) filter).onCreateDelegationTokenResponse(apiVersion, header, (CreateDelegationTokenResponseData) body,
+                        filterContext);
+            case CREATE_PARTITIONS ->
+                ((CreatePartitionsResponseFilter) filter).onCreatePartitionsResponse(apiVersion, header, (CreatePartitionsResponseData) body, filterContext);
+            case CREATE_TOPICS -> ((CreateTopicsResponseFilter) filter).onCreateTopicsResponse(apiVersion, header, (CreateTopicsResponseData) body, filterContext);
+            case DELETE_ACLS -> ((DeleteAclsResponseFilter) filter).onDeleteAclsResponse(apiVersion, header, (DeleteAclsResponseData) body, filterContext);
+            case DELETE_GROUPS -> ((DeleteGroupsResponseFilter) filter).onDeleteGroupsResponse(apiVersion, header, (DeleteGroupsResponseData) body, filterContext);
+            case DELETE_RECORDS -> ((DeleteRecordsResponseFilter) filter).onDeleteRecordsResponse(apiVersion, header, (DeleteRecordsResponseData) body, filterContext);
+            case DELETE_TOPICS -> ((DeleteTopicsResponseFilter) filter).onDeleteTopicsResponse(apiVersion, header, (DeleteTopicsResponseData) body, filterContext);
+            case DESCRIBE_ACLS -> ((DescribeAclsResponseFilter) filter).onDescribeAclsResponse(apiVersion, header, (DescribeAclsResponseData) body, filterContext);
             case DESCRIBE_CLIENT_QUOTAS ->
-                ((DescribeClientQuotasResponseFilter) filter).onDescribeClientQuotasResponse(header, (DescribeClientQuotasResponseData) body, filterContext);
-            case DESCRIBE_CLUSTER -> ((DescribeClusterResponseFilter) filter).onDescribeClusterResponse(header, (DescribeClusterResponseData) body, filterContext);
-            case DESCRIBE_CONFIGS -> ((DescribeConfigsResponseFilter) filter).onDescribeConfigsResponse(header, (DescribeConfigsResponseData) body, filterContext);
+                ((DescribeClientQuotasResponseFilter) filter).onDescribeClientQuotasResponse(apiVersion, header, (DescribeClientQuotasResponseData) body, filterContext);
+            case DESCRIBE_CLUSTER ->
+                ((DescribeClusterResponseFilter) filter).onDescribeClusterResponse(apiVersion, header, (DescribeClusterResponseData) body, filterContext);
+            case DESCRIBE_CONFIGS ->
+                ((DescribeConfigsResponseFilter) filter).onDescribeConfigsResponse(apiVersion, header, (DescribeConfigsResponseData) body, filterContext);
             case DESCRIBE_DELEGATION_TOKEN ->
-                ((DescribeDelegationTokenResponseFilter) filter).onDescribeDelegationTokenResponse(header, (DescribeDelegationTokenResponseData) body, filterContext);
-            case DESCRIBE_GROUPS -> ((DescribeGroupsResponseFilter) filter).onDescribeGroupsResponse(header, (DescribeGroupsResponseData) body, filterContext);
-            case DESCRIBE_LOG_DIRS -> ((DescribeLogDirsResponseFilter) filter).onDescribeLogDirsResponse(header, (DescribeLogDirsResponseData) body, filterContext);
+                ((DescribeDelegationTokenResponseFilter) filter).onDescribeDelegationTokenResponse(apiVersion, header, (DescribeDelegationTokenResponseData) body,
+                        filterContext);
+            case DESCRIBE_GROUPS ->
+                ((DescribeGroupsResponseFilter) filter).onDescribeGroupsResponse(apiVersion, header, (DescribeGroupsResponseData) body, filterContext);
+            case DESCRIBE_LOG_DIRS ->
+                ((DescribeLogDirsResponseFilter) filter).onDescribeLogDirsResponse(apiVersion, header, (DescribeLogDirsResponseData) body, filterContext);
             case DESCRIBE_PRODUCERS ->
-                ((DescribeProducersResponseFilter) filter).onDescribeProducersResponse(header, (DescribeProducersResponseData) body, filterContext);
-            case DESCRIBE_QUORUM -> ((DescribeQuorumResponseFilter) filter).onDescribeQuorumResponse(header, (DescribeQuorumResponseData) body, filterContext);
+                ((DescribeProducersResponseFilter) filter).onDescribeProducersResponse(apiVersion, header, (DescribeProducersResponseData) body, filterContext);
+            case DESCRIBE_QUORUM ->
+                ((DescribeQuorumResponseFilter) filter).onDescribeQuorumResponse(apiVersion, header, (DescribeQuorumResponseData) body, filterContext);
             case DESCRIBE_TRANSACTIONS ->
-                ((DescribeTransactionsResponseFilter) filter).onDescribeTransactionsResponse(header, (DescribeTransactionsResponseData) body, filterContext);
-            case DESCRIBE_USER_SCRAM_CREDENTIALS -> ((DescribeUserScramCredentialsResponseFilter) filter).onDescribeUserScramCredentialsResponse(header,
+                ((DescribeTransactionsResponseFilter) filter).onDescribeTransactionsResponse(apiVersion, header, (DescribeTransactionsResponseData) body, filterContext);
+            case DESCRIBE_USER_SCRAM_CREDENTIALS -> ((DescribeUserScramCredentialsResponseFilter) filter).onDescribeUserScramCredentialsResponse(apiVersion, header,
                     (DescribeUserScramCredentialsResponseData) body, filterContext);
-            case ELECT_LEADERS -> ((ElectLeadersResponseFilter) filter).onElectLeadersResponse(header, (ElectLeadersResponseData) body, filterContext);
-            case END_QUORUM_EPOCH -> ((EndQuorumEpochResponseFilter) filter).onEndQuorumEpochResponse(header, (EndQuorumEpochResponseData) body, filterContext);
-            case END_TXN -> ((EndTxnResponseFilter) filter).onEndTxnResponse(header, (EndTxnResponseData) body, filterContext);
-            case ENVELOPE -> ((EnvelopeResponseFilter) filter).onEnvelopeResponse(header, (EnvelopeResponseData) body, filterContext);
+            case ELECT_LEADERS -> ((ElectLeadersResponseFilter) filter).onElectLeadersResponse(apiVersion, header, (ElectLeadersResponseData) body, filterContext);
+            case END_QUORUM_EPOCH ->
+                ((EndQuorumEpochResponseFilter) filter).onEndQuorumEpochResponse(apiVersion, header, (EndQuorumEpochResponseData) body, filterContext);
+            case END_TXN -> ((EndTxnResponseFilter) filter).onEndTxnResponse(apiVersion, header, (EndTxnResponseData) body, filterContext);
+            case ENVELOPE -> ((EnvelopeResponseFilter) filter).onEnvelopeResponse(apiVersion, header, (EnvelopeResponseData) body, filterContext);
             case EXPIRE_DELEGATION_TOKEN ->
-                ((ExpireDelegationTokenResponseFilter) filter).onExpireDelegationTokenResponse(header, (ExpireDelegationTokenResponseData) body, filterContext);
-            case FETCH -> ((FetchResponseFilter) filter).onFetchResponse(header, (FetchResponseData) body, filterContext);
-            case FETCH_SNAPSHOT -> ((FetchSnapshotResponseFilter) filter).onFetchSnapshotResponse(header, (FetchSnapshotResponseData) body, filterContext);
-            case FIND_COORDINATOR -> ((FindCoordinatorResponseFilter) filter).onFindCoordinatorResponse(header, (FindCoordinatorResponseData) body, filterContext);
-            case HEARTBEAT -> ((HeartbeatResponseFilter) filter).onHeartbeatResponse(header, (HeartbeatResponseData) body, filterContext);
+                ((ExpireDelegationTokenResponseFilter) filter).onExpireDelegationTokenResponse(apiVersion, header, (ExpireDelegationTokenResponseData) body,
+                        filterContext);
+            case FETCH -> ((FetchResponseFilter) filter).onFetchResponse(apiVersion, header, (FetchResponseData) body, filterContext);
+            case FETCH_SNAPSHOT -> ((FetchSnapshotResponseFilter) filter).onFetchSnapshotResponse(apiVersion, header, (FetchSnapshotResponseData) body, filterContext);
+            case FIND_COORDINATOR ->
+                ((FindCoordinatorResponseFilter) filter).onFindCoordinatorResponse(apiVersion, header, (FindCoordinatorResponseData) body, filterContext);
+            case HEARTBEAT -> ((HeartbeatResponseFilter) filter).onHeartbeatResponse(apiVersion, header, (HeartbeatResponseData) body, filterContext);
             case INCREMENTAL_ALTER_CONFIGS ->
-                ((IncrementalAlterConfigsResponseFilter) filter).onIncrementalAlterConfigsResponse(header, (IncrementalAlterConfigsResponseData) body, filterContext);
-            case INIT_PRODUCER_ID -> ((InitProducerIdResponseFilter) filter).onInitProducerIdResponse(header, (InitProducerIdResponseData) body, filterContext);
-            case JOIN_GROUP -> ((JoinGroupResponseFilter) filter).onJoinGroupResponse(header, (JoinGroupResponseData) body, filterContext);
-            case LEADER_AND_ISR -> ((LeaderAndIsrResponseFilter) filter).onLeaderAndIsrResponse(header, (LeaderAndIsrResponseData) body, filterContext);
-            case LEAVE_GROUP -> ((LeaveGroupResponseFilter) filter).onLeaveGroupResponse(header, (LeaveGroupResponseData) body, filterContext);
-            case LIST_GROUPS -> ((ListGroupsResponseFilter) filter).onListGroupsResponse(header, (ListGroupsResponseData) body, filterContext);
-            case LIST_OFFSETS -> ((ListOffsetsResponseFilter) filter).onListOffsetsResponse(header, (ListOffsetsResponseData) body, filterContext);
-            case LIST_PARTITION_REASSIGNMENTS -> ((ListPartitionReassignmentsResponseFilter) filter).onListPartitionReassignmentsResponse(header,
+                ((IncrementalAlterConfigsResponseFilter) filter).onIncrementalAlterConfigsResponse(apiVersion, header, (IncrementalAlterConfigsResponseData) body,
+                        filterContext);
+            case INIT_PRODUCER_ID ->
+                ((InitProducerIdResponseFilter) filter).onInitProducerIdResponse(apiVersion, header, (InitProducerIdResponseData) body, filterContext);
+            case JOIN_GROUP -> ((JoinGroupResponseFilter) filter).onJoinGroupResponse(apiVersion, header, (JoinGroupResponseData) body, filterContext);
+            case LEADER_AND_ISR -> ((LeaderAndIsrResponseFilter) filter).onLeaderAndIsrResponse(apiVersion, header, (LeaderAndIsrResponseData) body, filterContext);
+            case LEAVE_GROUP -> ((LeaveGroupResponseFilter) filter).onLeaveGroupResponse(apiVersion, header, (LeaveGroupResponseData) body, filterContext);
+            case LIST_GROUPS -> ((ListGroupsResponseFilter) filter).onListGroupsResponse(apiVersion, header, (ListGroupsResponseData) body, filterContext);
+            case LIST_OFFSETS -> ((ListOffsetsResponseFilter) filter).onListOffsetsResponse(apiVersion, header, (ListOffsetsResponseData) body, filterContext);
+            case LIST_PARTITION_REASSIGNMENTS -> ((ListPartitionReassignmentsResponseFilter) filter).onListPartitionReassignmentsResponse(apiVersion, header,
                     (ListPartitionReassignmentsResponseData) body, filterContext);
-            case LIST_TRANSACTIONS -> ((ListTransactionsResponseFilter) filter).onListTransactionsResponse(header, (ListTransactionsResponseData) body, filterContext);
-            case METADATA -> ((MetadataResponseFilter) filter).onMetadataResponse(header, (MetadataResponseData) body, filterContext);
-            case OFFSET_COMMIT -> ((OffsetCommitResponseFilter) filter).onOffsetCommitResponse(header, (OffsetCommitResponseData) body, filterContext);
-            case OFFSET_DELETE -> ((OffsetDeleteResponseFilter) filter).onOffsetDeleteResponse(header, (OffsetDeleteResponseData) body, filterContext);
-            case OFFSET_FETCH -> ((OffsetFetchResponseFilter) filter).onOffsetFetchResponse(header, (OffsetFetchResponseData) body, filterContext);
+            case LIST_TRANSACTIONS ->
+                ((ListTransactionsResponseFilter) filter).onListTransactionsResponse(apiVersion, header, (ListTransactionsResponseData) body, filterContext);
+            case METADATA -> ((MetadataResponseFilter) filter).onMetadataResponse(apiVersion, header, (MetadataResponseData) body, filterContext);
+            case OFFSET_COMMIT -> ((OffsetCommitResponseFilter) filter).onOffsetCommitResponse(apiVersion, header, (OffsetCommitResponseData) body, filterContext);
+            case OFFSET_DELETE -> ((OffsetDeleteResponseFilter) filter).onOffsetDeleteResponse(apiVersion, header, (OffsetDeleteResponseData) body, filterContext);
+            case OFFSET_FETCH -> ((OffsetFetchResponseFilter) filter).onOffsetFetchResponse(apiVersion, header, (OffsetFetchResponseData) body, filterContext);
             case OFFSET_FOR_LEADER_EPOCH ->
-                ((OffsetForLeaderEpochResponseFilter) filter).onOffsetForLeaderEpochResponse(header, (OffsetForLeaderEpochResponseData) body, filterContext);
-            case PRODUCE -> ((ProduceResponseFilter) filter).onProduceResponse(header, (ProduceResponseData) body, filterContext);
+                ((OffsetForLeaderEpochResponseFilter) filter).onOffsetForLeaderEpochResponse(apiVersion, header, (OffsetForLeaderEpochResponseData) body, filterContext);
+            case PRODUCE -> ((ProduceResponseFilter) filter).onProduceResponse(apiVersion, header, (ProduceResponseData) body, filterContext);
             case RENEW_DELEGATION_TOKEN ->
-                ((RenewDelegationTokenResponseFilter) filter).onRenewDelegationTokenResponse(header, (RenewDelegationTokenResponseData) body, filterContext);
-            case SASL_AUTHENTICATE -> ((SaslAuthenticateResponseFilter) filter).onSaslAuthenticateResponse(header, (SaslAuthenticateResponseData) body, filterContext);
-            case SASL_HANDSHAKE -> ((SaslHandshakeResponseFilter) filter).onSaslHandshakeResponse(header, (SaslHandshakeResponseData) body, filterContext);
-            case STOP_REPLICA -> ((StopReplicaResponseFilter) filter).onStopReplicaResponse(header, (StopReplicaResponseData) body, filterContext);
-            case SYNC_GROUP -> ((SyncGroupResponseFilter) filter).onSyncGroupResponse(header, (SyncGroupResponseData) body, filterContext);
-            case TXN_OFFSET_COMMIT -> ((TxnOffsetCommitResponseFilter) filter).onTxnOffsetCommitResponse(header, (TxnOffsetCommitResponseData) body, filterContext);
-            case UNREGISTER_BROKER -> ((UnregisterBrokerResponseFilter) filter).onUnregisterBrokerResponse(header, (UnregisterBrokerResponseData) body, filterContext);
-            case UPDATE_FEATURES -> ((UpdateFeaturesResponseFilter) filter).onUpdateFeaturesResponse(header, (UpdateFeaturesResponseData) body, filterContext);
-            case UPDATE_METADATA -> ((UpdateMetadataResponseFilter) filter).onUpdateMetadataResponse(header, (UpdateMetadataResponseData) body, filterContext);
-            case VOTE -> ((VoteResponseFilter) filter).onVoteResponse(header, (VoteResponseData) body, filterContext);
-            case WRITE_TXN_MARKERS -> ((WriteTxnMarkersResponseFilter) filter).onWriteTxnMarkersResponse(header, (WriteTxnMarkersResponseData) body, filterContext);
+                ((RenewDelegationTokenResponseFilter) filter).onRenewDelegationTokenResponse(apiVersion, header, (RenewDelegationTokenResponseData) body, filterContext);
+            case SASL_AUTHENTICATE ->
+                ((SaslAuthenticateResponseFilter) filter).onSaslAuthenticateResponse(apiVersion, header, (SaslAuthenticateResponseData) body, filterContext);
+            case SASL_HANDSHAKE -> ((SaslHandshakeResponseFilter) filter).onSaslHandshakeResponse(apiVersion, header, (SaslHandshakeResponseData) body, filterContext);
+            case STOP_REPLICA -> ((StopReplicaResponseFilter) filter).onStopReplicaResponse(apiVersion, header, (StopReplicaResponseData) body, filterContext);
+            case SYNC_GROUP -> ((SyncGroupResponseFilter) filter).onSyncGroupResponse(apiVersion, header, (SyncGroupResponseData) body, filterContext);
+            case TXN_OFFSET_COMMIT ->
+                ((TxnOffsetCommitResponseFilter) filter).onTxnOffsetCommitResponse(apiVersion, header, (TxnOffsetCommitResponseData) body, filterContext);
+            case UNREGISTER_BROKER ->
+                ((UnregisterBrokerResponseFilter) filter).onUnregisterBrokerResponse(apiVersion, header, (UnregisterBrokerResponseData) body, filterContext);
+            case UPDATE_FEATURES ->
+                ((UpdateFeaturesResponseFilter) filter).onUpdateFeaturesResponse(apiVersion, header, (UpdateFeaturesResponseData) body, filterContext);
+            case UPDATE_METADATA ->
+                ((UpdateMetadataResponseFilter) filter).onUpdateMetadataResponse(apiVersion, header, (UpdateMetadataResponseData) body, filterContext);
+            case VOTE -> ((VoteResponseFilter) filter).onVoteResponse(apiVersion, header, (VoteResponseData) body, filterContext);
+            case WRITE_TXN_MARKERS ->
+                ((WriteTxnMarkersResponseFilter) filter).onWriteTxnMarkersResponse(apiVersion, header, (WriteTxnMarkersResponseData) body, filterContext);
             default -> throw new IllegalStateException("Unsupported RPC " + apiKey);
         }
     }
diff --git a/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter.java b/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter.java
index ea1615d9..326404fd 100644
--- a/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter.java
@@ -18,10 +18,10 @@ import io.kroxylicious.proxy.filter.ProduceResponseFilter;
 public class TwoInterfaceFilter implements ProduceResponseFilter, ProduceRequestFilter {
 
     @Override
-    public void onProduceRequest(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
     }
 
     @Override
-    public void onProduceResponse(ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter0.java b/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter0.java
index d3aac0a1..9b9d2059 100644
--- a/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter0.java
+++ b/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter0.java
@@ -18,10 +18,10 @@ import io.kroxylicious.proxy.filter.ProduceResponseFilter;
 public class TwoInterfaceFilter0 implements ProduceResponseFilter, ProduceRequestFilter {
 
     @Override
-    public void onProduceRequest(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
     }
 
     @Override
-    public void onProduceResponse(ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
     }
 }
diff --git a/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter1.java b/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter1.java
index 3966d923..6cb67618 100644
--- a/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter1.java
+++ b/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter1.java
@@ -18,10 +18,10 @@ import io.kroxylicious.proxy.filter.ProduceResponseFilter;
 public class TwoInterfaceFilter1 implements ProduceResponseFilter, ProduceRequestFilter {
 
     @Override
-    public void onProduceRequest(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
     }
 
     @Override
-    public void onProduceResponse(ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
     }
 }