Repository: kroxylicious/kroxylicious
Pull request #142 (2023-02-19 20:48:32)

Add header data to filter APIs and remove frame usages from KrpcFilter
Why:
1. We want to extract a filter-api for Filter Authors to build new filters against. We want this to have minimal dependencies and be focused on the Kafka apis. So the filter-api should not depend on the Frame concept. By passing through the components of the Frame (which are all classes provided by the  kafka-clients) we remove this dependency.
2. We want to react to (and modify in future) header data. This was not exposed to the per-message filters yet. With this change the generated signatures now contain a `RequestHeaderData` or `ResponseHeaderData` parameter, depending on what type of api they are.

Contributes to https://github.com/kroxylicious/kroxylicious/issues/133

k-wall: I think this change looks good.  I think exposing the `(Request|Response)HeaderData` to the `on*()` methods implementation is the right thing to do at the API level.   The `forward*()` methods will need to accept the `(Request|Response)HeaderData` but I think leaving this to a following PR is okay.

One interesting question is the fact that only parts of the HeaderData are mutable.  Filter authors shouldn't be able to mess with the correlation id (which is manipulated by kroxy itself) and the api keys.    Mutating clientid, tagged fields and version should be allowed. I wonder how kroxy should enforce that?    I suppose Kroxy could expose a RHD proxy with the banned setMethods overridden to throw an exception, but the overhead (GC pressire) might be too high, especially as most filter writers probably won't want to mutate the header.    I got me thinking is there some way the forward*() method could reject/ignore 'illegal' changes.







Patch:
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 83e98166..19fc8fcf 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
@@ -5,6 +5,8 @@
  */
 package io.kroxylicious.proxy.example.authn;
 
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.message.SaslAuthenticateRequestData;
 import org.apache.kafka.common.message.SaslAuthenticateResponseData;
 import org.apache.kafka.common.message.SaslHandshakeRequestData;
@@ -29,14 +31,15 @@ public class SaslAuthnObserver
     private long sessionLifetimeMs;
 
     @Override
