Repository: kroxylicious/kroxylicious
Pull request #343 (2023-05-29 04:03:27)

Bump netty.version from 4.1.92.Final to 4.1.93.Final
Bumps `netty.version` from 4.1.92.Final to 4.1.93.Final.
Updates `netty-bom` from 4.1.92.Final to 4.1.93.Final
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/netty/netty/commit/69270b102a6339ef3279e3f0755526db001850ef"><code>69270b1</code></a> [maven-release-plugin] prepare release netty-4.1.93.Final</li>
<li><a href="https://github.com/netty/netty/commit/4eda913d998898d8489565904b2a5d133d081413"><code>4eda913</code></a> Respect the number of bytes read per datagram when using recvmmsg (<a href="https://redirect.github.com/netty/netty/issues/13399">#13399</a>)</li>
<li><a href="https://github.com/netty/netty/commit/57d536c59e434e88c0d8374d6b03671edb02885f"><code>57d536c</code></a> HTTP/2 don't propagate <code>ChannelInputShutdownEvent</code> to active streams (<a href="https://redirect.github.com/netty/netty/issues/13395">#13395</a>)</li>
<li><a href="https://github.com/netty/netty/commit/3d8ff07b4ad60aa19d9630e75833881298d201d7"><code>3d8ff07</code></a> HTTP/2 should forward shutdown user events to active streams (<a href="https://redirect.github.com/netty/netty/issues/13394">#13394</a>)</li>
<li><a href="https://github.com/netty/netty/commit/fc58d2dd1b9a155c55b56787fb5ae7719ffb83cf"><code>fc58d2d</code></a> HTTP/2 should forward shutdown user events to active streams (<a href="https://redirect.github.com/netty/netty/issues/13394">#13394</a>)</li>
<li><a href="https://github.com/netty/netty/commit/2822dfdba3651cfda433dcb1842b98f5580b05d5"><code>2822dfd</code></a> fix instant RST/FIN after SYN/SYN-ACK/ACK (<a href="https://redirect.github.com/netty/netty/issues/12554">#12554</a>)</li>
<li><a href="https://github.com/netty/netty/commit/92f67bf6da68e7944484564d9aac2d0e9bbacc65"><code>92f67bf</code></a> Fix possible NPE when calling PoolArena.toString() (<a href="https://redirect.github.com/netty/netty/issues/13393">#13393</a>)</li>
<li><a href="https://github.com/netty/netty/commit/0e8b0252fb595ccf7c4fdc7cce461264c68b4b1c"><code>0e8b025</code></a> Channel#bytesBefore[un]writable off by 1 (<a href="https://redirect.github.com/netty/netty/issues/13389">#13389</a>)</li>
<li><a href="https://github.com/netty/netty/commit/f76d6462412fa5c2f6baf82b216356196ebb10bd"><code>f76d646</code></a> H2/H2C server stream channels deactivated while write still in progress (<a href="https://redirect.github.com/netty/netty/issues/13388">#13388</a>)</li>
<li><a href="https://github.com/netty/netty/commit/439e9074b4022e96881d82c976dd6df7d4c3a701"><code>439e907</code></a> Set top level permission to remaining workflows (<a href="https://redirect.github.com/netty/netty/issues/13385">#13385</a>)</li>
<li>Additional commits viewable in <a href="https://github.com/netty/netty/compare/netty-4.1.92.Final...netty-4.1.93.Final">compare view</a></li>
</ul>
</details>
<br />

Updates `netty-codec-http` from 4.1.92.Final to 4.1.93.Final
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/netty/netty/commit/69270b102a6339ef3279e3f0755526db001850ef"><code>69270b1</code></a> [maven-release-plugin] prepare release netty-4.1.93.Final</li>
<li><a href="https://github.com/netty/netty/commit/4eda913d998898d8489565904b2a5d133d081413"><code>4eda913</code></a> Respect the number of bytes read per datagram when using recvmmsg (<a href="https://redirect.github.com/netty/netty/issues/13399">#13399</a>)</li>
<li><a href="https://github.com/netty/netty/commit/57d536c59e434e88c0d8374d6b03671edb02885f"><code>57d536c</code></a> HTTP/2 don't propagate <code>ChannelInputShutdownEvent</code> to active streams (<a href="https://redirect.github.com/netty/netty/issues/13395">#13395</a>)</li>
<li><a href="https://github.com/netty/netty/commit/3d8ff07b4ad60aa19d9630e75833881298d201d7"><code>3d8ff07</code></a> HTTP/2 should forward shutdown user events to active streams (<a href="https://redirect.github.com/netty/netty/issues/13394">#13394</a>)</li>
<li><a href="https://github.com/netty/netty/commit/fc58d2dd1b9a155c55b56787fb5ae7719ffb83cf"><code>fc58d2d</code></a> HTTP/2 should forward shutdown user events to active streams (<a href="https://redirect.github.com/netty/netty/issues/13394">#13394</a>)</li>
<li><a href="https://github.com/netty/netty/commit/2822dfdba3651cfda433dcb1842b98f5580b05d5"><code>2822dfd</code></a> fix instant RST/FIN after SYN/SYN-ACK/ACK (<a href="https://redirect.github.com/netty/netty/issues/12554">#12554</a>)</li>
<li><a href="https://github.com/netty/netty/commit/92f67bf6da68e7944484564d9aac2d0e9bbacc65"><code>92f67bf</code></a> Fix possible NPE when calling PoolArena.toString() (<a href="https://redirect.github.com/netty/netty/issues/13393">#13393</a>)</li>
<li><a href="https://github.com/netty/netty/commit/0e8b0252fb595ccf7c4fdc7cce461264c68b4b1c"><code>0e8b025</code></a> Channel#bytesBefore[un]writable off by 1 (<a href="https://redirect.github.com/netty/netty/issues/13389">#13389</a>)</li>
<li><a href="https://github.com/netty/netty/commit/f76d6462412fa5c2f6baf82b216356196ebb10bd"><code>f76d646</code></a> H2/H2C server stream channels deactivated while write still in progress (<a href="https://redirect.github.com/netty/netty/issues/13388">#13388</a>)</li>
<li><a href="https://github.com/netty/netty/commit/439e9074b4022e96881d82c976dd6df7d4c3a701"><code>439e907</code></a> Set top level permission to remaining workflows (<a href="https://redirect.github.com/netty/netty/issues/13385">#13385</a>)</li>
<li>Additional commits viewable in <a href="https://github.com/netty/netty/compare/netty-4.1.92.Final...netty-4.1.93.Final">compare view</a></li>
</ul>
</details>
<br />

Updates `netty-codec-haproxy` from 4.1.92.Final to 4.1.93.Final
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/netty/netty/commit/69270b102a6339ef3279e3f0755526db001850ef"><code>69270b1</code></a> [maven-release-plugin] prepare release netty-4.1.93.Final</li>
<li><a href="https://github.com/netty/netty/commit/4eda913d998898d8489565904b2a5d133d081413"><code>4eda913</code></a> Respect the number of bytes read per datagram when using recvmmsg (<a href="https://redirect.github.com/netty/netty/issues/13399">#13399</a>)</li>
<li><a href="https://github.com/netty/netty/commit/57d536c59e434e88c0d8374d6b03671edb02885f"><code>57d536c</code></a> HTTP/2 don't propagate <code>ChannelInputShutdownEvent</code> to active streams (<a href="https://redirect.github.com/netty/netty/issues/13395">#13395</a>)</li>
<li><a href="https://github.com/netty/netty/commit/3d8ff07b4ad60aa19d9630e75833881298d201d7"><code>3d8ff07</code></a> HTTP/2 should forward shutdown user events to active streams (<a href="https://redirect.github.com/netty/netty/issues/13394">#13394</a>)</li>
<li><a href="https://github.com/netty/netty/commit/fc58d2dd1b9a155c55b56787fb5ae7719ffb83cf"><code>fc58d2d</code></a> HTTP/2 should forward shutdown user events to active streams (<a href="https://redirect.github.com/netty/netty/issues/13394">#13394</a>)</li>
<li><a href="https://github.com/netty/netty/commit/2822dfdba3651cfda433dcb1842b98f5580b05d5"><code>2822dfd</code></a> fix instant RST/FIN after SYN/SYN-ACK/ACK (<a href="https://redirect.github.com/netty/netty/issues/12554">#12554</a>)</li>
<li><a href="https://github.com/netty/netty/commit/92f67bf6da68e7944484564d9aac2d0e9bbacc65"><code>92f67bf</code></a> Fix possible NPE when calling PoolArena.toString() (<a href="https://redirect.github.com/netty/netty/issues/13393">#13393</a>)</li>
<li><a href="https://github.com/netty/netty/commit/0e8b0252fb595ccf7c4fdc7cce461264c68b4b1c"><code>0e8b025</code></a> Channel#bytesBefore[un]writable off by 1 (<a href="https://redirect.github.com/netty/netty/issues/13389">#13389</a>)</li>
<li><a href="https://github.com/netty/netty/commit/f76d6462412fa5c2f6baf82b216356196ebb10bd"><code>f76d646</code></a> H2/H2C server stream channels deactivated while write still in progress (<a href="https://redirect.github.com/netty/netty/issues/13388">#13388</a>)</li>
<li><a href="https://github.com/netty/netty/commit/439e9074b4022e96881d82c976dd6df7d4c3a701"><code>439e907</code></a> Set top level permission to remaining workflows (<a href="https://redirect.github.com/netty/netty/issues/13385">#13385</a>)</li>
<li>Additional commits viewable in <a href="https://github.com/netty/netty/compare/netty-4.1.92.Final...netty-4.1.93.Final">compare view</a></li>
</ul>
</details>
<br />


Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


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 a1d23a60..cab78e60 100644
--- a/pom.xml
+++ b/pom.xml
@@ -31,19 +31,18 @@
         <skipUTs>${skipTests}</skipUTs>
 
         <freemarker.version>2.3.32</freemarker.version>
-        <guava.version>31.1-jre</guava.version>
+        <guava.version>32.0.0-jre</guava.version>
         <jackson.version>2.15.1</jackson.version>
         <junit.version>5.9.3</junit.version>
         <kafka.version>3.4.0</kafka.version>
         <kroxy.extension.version>0.3.0</kroxy.extension.version>
         <log4j.version>2.20.0</log4j.version>
         <picocli.version>4.7.3</picocli.version>
-        <netty.version>4.1.92.Final</netty.version>
+        <netty.version>4.1.93.Final</netty.version>
         <netty.io_uring.version>0.0.21.Final</netty.io_uring.version>
         <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>
@@ -315,7 +308,7 @@
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-checkstyle-plugin</artifactId>
-                    <version>3.2.2</version>
+                    <version>3.3.0</version>
                 </plugin>
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
@@ -450,7 +443,7 @@
             <plugin>
                 <groupId>org.asciidoctor</groupId>
                 <artifactId>asciidoctor-maven-plugin</artifactId>
-                <version>2.2.3</version>
+                <version>2.2.4</version>
                 <executions>
                     <execution>
                         <id>convert-to-html</id>


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) {
     }
 }