-    public void onSaslHandshakeRequest(SaslHandshakeRequestData request,
+    public void onSaslHandshakeRequest(RequestHeaderData header,
+                                       SaslHandshakeRequestData request,
                                        KrpcFilterContext context) {
         this.mechanism = request.mechanism();
         context.forwardRequest(request);
     }
 
     @Override
-    public void onSaslHandshakeResponse(SaslHandshakeResponseData response,
+    public void onSaslHandshakeResponse(ResponseHeaderData header, SaslHandshakeResponseData response,
                                         KrpcFilterContext context) {
         if (response.errorCode() != Errors.NONE.code()) {
             this.mechanism = null;
@@ -45,7 +48,8 @@ public class SaslAuthnObserver
     }
 
     @Override
-    public void onSaslAuthenticateRequest(SaslAuthenticateRequestData request,
+    public void onSaslAuthenticateRequest(RequestHeaderData header,
+                                          SaslAuthenticateRequestData request,
                                           KrpcFilterContext context) {
         byte[] bytes = request.authBytes();
         switch (mechanism) {
@@ -63,7 +67,7 @@ public class SaslAuthnObserver
     }
 
     @Override
-    public void onSaslAuthenticateResponse(SaslAuthenticateResponseData response,
+    public void onSaslAuthenticateResponse(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 d117d008..2cf9cf3a 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
@@ -8,6 +8,8 @@ package io.kroxylicious.proxy.example.topicencryption;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
 
 import io.kroxylicious.proxy.filter.FetchResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
@@ -19,7 +21,7 @@ public class TopicEncryption implements ProduceRequestFilter, FetchResponseFilte
     // but other filters will be interested in keeping track of metadata
 
     @Override
-    public void onProduceRequest(ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
         boolean fragmented = false;
         if (fragmented) {
             // TODO forward the fragments
@@ -33,7 +35,7 @@ public class TopicEncryption implements ProduceRequestFilter, FetchResponseFilte
     }
 
     @Override
-    public void onFetchResponse(FetchResponseData response, KrpcFilterContext context) {
+    public void onFetchResponse(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 620b872d..f4b98354 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
@@ -12,6 +12,7 @@ import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.message.CreateTopicsResponseData;
 import org.apache.kafka.common.message.DeleteTopicsResponseData;
 import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.message.ResponseHeaderData;
 
 import io.kroxylicious.proxy.filter.CreateTopicsResponseFilter;
 import io.kroxylicious.proxy.filter.DeleteTopicsResponseFilter;
@@ -27,7 +28,7 @@ public class TopicNameFilter
     private final Map<Uuid, String> topicNames = new HashMap<>();
 
     @Override
-    public void onMetadataResponse(MetadataResponseData response, KrpcFilterContext context) {
+    public void onMetadataResponse(ResponseHeaderData header, MetadataResponseData response, KrpcFilterContext context) {
         if (response.topics() != null) {
             for (var topic : response.topics()) {
                 topicNames.put(topic.topicId(), topic.name());
@@ -41,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(DeleteTopicsResponseData response, KrpcFilterContext context) {
+    public void onDeleteTopicsResponse(ResponseHeaderData header, DeleteTopicsResponseData response, KrpcFilterContext context) {
         for (var resp : response.responses()) {
             topicNames.remove(resp.topicId());
         }
@@ -49,7 +50,7 @@ public class TopicNameFilter
     }
 
     @Override
-    public void onCreateTopicsResponse(CreateTopicsResponseData response, KrpcFilterContext context) {
+    public void onCreateTopicsResponse(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/filter/multitenant/MultiTenantTransformationFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
index 96b81c9c..e4c8065a 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
@@ -26,6 +26,8 @@ import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData;
 import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData;
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceResponseData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -77,31 +79,31 @@ public class MultiTenantTransformationFilter
     private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantTransformationFilter.class);
 
     @Override
-    public void onCreateTopicsRequest(CreateTopicsRequestData request, KrpcFilterContext context) {
+    public void onCreateTopicsRequest(RequestHeaderData data, CreateTopicsRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardRequest(request);
     }
 
     @Override
-    public void onCreateTopicsResponse(CreateTopicsResponseData response, KrpcFilterContext context) {
+    public void onCreateTopicsResponse(ResponseHeaderData data, CreateTopicsResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardResponse(response);
     }
 
     @Override
-    public void onDeleteTopicsRequest(DeleteTopicsRequestData request, KrpcFilterContext context) {
+    public void onDeleteTopicsRequest(RequestHeaderData data, DeleteTopicsRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, topic.topicId() != null));
         context.forwardRequest(request);
     }
 
     @Override
-    public void onDeleteTopicsResponse(DeleteTopicsResponseData response, KrpcFilterContext context) {
+    public void onDeleteTopicsResponse(ResponseHeaderData data, DeleteTopicsResponseData response, KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardResponse(response);
     }
 
     @Override
-    public void onMetadataRequest(MetadataRequestData request, KrpcFilterContext context) {
+    public void onMetadataRequest(RequestHeaderData data, 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));
@@ -111,7 +113,7 @@ public class MultiTenantTransformationFilter
     }
 
     @Override
-    public void onMetadataResponse(MetadataResponseData response, KrpcFilterContext context) {
+    public void onMetadataResponse(ResponseHeaderData data, 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));
@@ -119,77 +121,77 @@ public class MultiTenantTransformationFilter
     }
 
     @Override
-    public void onProduceRequest(ProduceRequestData request, KrpcFilterContext context) {
+    public void onProduceRequest(RequestHeaderData data, ProduceRequestData request, KrpcFilterContext context) {
         request.topicData().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardRequest(request);
 
     }
 
     @Override
-    public void onProduceResponse(ProduceResponseData response, KrpcFilterContext context) {
+    public void onProduceResponse(ResponseHeaderData data, ProduceResponseData response, KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardResponse(response);
 
     }
 
     @Override
-    public void onListOffsetsRequest(ListOffsetsRequestData request, KrpcFilterContext context) {
+    public void onListOffsetsRequest(RequestHeaderData data, ListOffsetsRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardRequest(request);
     }
 
     @Override
-    public void onListOffsetsResponse(ListOffsetsResponseData response, KrpcFilterContext context) {
+    public void onListOffsetsResponse(ResponseHeaderData data, ListOffsetsResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardResponse(response);
     }
 
     @Override
-    public void onOffsetFetchRequest(OffsetFetchRequestData request, KrpcFilterContext context) {
+    public void onOffsetFetchRequest(RequestHeaderData data, OffsetFetchRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         request.groups().forEach(requestGroup -> requestGroup.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false)));
         context.forwardRequest(request);
     }
 
     @Override
-    public void onOffsetFetchResponse(OffsetFetchResponseData response, KrpcFilterContext context) {
+    public void onOffsetFetchResponse(ResponseHeaderData data, OffsetFetchResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         response.groups().forEach(responseGroup -> responseGroup.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false)));
         context.forwardResponse(response);
     }
 
     @Override
-    public void onOffsetForLeaderEpochRequest(OffsetForLeaderEpochRequestData request, KrpcFilterContext context) {
+    public void onOffsetForLeaderEpochRequest(RequestHeaderData data, OffsetForLeaderEpochRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::topic, topic::setTopic, false));
         context.forwardRequest(request);
     }
 
     @Override
-    public void onOffsetForLeaderEpochResponse(OffsetForLeaderEpochResponseData response, KrpcFilterContext context) {
+    public void onOffsetForLeaderEpochResponse(ResponseHeaderData data, OffsetForLeaderEpochResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::topic, topic::setTopic, false));
         context.forwardResponse(response);
     }
 
     @Override
-    public void onOffsetCommitRequest(OffsetCommitRequestData request, KrpcFilterContext context) {
+    public void onOffsetCommitRequest(RequestHeaderData data, OffsetCommitRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardRequest(request);
     }
 
     @Override
-    public void onOffsetCommitResponse(OffsetCommitResponseData response, KrpcFilterContext context) {
+    public void onOffsetCommitResponse(ResponseHeaderData data, OffsetCommitResponseData response, KrpcFilterContext context) {
         response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
         context.forwardResponse(response);
     }
 
     @Override
-    public void onFetchRequest(FetchRequestData request, KrpcFilterContext context) {
+    public void onFetchRequest(RequestHeaderData data, FetchRequestData request, KrpcFilterContext context) {
         request.topics().forEach(topic -> applyTenantPrefix(context, topic::topic, topic::setTopic, topic.topicId() != null));
         context.forwardRequest(request);
     }
 
     @Override
-    public void onFetchResponse(FetchResponseData response, KrpcFilterContext context) {
+    public void onFetchResponse(ResponseHeaderData data, FetchResponseData response, KrpcFilterContext context) {
         response.responses().forEach(topic -> removeTenantPrefix(context, topic::topic, topic::setTopic, topic.topicId() != null));
         context.forwardResponse(response);
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
index 8a5cd9a1..30439857 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
@@ -57,7 +57,7 @@ public class FilterHandler
                     LOGGER.debug("{}: Dispatching downstream {} request to filter{}: {}",
                             ctx.channel(), decodedFrame.apiKey(), filterDescriptor(), msg);
                 }
-                filter.onRequest(decodedFrame, filterContext);
+                filter.onRequest(decodedFrame.apiKey(), decodedFrame.header(), decodedFrame.body(), filterContext);
             }
             else {
                 ctx.write(msg, promise);
@@ -102,7 +102,7 @@ public class FilterHandler
                     LOGGER.debug("{}: Dispatching upstream {} response to filter {}: {}",
                             ctx.channel(), decodedFrame.apiKey(), filterDescriptor(), msg);
                 }
-                filter.onResponse(decodedFrame, filterContext);
+                filter.onResponse(decodedFrame.apiKey(), decodedFrame.header(), decodedFrame.body(), filterContext);
             }
             else {
                 ctx.fireChannelRead(msg);
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 5a0b4459..8989ee4f 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
@@ -6,6 +6,7 @@
 package io.kroxylicious.proxy.internal.filter;
 
 import org.apache.kafka.common.message.ApiVersionsResponseData;
+import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -66,7 +67,7 @@ public class ApiVersionsFilter implements ApiVersionsResponseFilter {
     }
 
     @Override
-    public void onApiVersionsResponse(ApiVersionsResponseData data, KrpcFilterContext context) {
+    public void onApiVersionsResponse(ResponseHeaderData header, ApiVersionsResponseData data, KrpcFilterContext context) {
         intersectApiVersions(context.channelDescriptor(), data);
         context.forwardResponse(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 222930a9..fe68f271 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
@@ -17,6 +17,7 @@ import org.apache.kafka.common.message.FindCoordinatorResponseData;
 import org.apache.kafka.common.message.FindCoordinatorResponseData.Coordinator;
 import org.apache.kafka.common.message.MetadataResponseData;
 import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseBroker;
+import org.apache.kafka.common.message.ResponseHeaderData;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -66,7 +67,7 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
     }
 
     @Override
-    public void onMetadataResponse(MetadataResponseData data, KrpcFilterContext context) {
+    public void onMetadataResponse(ResponseHeaderData header, MetadataResponseData data, KrpcFilterContext context) {
         for (MetadataResponseBroker broker : data.brokers()) {
             apply(context, broker, MetadataResponseBroker::host, MetadataResponseBroker::port, MetadataResponseBroker::setHost, MetadataResponseBroker::setPort);
         }
@@ -74,7 +75,7 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
     }
 
     @Override
-    public void onDescribeClusterResponse(DescribeClusterResponseData data, KrpcFilterContext context) {
+    public void onDescribeClusterResponse(ResponseHeaderData header, DescribeClusterResponseData data, KrpcFilterContext context) {
         for (DescribeClusterBroker broker : data.brokers()) {
             apply(context, broker, DescribeClusterBroker::host, DescribeClusterBroker::port, DescribeClusterBroker::setHost, DescribeClusterBroker::setPort);
         }
@@ -82,7 +83,7 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
     }
 
     @Override
-    public void onFindCoordinatorResponse(FindCoordinatorResponseData data, KrpcFilterContext context) {
+    public void onFindCoordinatorResponse(ResponseHeaderData header, FindCoordinatorResponseData data, KrpcFilterContext context) {
         for (Coordinator coordinator : data.coordinators()) {
             apply(context, coordinator, Coordinator::host, Coordinator::port, Coordinator::setHost, Coordinator::setPort);
         }
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 dd00b2a3..a6474a80 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
@@ -17,6 +17,7 @@ import org.apache.kafka.common.message.FetchResponseData.FetchableTopicResponse;
 import org.apache.kafka.common.message.FetchResponseData.PartitionData;
 import org.apache.kafka.common.message.MetadataRequestData;
 import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.record.CompressionType;
 import org.apache.kafka.common.record.MemoryRecords;
@@ -70,7 +71,7 @@ public class FetchResponseTransformationFilter implements FetchResponseFilter {
     }
 
     @Override
-    public void onFetchResponse(FetchResponseData fetchResponse, KrpcFilterContext context) {
+    public void onFetchResponse(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 2c9f580b..11c88420 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
@@ -12,6 +12,7 @@ import java.util.Iterator;
 
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceRequestData.PartitionProduceData;
+import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.record.CompressionType;
 import org.apache.kafka.common.record.MemoryRecords;
@@ -68,7 +69,7 @@ public class ProduceRequestTransformationFilter implements ProduceRequestFilter
     }
 
     @Override
-    public void onProduceRequest(ProduceRequestData data, KrpcFilterContext context) {
+    public void onProduceRequest(RequestHeaderData header, ProduceRequestData data, KrpcFilterContext context) {
         applyTransformation(context, data);
         context.forwardRequest(data);
     }
diff --git a/kroxylicious/src/main/templates/Kproxy/Filter.ftl b/kroxylicious/src/main/templates/Kproxy/Filter.ftl
index 9f867916..e21bda1b 100644
--- a/kroxylicious/src/main/templates/Kproxy/Filter.ftl
+++ b/kroxylicious/src/main/templates/Kproxy/Filter.ftl
@@ -29,6 +29,11 @@
 package io.kroxylicious.proxy.filter;
 
 import org.apache.kafka.common.message.${messageSpec.name}Data;
+<#if messageSpec.type?lower_case == 'response'>
+import org.apache.kafka.common.message.ResponseHeaderData;
+<#else>
+import org.apache.kafka.common.message.RequestHeaderData;
+</#if>
 
 /**
  * A stateless filter for ${messageSpec.name}s.
@@ -45,6 +50,6 @@ public interface ${filterClass} extends KrpcFilter {
      * @param ${msgType} The KRPC message to handle.
      * @param context The context.
      */
-    public void on${messageSpec.name}(${dataClass} ${msgType}, KrpcFilterContext context);
+    public void on${messageSpec.name}(<#if messageSpec.type?lower_case == 'response'>ResponseHeaderData<#else>RequestHeaderData</#if> header, ${dataClass} ${msgType}, KrpcFilterContext context);
 
 }
\ No newline at end of file
diff --git a/kroxylicious/src/main/templates/Kproxy/KrpcFilter.ftl b/kroxylicious/src/main/templates/Kproxy/KrpcFilter.ftl
index 8b1b5e49..f3b33994 100644
--- a/kroxylicious/src/main/templates/Kproxy/KrpcFilter.ftl
+++ b/kroxylicious/src/main/templates/Kproxy/KrpcFilter.ftl
@@ -26,10 +26,10 @@ package ${outputPackage};
 <#list messageSpecs as messageSpec>
 import org.apache.kafka.common.message.${messageSpec.name}Data;
 </#list>
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
-
-import io.kroxylicious.proxy.frame.DecodedRequestFrame;
-import io.kroxylicious.proxy.frame.DecodedResponseFrame;
+import org.apache.kafka.common.protocol.ApiMessage;
 
 /**
  * <p>Interface for {@code *RequestFilter}s.
@@ -46,7 +46,7 @@ import io.kroxylicious.proxy.frame.DecodedResponseFrame;
  * <ol>
  *     <li>That each instance of the filter is associated with a single channel</li>
  *     <li>That {@link #shouldDeserializeRequest(ApiKeys, short)} and
- *     {@link #onRequest(DecodedRequestFrame, KrpcFilterContext)} (or {@code on*Request} as appropriate)
+ *     {@link #onRequest(ApiKeys, RequestHeaderData, ApiMessage, KrpcFilterContext)} (or {@code on*Request} as appropriate)
  *     will always be invoked on the same thread.</li>
  *     <li>That filters are applied in the order they were configured.</li>
  * </ol>
@@ -62,42 +62,50 @@ import io.kroxylicious.proxy.frame.DecodedResponseFrame;
 public /* sealed */ interface KrpcFilter /* TODO permits ... */ {
 
     /**
-     * Apply the filter to the given {@code decodedFrame} using the given {@code filterContext}.
-     * @param decodedFrame The request frame.
+     * Apply the filter to the given {@code header} and {@code body} using the given {@code filterContext}.
+     * @param apiKey The request api key.
+     * @param header The request header.
+     * @param body The request body.
      * @param filterContext The filter context.
      */
-    public default void onRequest(DecodedRequestFrame<?> decodedFrame,
+    public default void onRequest(ApiKeys apiKey,
+                                  RequestHeaderData header,
+                                  ApiMessage body,
                                   KrpcFilterContext filterContext) {
-        switch (decodedFrame.apiKey()) {
+        switch (apiKey) {
 <#list messageSpecs as messageSpec>
 <#if messageSpec.type?lower_case == 'request'>
             case ${retrieveApiKey(messageSpec)}:
-                ((${messageSpec.name}Filter) this).on${messageSpec.name}((${messageSpec.name}Data) decodedFrame.body(), filterContext);
+                ((${messageSpec.name}Filter) this).on${messageSpec.name}(header, (${messageSpec.name}Data) body, filterContext);
                 break;
 </#if>
 </#list>
             default:
-                throw new IllegalStateException("Unsupported RPC " + decodedFrame.apiKey());
+                throw new IllegalStateException("Unsupported RPC " + apiKey);
         }
     }
 
     /**
-     * Apply the filter to the given {@code decodedFrame} using the given {@code filterContext}.
-     * @param decodedFrame The response frame.
+     * Apply the filter to the given {@code header} and {@code body} using the given {@code filterContext}.
+     * @param apiKey The request api key.
+     * @param header The request header.
+     * @param body The request body.
      * @param filterContext The filter context.
      */
-    public default void onResponse(DecodedResponseFrame<?> decodedFrame,
+    public default void onResponse(ApiKeys apiKey,
+                                   ResponseHeaderData header,
+                                   ApiMessage body,
                                    KrpcFilterContext filterContext) {
-        switch (decodedFrame.apiKey()) {
+        switch (apiKey) {
 <#list messageSpecs as messageSpec>
 <#if messageSpec.type?lower_case == 'response'>
             case ${retrieveApiKey(messageSpec)}:
-                ((${messageSpec.name}Filter) this).on${messageSpec.name}((${messageSpec.name}Data) decodedFrame.body(), filterContext);
+                ((${messageSpec.name}Filter) this).on${messageSpec.name}(header, (${messageSpec.name}Data) body, filterContext);
                 break;
 </#if>
 </#list>
             default:
-                throw new IllegalStateException("Unsupported RPC " + decodedFrame.apiKey());
+                throw new IllegalStateException("Unsupported RPC " + apiKey);
         }
     }
 


Test patch:
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 959134df..890e095d 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
@@ -11,6 +11,8 @@ import org.apache.kafka.common.message.ApiVersionsResponseData;
 import org.apache.kafka.common.message.FetchRequestData;
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ApiMessage;
 import org.junit.jupiter.api.Test;
@@ -31,7 +33,7 @@ public class FilterHandlerTest extends FilterHarness {
 
     @Test
     public void testForwardRequest() {
-        ApiVersionsRequestFilter filter = (request, context) -> context.forwardRequest(request);
+        ApiVersionsRequestFilter filter = (header, request, context) -> context.forwardRequest(request);
         buildChannel(filter);
         var frame = writeRequest(new ApiVersionsRequestData());
         var propagated = channel.readOutbound();
@@ -47,7 +49,7 @@ public class FilterHandlerTest extends FilterHarness {
             }
 
             @Override
-            public void onApiVersionsRequest(ApiVersionsRequestData request, KrpcFilterContext context) {
+            public void onApiVersionsRequest(RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
                 fail("Should not be called");
             }
         };
@@ -59,7 +61,7 @@ public class FilterHandlerTest extends FilterHarness {
 
     @Test
     public void testDropRequest() {
-        ApiVersionsRequestFilter filter = (request, context) -> {
+        ApiVersionsRequestFilter filter = (header, request, context) -> {
             /* don't call forwardRequest => drop the request */ };
         buildChannel(filter);
         var frame = writeRequest(new ApiVersionsRequestData());
@@ -67,7 +69,7 @@ public class FilterHandlerTest extends FilterHarness {
 
     @Test
     public void testForwardResponse() {
-        ApiVersionsResponseFilter filter = (response, context) -> context.forwardResponse(response);
+        ApiVersionsResponseFilter filter = (header, response, context) -> context.forwardResponse(response);
         buildChannel(filter);
         var frame = writeResponse(new ApiVersionsResponseData());
         var propagated = channel.readInbound();
@@ -83,7 +85,7 @@ public class FilterHandlerTest extends FilterHarness {
             }
 
             @Override
-            public void onApiVersionsResponse(ApiVersionsResponseData response, KrpcFilterContext context) {
+            public void onApiVersionsResponse(ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
                 fail("Should not be called");
             }
         };
@@ -95,7 +97,7 @@ public class FilterHandlerTest extends FilterHarness {
 
     @Test
     public void testDropResponse() {
-        ApiVersionsResponseFilter filter = (response, context) -> {
+        ApiVersionsResponseFilter filter = (header, response, context) -> {
             /* don't call forwardRequest => drop the request */ };
         buildChannel(filter);
         var frame = writeResponse(new ApiVersionsResponseData());
@@ -105,7 +107,7 @@ public class FilterHandlerTest extends FilterHarness {
     public void testSendRequest() {
         FetchRequestData body = new FetchRequestData();
         Future<?>[] fut = { null };
-        ApiVersionsRequestFilter filter = (request, context) -> {
+        ApiVersionsRequestFilter filter = (header, request, context) -> {
             assertNull(fut[0],
                     "Expected to only be called once");
             fut[0] = Future.fromCompletionStage(context.sendRequest((short) 3, body));
@@ -140,7 +142,7 @@ public class FilterHandlerTest extends FilterHarness {
     public void testSendAcklessProduceRequest() {
         ProduceRequestData body = new ProduceRequestData().setAcks((short) 0);
         Future<?>[] fut = { null };
-        ApiVersionsRequestFilter filter = (request, context) -> {
+        ApiVersionsRequestFilter filter = (header, request, context) -> {
             assertNull(fut[0],
                     "Expected to only be called once");
             fut[0] = Future.fromCompletionStage(context.sendRequest((short) 3, body));
@@ -166,7 +168,7 @@ public class FilterHandlerTest extends FilterHarness {
     public void testSendRequestTimeout() throws InterruptedException {
         FetchRequestData body = new FetchRequestData();
         Future<?>[] fut = { null };
-        ApiVersionsRequestFilter filter = (request, context) -> {
+        ApiVersionsRequestFilter filter = (header, request, context) -> {
             assertNull(fut[0],
                     "Expected to only be called once");
             fut[0] = Future.fromCompletionStage(context.sendRequest((short) 3, body));
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaAuthnHandlerTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaAuthnHandlerTest.java
index 45093a6b..16c89d8e 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaAuthnHandlerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaAuthnHandlerTest.java
@@ -22,6 +22,7 @@ import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.MetadataRequestData;
 import org.apache.kafka.common.message.MetadataResponseData;
 import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.message.SaslAuthenticateRequestData;
 import org.apache.kafka.common.message.SaslAuthenticateResponseData;
 import org.apache.kafka.common.message.SaslHandshakeRequestData;
@@ -203,13 +204,11 @@ public class KafkaAuthnHandlerTest {
 
         correlationManager.putBrokerRequest(body.apiKey(), apiVersion, downstreamCorrelationId, true, new KrpcFilter() {
             @Override
-            public void onRequest(DecodedRequestFrame<?> decodedFrame, KrpcFilterContext filterContext) {
-
+            public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
             }
 
             @Override
-            public void onResponse(DecodedResponseFrame<?> decodedFrame, KrpcFilterContext filterContext) {
-
+            public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
             }
 
             @Override
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 e731bb1f..7f15497c 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,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
                         AbstractCodecTest::deserializeRequestHeaderUsingKafkaApis,
                         AbstractCodecTest::deserializeApiVersionsRequestUsingKafkaApis,
                         new KafkaRequestDecoder(
-                                DecodePredicate.forFilters((ApiVersionsRequestFilter) (request, context) -> context.forwardRequest(request))),
+                                DecodePredicate.forFilters((ApiVersionsRequestFilter) (header, request, context) -> context.forwardRequest(request))),
                         DecodedRequestFrame.class,
                         (RequestHeaderData header) -> header),
                 "Unexpected correlation id");
@@ -60,7 +60,8 @@ public class RequestDecoderTest extends AbstractCodecTest {
                                     }
 
                                     @Override
-                                    public void onApiVersionsRequest(ApiVersionsRequestData request, KrpcFilterContext context) {
+                                    public void onApiVersionsRequest(RequestHeaderData header, ApiVersionsRequestData request,
+                                                                     KrpcFilterContext context) {
                                         context.forwardRequest(request);
                                     }
                                 })),
@@ -81,7 +82,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
 
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
-                DecodePredicate.forFilters((ApiVersionsRequestFilter) (request, context) -> context.forwardRequest(request)))
+                DecodePredicate.forFilters((ApiVersionsRequestFilter) (header, request, context) -> context.forwardRequest(request)))
                         .decode(null, byteBuf, messages);
 
         assertEquals(List.of(), messageClasses(messages));
@@ -99,7 +100,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
 
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
-                DecodePredicate.forFilters((ApiVersionsRequestFilter) (request, context) -> context.forwardRequest(request)))
+                DecodePredicate.forFilters((ApiVersionsRequestFilter) (header, request, context) -> context.forwardRequest(request)))
                         .decode(null, byteBuf, messages);
 
         assertEquals(List.of(), messageClasses(messages));
@@ -141,7 +142,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
 
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
-                DecodePredicate.forFilters((ApiVersionsRequestFilter) (request, context) -> context.forwardRequest(request)))
+                DecodePredicate.forFilters((ApiVersionsRequestFilter) (head, request, context) -> context.forwardRequest(request)))
                         .decode(null, byteBuf, messages);
 
         assertEquals(List.of(DecodedRequestFrame.class, DecodedRequestFrame.class), messageClasses(messages));