Repository: kroxylicious/kroxylicious
Pull request #389 (2023-06-14 04:19:17)

Bump kafka-clients from 3.4.0 to 3.5.0
Bumps kafka-clients from 3.4.0 to 3.5.0.


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=org.apache.kafka:kafka-clients&package-manager=maven&previous-version=3.4.0&new-version=3.5.0)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

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>

robobario: Hrm looks like a problem with client 3.5.0 and the Krpc code generator:

```
java.lang.RuntimeException: Exception while processing /home/runner/work/kroxylicious/kroxylicious/krpc-code-gen/target/message-specs/common/message/ConsumerGroupHeartbeatRequest.json
	at io.kroxylicious.krpccodegen.main.KrpcGenerator.lambda$messageSpecs$2(KrpcGenerator.java:274)
	at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
	at java.base/java.util.HashMap$KeySpliterator.forEachRemaining(HashMap.java:1715)
	at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
	at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
	at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921)
	at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
	at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682)
	at io.kroxylicious.krpccodegen.main.KrpcGenerator.messageSpecs(KrpcGenerator.java:278)
	at io.kroxylicious.krpccodegen.main.KrpcGenerator.generate(KrpcGenerator.java:168)
	at io.kroxylicious.krpccodegen.main.KrpcGeneratorTest.testKproxyRequestFilter(KrpcGeneratorTest.java:95)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)
	at java.base/java.lang.reflect.Method.invoke(Method.java:578)
	at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:727)
	at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
	at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:156)
	at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:147)
	at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:86)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(InterceptingExecutableInvoker.java:103)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.lambda$invoke$0(InterceptingExecutableInvoker.java:93)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.invoke(InterceptingExecutableInvoker.java:92)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.invoke(InterceptingExecutableInvoker.java:86)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$7(TestMethodTestDescriptor.java:217)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeTestMethod(TestMethodTestDescriptor.java:213)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:138)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:68)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:151)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
	at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
	at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:41)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
	at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
	at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:41)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
	at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
	at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.submit(SameThreadHierarchicalTestExecutorService.java:35)
	at org.junit.platform.engine.support.hierarchical.HierarchicalTestExecutor.execute(HierarchicalTestExecutor.java:57)
	at org.junit.platform.engine.support.hierarchical.HierarchicalTestEngine.execute(HierarchicalTestEngine.java:54)
	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:147)
	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:127)
	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:90)
	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.lambda$execute$0(EngineExecutionOrchestrator.java:55)
	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.withInterceptedStreams(EngineExecutionOrchestrator.java:102)
	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:54)
	at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:114)
	at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:86)
	at org.junit.platform.launcher.core.DefaultLauncherSession$DelegatingLauncher.execute(DefaultLauncherSession.java:86)
	at org.apache.maven.surefire.junitplatform.LazyLauncher.execute(LazyLauncher.java:56)
	at org.apache.maven.surefire.junitplatform.JUnitPlatformProvider.execute(JUnitPlatformProvider.java:184)
	at org.apache.maven.surefire.junitplatform.JUnitPlatformProvider.invokeAllTests(JUnitPlatformProvider.java:148)
	at org.apache.maven.surefire.junitplatform.JUnitPlatformProvider.invoke(JUnitPlatformProvider.java:122)
	at org.apache.maven.surefire.booter.ForkedBooter.runSuitesInProcess(ForkedBooter.java:385)
	at org.apache.maven.surefire.booter.ForkedBooter.execute(ForkedBooter.java:162)
	at org.apache.maven.surefire.booter.ForkedBooter.run(ForkedBooter.java:507)
	at org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:495)
Caused by: com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException: Unrecognized field "latestVersionUnstable" (class io.kroxylicious.krpccodegen.schema.MessageSpec), not marked as ignorable (8 known properties: "commonStructs", "apiKey", "flexibleVersions", "validVersions", "fields", "type", "name", "listeners"])
 at [Source: (File); line: 69, column: 2] (through reference chain: io.kroxylicious.krpccodegen.schema.MessageSpec["latestVersionUnstable"])
	at com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException.from(UnrecognizedPropertyException.java:61)
	at com.fasterxml.jackson.databind.DeserializationContext.handleUnknownProperty(DeserializationContext.java:1138)
	at com.fasterxml.jackson.databind.deser.std.StdDeserializer.handleUnknownProperty(StdDeserializer.java:2[224](https://github.com/kroxylicious/kroxylicious/actions/runs/5263011933/jobs/9512754412?pr=389#step:5:225))
	at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.handleUnknownProperty(BeanDeserializerBase.java:1709)
	at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.handleUnknownProperties(BeanDeserializerBase.java:1659)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer._deserializeUsingPropertyBased(BeanDeserializer.java:546)
	at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.deserializeFromObjectUsingNonDefault(BeanDeserializerBase.java:1409)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserializeFromObject(BeanDeserializer.java:352)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:185)
	at com.fasterxml.jackson.databind.deser.DefaultDeserializationContext.readRootValue(DefaultDeserializationContext.java:323)
	at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4825)
	at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3637)
	```
robobario: Caused by:

```
commit 3be7f7d611d0786f2f98159d5c7492b0d94a2bb7
Author: David Jacot <djacot@confluent.io>
Date:   Thu Feb 9 09:13:31 2023 +0100

    KAFKA-14391; Add ConsumerGroupHeartbeat API (#12972)
    
    This patch does a few things:
    1) It introduces a new flag to the request spec: `latestVersionUnstable`. It signifies that the last version of the API is considere
d unstable (or still in development). As such, the last API version is not exposed by the server unless specified otherwise with the new
 internal `unstable.api.versions.enable`. This allows us to commit new APIs which are still in development.
    3) It adds the ConsumerGroupHeartbeat API, part of KIP-848, and marks it as unreleased for now.
    4) It adds the new error codes required by the new ConsumerGroupHeartbeat API.
```

So we just have to make the code generator aware of the new flag
robobario: Now it's failing due to incompatibility with the junit extension, we will likely have to upgrade the kafka version in the junit extension first.

Patch:
diff --git a/kroxylicious-additional-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java b/kroxylicious-additional-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
index a49b8bd1..44ec7caa 100644
--- a/kroxylicious-additional-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
+++ b/kroxylicious-additional-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
@@ -341,14 +341,28 @@ public class MultiTenantTransformationFilter
 
     @Override
     public void onAddPartitionsToTxnRequest(short apiVersion, RequestHeaderData header, AddPartitionsToTxnRequestData request, KrpcFilterContext context) {
-        request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
-        applyTenantPrefix(context, request::transactionalId, request::setTransactionalId, true);
+        request.v3AndBelowTopics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
+        applyTenantPrefix(context, request::v3AndBelowTransactionalId, request::setV3AndBelowTransactionalId, true);
+
+        request.transactions().forEach(addPartitionsToTxnTransaction -> {
+            applyTenantPrefix(context, addPartitionsToTxnTransaction::transactionalId, addPartitionsToTxnTransaction::setTransactionalId, true);
+            addPartitionsToTxnTransaction.topics().forEach(addPartitionsToTxnTopic -> {
+                applyTenantPrefix(context, addPartitionsToTxnTopic::name, addPartitionsToTxnTopic::setName, true);
+            });
+        });
         context.forwardRequest(header, request);
     }
 
     @Override
     public void onAddPartitionsToTxnResponse(short apiVersion, ResponseHeaderData header, AddPartitionsToTxnResponseData response, KrpcFilterContext context) {
-        response.results().forEach(results -> removeTenantPrefix(context, results::name, results::setName, false));
+        response.resultsByTopicV3AndBelow().forEach(results -> removeTenantPrefix(context, results::name, results::setName, false));
+
+        response.resultsByTransaction().forEach(addPartitionsToTxnResult -> {
+            removeTenantPrefix(context, addPartitionsToTxnResult::transactionalId, addPartitionsToTxnResult::setTransactionalId, false);
+            for (AddPartitionsToTxnResponseData.AddPartitionsToTxnTopicResult topicResult : addPartitionsToTxnResult.topicResults()) {
+                removeTenantPrefix(context, topicResult::name, topicResult::setName, true);
+            }
+        });
         context.forwardResponse(header, response);
     }
 
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 289b7d7d..60bc0b1f 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
@@ -53,7 +53,7 @@ public class ApiVersionsFilter implements ApiVersionsResponseFilter {
 
         short mutualMax = (short) Math.min(
                 key.maxVersion(),
-                apiKey.messageType.highestSupportedVersion());
+                apiKey.messageType.highestSupportedVersion(true));
         if (mutualMax != key.maxVersion()) {
             LOGGER.trace("{}: {} max version changed to {} (was: {})", channel, apiKey, mutualMin, key.maxVersion());
             key.setMaxVersion(mutualMax);
diff --git a/krpc-code-gen/pom.xml b/krpc-code-gen/pom.xml
index c309bc5d..c3771a96 100644
--- a/krpc-code-gen/pom.xml
+++ b/krpc-code-gen/pom.xml
@@ -77,6 +77,12 @@
             <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
+        <!-- https://mvnrepository.com/artifact/org.assertj/assertj-core -->
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>
diff --git a/krpc-code-gen/src/main/java/io/kroxylicious/krpccodegen/model/MessageSpecModel.java b/krpc-code-gen/src/main/java/io/kroxylicious/krpccodegen/model/MessageSpecModel.java
index e58288fb..c3d42662 100644
--- a/krpc-code-gen/src/main/java/io/kroxylicious/krpccodegen/model/MessageSpecModel.java
+++ b/krpc-code-gen/src/main/java/io/kroxylicious/krpccodegen/model/MessageSpecModel.java
@@ -42,6 +42,8 @@ public class MessageSpecModel implements TemplateHashModel, AdapterTemplateModel
                 return wrapper.wrap(spec.flexibleVersionsString());
             case "apiKey":
                 return wrapper.wrap(spec.apiKey());
+            case "latestVersionUnstable":
+                return wrapper.wrap(spec.latestVersionUnstable());
             case "type":
                 return wrapper.wrap(spec.type());
             case "listeners":
diff --git a/krpc-code-gen/src/main/java/io/kroxylicious/krpccodegen/schema/FieldType.java b/krpc-code-gen/src/main/java/io/kroxylicious/krpccodegen/schema/FieldType.java
index 64032eeb..0be0741f 100644
--- a/krpc-code-gen/src/main/java/io/kroxylicious/krpccodegen/schema/FieldType.java
+++ b/krpc-code-gen/src/main/java/io/kroxylicious/krpccodegen/schema/FieldType.java
@@ -227,6 +227,11 @@ public interface FieldType {
             return true;
         }
 
+        @Override
+        public boolean canBeNullable() {
+            return true;
+        }
+
         public String typeName() {
             return type;
         }
diff --git a/krpc-code-gen/src/main/java/io/kroxylicious/krpccodegen/schema/MessageSpec.java b/krpc-code-gen/src/main/java/io/kroxylicious/krpccodegen/schema/MessageSpec.java
index c768d7e3..5a4b61df 100644
--- a/krpc-code-gen/src/main/java/io/kroxylicious/krpccodegen/schema/MessageSpec.java
+++ b/krpc-code-gen/src/main/java/io/kroxylicious/krpccodegen/schema/MessageSpec.java
@@ -27,17 +27,21 @@ public final class MessageSpec {
 
     private final List<RequestListenerType> listeners;
 
+    private final Optional<Boolean> latestVersionUnstable;
+
     @JsonCreator
     public MessageSpec(@JsonProperty("name") String name,
                        @JsonProperty("validVersions") String validVersions,
                        @JsonProperty("fields") List<FieldSpec> fields,
                        @JsonProperty("apiKey") Short apiKey,
+                       @JsonProperty("latestVersionUnstable") Boolean latestVersionUnstable,
                        @JsonProperty("type") MessageSpecType type,
                        @JsonProperty("commonStructs") List<StructSpec> commonStructs,
                        @JsonProperty("flexibleVersions") String flexibleVersions,
                        @JsonProperty("listeners") List<RequestListenerType> listeners) {
         this.struct = new StructSpec(name, validVersions, fields);
         this.apiKey = apiKey == null ? Optional.empty() : Optional.of(apiKey);
+        this.latestVersionUnstable = latestVersionUnstable == null ? Optional.empty() : Optional.of(latestVersionUnstable);
         this.type = Objects.requireNonNull(type);
         this.commonStructs = commonStructs == null ? Collections.emptyList() : Collections.unmodifiableList(new ArrayList<>(commonStructs));
         if (flexibleVersions == null) {
@@ -87,6 +91,11 @@ public final class MessageSpec {
         return apiKey;
     }
 
+    @JsonProperty("latestVersionUnstable")
+    public Optional<Boolean> latestVersionUnstable() {
+        return latestVersionUnstable;
+    }
+
     @JsonProperty("type")
     public MessageSpecType type() {
         return type;
diff --git a/pom.xml b/pom.xml
index f867cb9d..334b8ff9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -34,8 +34,8 @@
         <guava.version>32.0.1-jre</guava.version>
         <jackson.version>2.15.2</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>
+        <kafka.version>3.5.0</kafka.version>
+        <kroxy.extension.version>0.4.0</kroxy.extension.version>
         <log4j.version>2.20.0</log4j.version>
         <picocli.version>4.7.4</picocli.version>
         <netty.version>4.1.93.Final</netty.version>


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java b/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
index b62147b6..079919f1 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
@@ -93,7 +93,7 @@ public class ProxyRpcTest {
 
     private static @NotNull Stream<Scenario> toScenario(Map<ApiAndVersion, ApiMessage> requestSamples, Map<ApiAndVersion, ApiMessage> responseSample, ApiKeys apiKeys) {
         ApiMessageType messageType = apiKeys.messageType;
-        IntStream supported = IntStream.range(messageType.lowestSupportedVersion(), apiKeys.messageType.highestSupportedVersion() + 1);
+        IntStream supported = IntStream.range(messageType.lowestSupportedVersion(), apiKeys.messageType.highestSupportedVersion(true) + 1);
         return supported.mapToObj(version -> new ApiAndVersion(apiKeys, (short) version)).map(apiAndVersion -> {
             ApiMessage request = requestSamples.get(apiAndVersion);
             ApiMessage response = responseSample.get(apiAndVersion);
diff --git a/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/AddPartitionsToTxn.test.yaml b/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/AddPartitionsToTxn.test.yaml
index 78f61af6..82077689 100644
--- a/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/AddPartitionsToTxn.test.yaml
+++ b/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/AddPartitionsToTxn.test.yaml
@@ -8,29 +8,29 @@
   version: 3
   request:
     payload:
-      transactionalId: "12345"
-      producerId: 0
-      producerEpoch: 0
-      topics:
+      v3AndBelowTransactionalId: "12345"
+      v3AndBelowProducerId: 0
+      v3AndBelowProducerEpoch: 0
+      v3AndBelowTopics:
         - name: foo
           partitions: [0]
     diff:
       - op: replace
-        path: "/transactionalId"
+        path: "/v3AndBelowTransactionalId"
         value: tenant1-12345
       - op: replace
-        path: "/topics/0/name"
+        path: "/v3AndBelowTopics/0/name"
         value: tenant1-foo
   response:
     payload:
-      results:
+      resultsByTopicV3AndBelow:
         - name: tenant1-foo
-          results:
+          resultsByPartition:
             - partitionIndex: 0
-              errorCode: 0
+              partitionErrorCode: 0
       throttleTimeMs: 0
     diff:
       - op: replace
-        path: "/results/0/name"
+        path: "/resultsByTopicV3AndBelow/0/name"
         value: foo
   disabled: false
diff --git a/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/AddPartitionsToTxnV4.test.yaml b/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/AddPartitionsToTxnV4.test.yaml
new file mode 100644
index 00000000..d164761a
--- /dev/null
+++ b/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/AddPartitionsToTxnV4.test.yaml
@@ -0,0 +1,44 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: ADD_PARTITIONS_TO_TXN
+  version: 4
+  request:
+    payload:
+      transactions:
+      - transactionalId: "12345"
+        producerId: 0
+        producerEpoch: 1
+        verifyOnly: false
+        topics:
+          - name: foo
+            partitions: [0]
+    diff:
+      - op: replace
+        path: "/transactions/0/transactionalId"
+        value: tenant1-12345
+      - op: replace
+        path: "/transactions/0/topics/0/name"
+        value: tenant1-foo
+  response:
+    payload:
+      resultsByTransaction:
+        - transactionalId: "tenant1-12345"
+          topicResults:
+            - name: "tenant1-topico"
+              resultsByPartition:
+                - partitionIndex: 1
+                  partitionErrorCode: 0
+      throttleTimeMs: 0
+      errorCode: 0
+    diff:
+      - op: replace
+        path: "/resultsByTransaction/0/transactionalId"
+        value: "12345"
+      - op: replace
+        path: "/resultsByTransaction/0/topicResults/0/name"
+        value: "topico"
+  disabled: false
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/ApiMessageSampleGenerator.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/ApiMessageSampleGenerator.java
index 4298f3cf..e160a9f7 100644
--- a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/ApiMessageSampleGenerator.java
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/ApiMessageSampleGenerator.java
@@ -85,7 +85,7 @@ public class ApiMessageSampleGenerator {
     private static Stream<AbstractMap.SimpleEntry<ApiAndVersion, Class<? extends ApiMessage>>> getSupportedVersions(
                                                                                                                     Map.Entry<ApiKeys, Class<? extends ApiMessage>> apiKeysClassEntry) {
         short lowestSupportedVersion = apiKeysClassEntry.getKey().messageType.lowestSupportedVersion();
-        short highestSupportedVersion = apiKeysClassEntry.getKey().messageType.highestSupportedVersion();
+        short highestSupportedVersion = apiKeysClassEntry.getKey().messageType.highestSupportedVersion(true);
         return IntStream.range(lowestSupportedVersion, highestSupportedVersion + 1).mapToObj(version -> new AbstractMap.SimpleEntry<>(
                 new ApiAndVersion(apiKeysClassEntry.getKey(), (short) version), apiKeysClassEntry.getValue()));
     }
diff --git a/kroxylicious-test-tools/src/test/java/io/kroxylicious/mock/MockServerTest.java b/kroxylicious-test-tools/src/test/java/io/kroxylicious/mock/MockServerTest.java
index 7d962f72..a2a36fbf 100644
--- a/kroxylicious-test-tools/src/test/java/io/kroxylicious/mock/MockServerTest.java
+++ b/kroxylicious-test-tools/src/test/java/io/kroxylicious/mock/MockServerTest.java
@@ -34,7 +34,7 @@ class MockServerTest {
     public static Stream<ApiAndVersion> allSupportedApiVersions() {
         return DataClasses.getRequestClasses().keySet().stream().flatMap(apiKeys -> {
             ApiMessageType messageType = apiKeys.messageType;
-            IntStream supported = IntStream.range(messageType.lowestSupportedVersion(), apiKeys.messageType.highestSupportedVersion() + 1);
+            IntStream supported = IntStream.range(messageType.lowestSupportedVersion(), apiKeys.messageType.highestSupportedVersion(true) + 1);
             return supported.mapToObj(version -> new ApiAndVersion(apiKeys, (short) version));
         });
     }
diff --git a/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/ApiMessageSampleGeneratorTest.java b/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/ApiMessageSampleGeneratorTest.java
index 83754d22..15cee580 100644
--- a/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/ApiMessageSampleGeneratorTest.java
+++ b/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/ApiMessageSampleGeneratorTest.java
@@ -56,7 +56,7 @@ class ApiMessageSampleGeneratorTest {
     public static Stream<ApiMessageSampleGenerator.ApiAndVersion> allSupportedApiVersions() {
         return DataClasses.getRequestClasses().keySet().stream().flatMap(apiKeys -> {
             ApiMessageType messageType = apiKeys.messageType;
-            IntStream supported = IntStream.range(messageType.lowestSupportedVersion(), apiKeys.messageType.highestSupportedVersion() + 1);
+            IntStream supported = IntStream.range(messageType.lowestSupportedVersion(), apiKeys.messageType.highestSupportedVersion(true) + 1);
             return supported.mapToObj(version -> new ApiMessageSampleGenerator.ApiAndVersion(apiKeys, (short) version));
         });
     }
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/AbstractCodecTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/AbstractCodecTest.java
index ec40105f..41bd4107 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/AbstractCodecTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/AbstractCodecTest.java
@@ -57,7 +57,7 @@ public abstract class AbstractCodecTest {
     }
 
     public static Stream<Short> requestApiVersions(ApiMessageType type) {
-        return IntStream.rangeClosed(type.lowestSupportedVersion(), type.highestSupportedVersion())
+        return IntStream.rangeClosed(type.lowestSupportedVersion(), type.highestSupportedVersion(true))
                 .mapToObj(version -> (short) version);
     }
 
@@ -73,7 +73,7 @@ public abstract class AbstractCodecTest {
             ak.add(new ApiVersionsResponseData.ApiVersion()
                     .setApiKey(apiKey.id)
                     .setMinVersion(apiKey.messageType.lowestSupportedVersion())
-                    .setMaxVersion(apiKey.messageType.highestSupportedVersion()));
+                    .setMaxVersion(apiKey.messageType.highestSupportedVersion(true)));
         }
 
         return new ApiVersionsResponseData()
diff --git a/krpc-code-gen/src/test/java/io/kroxylicious/krpccodegen/main/KrpcGeneratorTest.java b/krpc-code-gen/src/test/java/io/kroxylicious/krpccodegen/main/KrpcGeneratorTest.java
index 81b07902..53efb3ed 100644
--- a/krpc-code-gen/src/test/java/io/kroxylicious/krpccodegen/main/KrpcGeneratorTest.java
+++ b/krpc-code-gen/src/test/java/io/kroxylicious/krpccodegen/main/KrpcGeneratorTest.java
@@ -6,12 +6,24 @@
 package io.kroxylicious.krpccodegen.main;
 
 import java.io.File;
+import java.io.IOException;
 import java.net.URISyntaxException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.List;
+import java.util.Objects;
+import java.util.StringJoiner;
 
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import com.google.common.io.Files;
+import com.google.common.io.Resources;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
@@ -32,75 +44,121 @@ import org.junit.jupiter.api.Test;
 class KrpcGeneratorTest {
 
     private static final String MESSAGE_SPECS_PATH = "message-specs/common/message";
-    private static final String OUTPUT_DIR = "krpc-test-sources/krpc";
     private static final String TEST_CLASSES_DIR = "test-classes";
 
     @Test
-    public void testHelloWorld() throws Exception {
+    public void testHelloWorld(@TempDir File tempDir) throws Exception {
         KrpcGenerator gen = KrpcGenerator.single()
                 .withMessageSpecDir(getMessageSpecDir())
                 .withMessageSpecFilter("*.json")
                 .withTemplateDir(getTemplateDir())
                 .withTemplateNames(List.of("hello-world/example.ftl"))
                 .withOutputPackage("com.foo")
-                .withOutputDir(getOutputDir())
+                .withOutputDir(tempDir)
                 .withOutputFilePattern("${messageSpecName}.txt")
                 .build();
 
         gen.generate();
+
+        File file = join(tempDir, "com", "foo", "FetchRequest.txt");
+        assertFileHasExpectedContents(file, "hello-world/example-expected-FetchRequest.txt");
     }
 
     @Test
-    public void testKrpcData() throws Exception {
+    public void testKrpcData(@TempDir File tempDir) throws Exception {
         KrpcGenerator gen = KrpcGenerator.single()
                 .withMessageSpecDir(getMessageSpecDir())
                 .withMessageSpecFilter("*.json")
                 .withTemplateDir(getTemplateDir())
                 .withTemplateNames(List.of("Data/example.ftl"))
                 .withOutputPackage("com.foo")
-                .withOutputDir(getOutputDir())
+                .withOutputDir(tempDir)
                 .withOutputFilePattern("${messageSpecName}.java")
                 .build();
 
         gen.generate();
+
+        File file = join(tempDir, "com", "foo", "FetchRequest.java");
+        assertFileHasExpectedContents(file, "Data/example-expected-FetchRequest.java.txt");
     }
 
     @Test
-    public void testKproxyFilter() throws Exception {
+    public void testKproxyFilter(@TempDir File tempDir) throws Exception {
         KrpcGenerator gen = KrpcGenerator.single()
                 .withMessageSpecDir(getMessageSpecDir())
                 .withMessageSpecFilter("*.json")
                 .withTemplateDir(getTemplateDir())
                 .withTemplateNames(List.of("Kproxy/Filter.ftl"))
                 .withOutputPackage("com.foo")
-                .withOutputDir(getOutputDir())
+                .withOutputDir(tempDir)
                 .withOutputFilePattern("${messageSpecName}Filter.java")
                 .build();
 
         gen.generate();
+
+        File file = join(tempDir, "com", "foo", "FetchRequestFilter.java");
+        assertFileHasExpectedContents(file, "Kproxy/Filter-expected-FetchRequestFilter.java.txt");
     }
 
     @Test
-    public void testKproxyRequestFilter() throws Exception {
+    public void testKproxyRequestFilter(@TempDir File tempDir) throws Exception {
         KrpcGenerator gen = KrpcGenerator.multi()
                 .withMessageSpecDir(getMessageSpecDir())
                 .withMessageSpecFilter("*{Request}.json")
                 .withTemplateDir(getTemplateDir())
                 .withTemplateNames(List.of("Kproxy/KrpcRequestFilter.ftl"))
                 .withOutputPackage("com.foo")
-                .withOutputDir(getOutputDir())
+                .withOutputDir(tempDir)
                 .withOutputFilePattern("${templateName}.java")
                 .build();
 
         gen.generate();
+
+        File file = join(tempDir, "com", "foo", "KrpcRequestFilter.java");
+        assertFileHasExpectedContents(file, "Kproxy/KrpcRequestFilter-expected.txt");
     }
 
-    private static File getMessageSpecDir() {
-        return getBuildDir().resolve(MESSAGE_SPECS_PATH).toFile();
+    @ParameterizedTest
+    @CsvSource(value = { "AddPartitionsToTxnRequest.json,yes", "FetchRequest.json,no" })
+    public void testLatestVersionUnstable(String messageSpec, String expectedContents, @TempDir File tempDir) throws Exception {
+        testSingleGeneration(tempDir, messageSpec, "${messageSpec.latestVersionUnstable.isPresent()?string('yes', 'no')}", expectedContents);
+        testSingleGeneration(tempDir, messageSpec, "${messageSpec.latestVersionUnstable.orElse(false)?string('yes', 'no')}", expectedContents);
+    }
+
+    private void assertFileHasExpectedContents(File file, String expectedFile) throws IOException {
+        String expected = Resources.asCharSource(
+                Objects.requireNonNull(getClass().getClassLoader().getResource(expectedFile)), UTF_8).read();
+        assertThat(file).content().isEqualTo(expected);
+    }
+
+    private static void testSingleGeneration(File tempDir, String messageSpec, String template, String expectedContents) throws Exception {
+        String templateFile = "template.ftl";
+        String outputFile = "output.txt";
+        Files.asCharSink(new File(tempDir, templateFile), UTF_8).write(template);
+        KrpcGenerator gen = KrpcGenerator.single()
+                .withMessageSpecDir(getMessageSpecDir())
+                .withMessageSpecFilter(messageSpec)
+                .withTemplateDir(tempDir)
+                .withTemplateNames(List.of(templateFile))
+                .withOutputPackage("com.foo")
+                .withOutputDir(tempDir)
+                .withOutputFilePattern(outputFile)
+                .build();
+        gen.generate();
+        File file = join(tempDir, "com", "foo", outputFile);
+        assertThat(file).hasContent(expectedContents);
     }
 
-    private static File getOutputDir() {
-        return getBuildDir().resolve(OUTPUT_DIR).toFile();
+    private static File join(File dir, String... pathElements) {
+        StringJoiner stringJoiner = new StringJoiner(File.separator);
+        for (String pathElement : pathElements) {
+            stringJoiner.add(pathElement);
+        }
+        return new File(dir, stringJoiner.toString());
+    }
+
+    private static File getMessageSpecDir() {
+        return getBuildDir().resolve(MESSAGE_SPECS_PATH).toFile();
     }
 
     private static File getTemplateDir() {
diff --git a/krpc-code-gen/src/test/resources/Data/example-expected-FetchRequest.java.txt b/krpc-code-gen/src/test/resources/Data/example-expected-FetchRequest.java.txt
new file mode 100644
index 00000000..9a35c657
--- /dev/null
+++ b/krpc-code-gen/src/test/resources/Data/example-expected-FetchRequest.java.txt
@@ -0,0 +1,434 @@
+====
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+====
+
+// THIS CODE IS AUTOMATICALLY GENERATED.  DO NOT EDIT.
+
+package org.apache.kafka.common.message;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.Message;
+import org.apache.kafka.common.protocol.MessageSizeAccumulator;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.protocol.types.ArrayOf;
+import org.apache.kafka.common.protocol.types.CompactArrayOf;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.RawTaggedField;
+import org.apache.kafka.common.protocol.types.RawTaggedFieldWriter;
+import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.types.Type;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.common.utils.ImplicitLinkedHashCollection;
+import org.apache.kafka.common.utils.ImplicitLinkedHashMultiCollection;
+
+import static org.apache.kafka.common.protocol.types.Field.TaggedFieldsSection;
+
+public class FetchRequestData implements ApiMessage {
+
+    UNKNOWN - ClusterId clusterId;
+    int replicaId;
+    UNKNOWN - ReplicaState replicaState;
+    int maxWaitMs;
+    int minBytes;
+    int maxBytes;
+    UNKNOWN - IsolationLevel isolationLevel;
+    int sessionId;
+    int sessionEpoch;
+    FetchTopicCollection topics;
+    ForgottenTopicCollection forgottenTopicsData;
+    UNKNOWN - RackId rackId;
+    private List<RawTaggedField> _unknownTaggedFields;
+
+
+
+    public static final Schema[] SCHEMAS = new Schema[] {
+        SCHEMA_0,
+        SCHEMA_1,
+        SCHEMA_2,
+        SCHEMA_3,
+        SCHEMA_4,
+        SCHEMA_5,
+        SCHEMA_6,
+        SCHEMA_7,
+        SCHEMA_8,
+        SCHEMA_9,
+        SCHEMA_10,
+        SCHEMA_11,
+        SCHEMA_12,
+        SCHEMA_13,
+        SCHEMA_14,
+        SCHEMA_15
+    };
+
+    public static final short LOWEST_SUPPORTED_VERSION = 0;
+    public static final short HIGHEST_SUPPORTED_VERSION = 15;
+
+    public FetchRequestData(Readable _readable, short _version) {
+        read(_readable, _version);
+    }
+
+    public FetchRequestData() {
+        this.clusterId = null;
+        this.replicaId = -1;
+        this.replicaState = ;
+        this.maxWaitMs = ;
+        this.minBytes = ;
+        this.maxBytes = 0x7fffffff;
+        this.isolationLevel = 0;
+        this.sessionId = 0;
+        this.sessionEpoch = -1;
+        this.topics = new FetchTopicCollection(0);
+        this.forgottenTopicsData = new ForgottenTopicCollection(0);
+        this.rackId = ;
+    }
+
+    /** The API key */
+    @Override
+    public short apiKey() {
+        return 1;
+    }
+
+    /** @return the lowest valid version for this RPC. */
+    @Override
+    public short lowestSupportedVersion() {
+        return 0;
+    }
+
+    /** @return the highest valid version for this message. */
+    @Override
+    public short highestSupportedVersion() {
+        return 15;
+    }
+
+    @Override
+    public void read(Readable _readable, short _version) {
+        // TODO
+    }
+
+    @Override
+    public void write(Writeable _writeable, ObjectSerializationCache _cache, short _version) {
+        // TODO
+    }
+
+    @Override
+    public void addSize(MessageSizeAccumulator _size, ObjectSerializationCache _cache, short _version) {
+        // TODO
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (!(obj instanceof FetchRequestData)) return false;
+        FetchRequestData other = (FetchRequestData) obj;
+        if (this.clusterId == null) {
+            if (other.clusterId != null) return false;
+        } else {
+            if (!this.clusterId.equals(other.clusterId)) return false;
+        }
+        if (replicaId != other.replicaId) return false;
+        if (this.replicaState == null) {
+            if (other.replicaState != null) return false;
+        } else {
+            if (!this.replicaState.equals(other.replicaState)) return false;
+        }
+        if (maxWaitMs != other.maxWaitMs) return false;
+        if (minBytes != other.minBytes) return false;
+        if (maxBytes != other.maxBytes) return false;
+        if (isolationLevel != other.isolationLevel) return false;
+        if (sessionId != other.sessionId) return false;
+        if (sessionEpoch != other.sessionEpoch) return false;
+        if (this.topics == null) {
+            if (other.topics != null) return false;
+        } else {
+            if (!this.topics.equals(other.topics)) return false;
+        }
+        if (this.forgottenTopicsData == null) {
+            if (other.forgottenTopicsData != null) return false;
+        } else {
+            if (!this.forgottenTopicsData.equals(other.forgottenTopicsData)) return false;
+        }
+        if (this.rackId == null) {
+            if (other.rackId != null) return false;
+        } else {
+            if (!this.rackId.equals(other.rackId)) return false;
+        }
+        return MessageUtil.compareRawTaggedFields(_unknownTaggedFields, other._unknownTaggedFields);
+    }
+
+    @Override
+    public int hashCode() {
+        int hashCode = 0;
+        hashCode = 31 * hashCode + (clusterId == null ? 0 : clusterId.hashCode());
+        hashCode = 31 * hashCode + replicaId;
+        hashCode = 31 * hashCode + (replicaState == null ? 0 : replicaState.hashCode());
+        hashCode = 31 * hashCode + maxWaitMs;
+        hashCode = 31 * hashCode + minBytes;
+        hashCode = 31 * hashCode + maxBytes;
+        hashCode = 31 * hashCode + isolationLevel;
+        hashCode = 31 * hashCode + sessionId;
+        hashCode = 31 * hashCode + sessionEpoch;
+        hashCode = 31 * hashCode + (topics == null ? 0 : topics.hashCode());
+        hashCode = 31 * hashCode + (forgottenTopicsData == null ? 0 : forgottenTopicsData.hashCode());
+        hashCode = 31 * hashCode + (rackId == null ? 0 : rackId.hashCode());
+        return hashCode;
+    }
+
+    @Override
+    public FetchRequestData duplicate() {
+        FetchRequestData _duplicate = new FetchRequestData();
+        _duplicate.clusterId = clusterId;
+        _duplicate.replicaId = replicaId;
+        _duplicate.replicaState = replicaState;
+        _duplicate.maxWaitMs = maxWaitMs;
+        _duplicate.minBytes = minBytes;
+        _duplicate.maxBytes = maxBytes;
+        _duplicate.isolationLevel = isolationLevel;
+        _duplicate.sessionId = sessionId;
+        _duplicate.sessionEpoch = sessionEpoch;
+        FetchTopicCollection newTopics = new FetchTopicCollection(topics.size());
+        for (FetchTopic _element : topics) {
+            newTopics.add(_element.duplicate());
+        }
+        _duplicate.topics = newTopics;
+        ForgottenTopicCollection newForgottenTopicsData = new ForgottenTopicCollection(forgottenTopicsData.size());
+        for (ForgottenTopic _element : forgottenTopicsData) {
+            newForgottenTopicsData.add(_element.duplicate());
+        }
+        _duplicate.forgottenTopicsData = newForgottenTopicsData;
+        _duplicate.rackId = rackId;
+        return _duplicate;
+    }
+
+    @Override
+    public String toString() {
+        return "FetchRequestData("
+            // TODO
+            + ")";
+    }
+
+    /** 
+     * Gets the ClusterId. 
+     * The clusterId if known. This is used to validate metadata fetches prior to broker registration.
+     * @return the ClusterId.
+     */
+    public UNKNOWN - ClusterId clusterid() {
+        return this.clusterId;
+    }
+    /** 
+     * Gets the ReplicaId. 
+     * The broker ID of the follower, of -1 if this request is from a consumer.
+     * @return the ReplicaId.
+     */
+    public int replicaid() {
+        return this.replicaId;
+    }
+    /** 
+     * Gets the ReplicaState. 
+     * 
+     * @return the ReplicaState.
+     */
+    public UNKNOWN - ReplicaState replicastate() {
+        return this.replicaState;
+    }
+    /** 
+     * Gets the MaxWaitMs. 
+     * The maximum time in milliseconds to wait for the response.
+     * @return the MaxWaitMs.
+     */
+    public int maxwaitms() {
+        return this.maxWaitMs;
+    }
+    /** 
+     * Gets the MinBytes. 
+     * The minimum bytes to accumulate in the response.
+     * @return the MinBytes.
+     */
+    public int minbytes() {
+        return this.minBytes;
+    }
+    /** 
+     * Gets the MaxBytes. 
+     * The maximum bytes to fetch.  See KIP-74 for cases where this limit may not be honored.
+     * @return the MaxBytes.
+     */
+    public int maxbytes() {
+        return this.maxBytes;
+    }
+    /** 
+     * Gets the IsolationLevel. 
+     * This setting controls the visibility of transactional records. Using READ_UNCOMMITTED (isolation_level = 0) makes all records visible. With READ_COMMITTED (isolation_level = 1), non-transactional and COMMITTED transactional records are visible. To be more concrete, READ_COMMITTED returns all data from offsets smaller than the current LSO (last stable offset), and enables the inclusion of the list of aborted transactions in the result, which allows consumers to discard ABORTED transactional records
+     * @return the IsolationLevel.
+     */
+    public UNKNOWN - IsolationLevel isolationlevel() {
+        return this.isolationLevel;
+    }
+    /** 
+     * Gets the SessionId. 
+     * The fetch session ID.
+     * @return the SessionId.
+     */
+    public int sessionid() {
+        return this.sessionId;
+    }
+    /** 
+     * Gets the SessionEpoch. 
+     * The fetch session epoch, which is used for ordering requests in a session.
+     * @return the SessionEpoch.
+     */
+    public int sessionepoch() {
+        return this.sessionEpoch;
+    }
+    /** 
+     * Gets the Topics. 
+     * The topics to fetch.
+     * @return the Topics.
+     */
+    public FetchTopicCollection topics() {
+        return this.topics;
+    }
+    /** 
+     * Gets the ForgottenTopicsData. 
+     * In an incremental fetch request, the partitions to remove.
+     * @return the ForgottenTopicsData.
+     */
+    public ForgottenTopicCollection forgottentopicsdata() {
+        return this.forgottenTopicsData;
+    }
+    /** 
+     * Gets the RackId. 
+     * Rack ID of the consumer making this request
+     * @return the RackId.
+     */
+    public UNKNOWN - RackId rackid() {
+        return this.rackId;
+    }
+
+    @Override
+    public List<RawTaggedField> unknownTaggedFields() {
+        if (_unknownTaggedFields == null) {
+            _unknownTaggedFields = new ArrayList<>(0);
+        }
+        return _unknownTaggedFields;
+    }
+
+    /** 
+     * Sets the ClusterId. 
+     * The clusterId if known. This is used to validate metadata fetches prior to broker registration.
+     * @param ClusterId the ClusterId.
+     */
+    public FetchRequestData setClusterId(UNKNOWN - ClusterId clusterId) {
+        this.clusterId = clusterId;
+        return this;
+    }
+    /** 
+     * Sets the ReplicaId. 
+     * The broker ID of the follower, of -1 if this request is from a consumer.
+     * @param ReplicaId the ReplicaId.
+     */
+    public FetchRequestData setReplicaId(int replicaId) {
+        this.replicaId = replicaId;
+        return this;
+    }
+    /** 
+     * Sets the ReplicaState. 
+     * 
+     * @param ReplicaState the ReplicaState.
+     */
+    public FetchRequestData setReplicaState(UNKNOWN - ReplicaState replicaState) {
+        this.replicaState = replicaState;
+        return this;
+    }
+    /** 
+     * Sets the MaxWaitMs. 
+     * The maximum time in milliseconds to wait for the response.
+     * @param MaxWaitMs the MaxWaitMs.
+     */
+    public FetchRequestData setMaxWaitMs(int maxWaitMs) {
+        this.maxWaitMs = maxWaitMs;
+        return this;
+    }
+    /** 
+     * Sets the MinBytes. 
+     * The minimum bytes to accumulate in the response.
+     * @param MinBytes the MinBytes.
+     */
+    public FetchRequestData setMinBytes(int minBytes) {
+        this.minBytes = minBytes;
+        return this;
+    }
+    /** 
+     * Sets the MaxBytes. 
+     * The maximum bytes to fetch.  See KIP-74 for cases where this limit may not be honored.
+     * @param MaxBytes the MaxBytes.
+     */
+    public FetchRequestData setMaxBytes(int maxBytes) {
+        this.maxBytes = maxBytes;
+        return this;
+    }
+    /** 
+     * Sets the IsolationLevel. 
+     * This setting controls the visibility of transactional records. Using READ_UNCOMMITTED (isolation_level = 0) makes all records visible. With READ_COMMITTED (isolation_level = 1), non-transactional and COMMITTED transactional records are visible. To be more concrete, READ_COMMITTED returns all data from offsets smaller than the current LSO (last stable offset), and enables the inclusion of the list of aborted transactions in the result, which allows consumers to discard ABORTED transactional records
+     * @param IsolationLevel the IsolationLevel.
+     */
+    public FetchRequestData setIsolationLevel(UNKNOWN - IsolationLevel isolationLevel) {
+        this.isolationLevel = isolationLevel;
+        return this;
+    }
+    /** 
+     * Sets the SessionId. 
+     * The fetch session ID.
+     * @param SessionId the SessionId.
+     */
+    public FetchRequestData setSessionId(int sessionId) {
+        this.sessionId = sessionId;
+        return this;
+    }
+    /** 
+     * Sets the SessionEpoch. 
+     * The fetch session epoch, which is used for ordering requests in a session.
+     * @param SessionEpoch the SessionEpoch.
+     */
+    public FetchRequestData setSessionEpoch(int sessionEpoch) {
+        this.sessionEpoch = sessionEpoch;
+        return this;
+    }
+    /** 
+     * Sets the Topics. 
+     * The topics to fetch.
+     * @param Topics the Topics.
+     */
+    public FetchRequestData setTopics(FetchTopicCollection topics) {
+        this.topics = topics;
+        return this;
+    }
+    /** 
+     * Sets the ForgottenTopicsData. 
+     * In an incremental fetch request, the partitions to remove.
+     * @param ForgottenTopicsData the ForgottenTopicsData.
+     */
+    public FetchRequestData setForgottenTopicsData(ForgottenTopicCollection forgottenTopicsData) {
+        this.forgottenTopicsData = forgottenTopicsData;
+        return this;
+    }
+    /** 
+     * Sets the RackId. 
+     * Rack ID of the consumer making this request
+     * @param RackId the RackId.
+     */
+    public FetchRequestData setRackId(UNKNOWN - RackId rackId) {
+        this.rackId = rackId;
+        return this;
+    }
+
+    // TODO nested classes
+
+}
diff --git a/krpc-code-gen/src/test/resources/Data/example.ftl b/krpc-code-gen/src/test/resources/Data/example.ftl
index b031b7b2..2ebad8fe 100644
--- a/krpc-code-gen/src/test/resources/Data/example.ftl
+++ b/krpc-code-gen/src/test/resources/Data/example.ftl
@@ -94,22 +94,11 @@ ${field.defaultString}
 </#if>
 </#compress>
 </#macro>
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
+====
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+====
 
 // THIS CODE IS AUTOMATICALLY GENERATED.  DO NOT EDIT.
 
diff --git a/krpc-code-gen/src/test/resources/Kproxy/Filter-expected-FetchRequestFilter.java.txt b/krpc-code-gen/src/test/resources/Kproxy/Filter-expected-FetchRequestFilter.java.txt
new file mode 100644
index 00000000..85fd92db
--- /dev/null
+++ b/krpc-code-gen/src/test/resources/Kproxy/Filter-expected-FetchRequestFilter.java.txt
@@ -0,0 +1,29 @@
+====
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+====
+
+package com.foo;
+
+import org.apache.kafka.common.message.FetchRequestData;
+
+/**
+ * A stateless filter for FetchRequests.
+ * The same instance may be invoked on multiple channels.
+ */
+public interface FetchRequestFilter extends KrpcRequestFilter {
+    /**
+     * Handle the given {@code request},
+     * returning the {@code FetchRequestData} instance to be passed to the next filter.
+     * The implementation may modify the given {@code data} in-place and return it,
+     * or instantiate a new one.
+     *
+     * @param request The KRPC message to handle.
+     * @param context The context.
+     * @return the {@code request} to be passed to the next filter.
+     * If null is returned then the given {@code request} will be used.
+     */
+    public KrpcFilterState onFetchRequest(FetchRequestData request, KrpcFilterContext context);
+
+}
\ No newline at end of file
diff --git a/krpc-code-gen/src/test/resources/Kproxy/Filter.ftl b/krpc-code-gen/src/test/resources/Kproxy/Filter.ftl
index 38e6dc40..533ce700 100644
--- a/krpc-code-gen/src/test/resources/Kproxy/Filter.ftl
+++ b/krpc-code-gen/src/test/resources/Kproxy/Filter.ftl
@@ -10,22 +10,12 @@
   filterClass="${messageSpec.name}Filter"
   msgType=messageSpec.type?lower_case
 />
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
+====
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+====
+
 package ${outputPackage};
 
 import org.apache.kafka.common.message.${messageSpec.name}Data;
diff --git a/krpc-code-gen/src/test/resources/Kproxy/KrpcRequestFilter-expected.txt b/krpc-code-gen/src/test/resources/Kproxy/KrpcRequestFilter-expected.txt
new file mode 100644
index 00000000..fe810b35
--- /dev/null
+++ b/krpc-code-gen/src/test/resources/Kproxy/KrpcRequestFilter-expected.txt
@@ -0,0 +1,500 @@
+====
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+====
+
+package com.foo;
+
+import org.apache.kafka.common.message.AddOffsetsToTxnRequestData;
+import org.apache.kafka.common.message.AddPartitionsToTxnRequestData;
+import org.apache.kafka.common.message.AllocateProducerIdsRequestData;
+import org.apache.kafka.common.message.AlterClientQuotasRequestData;
+import org.apache.kafka.common.message.AlterConfigsRequestData;
+import org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterReplicaLogDirsRequestData;
+import org.apache.kafka.common.message.AlterUserScramCredentialsRequestData;
+import org.apache.kafka.common.message.ApiVersionsRequestData;
+import org.apache.kafka.common.message.BeginQuorumEpochRequestData;
+import org.apache.kafka.common.message.BrokerHeartbeatRequestData;
+import org.apache.kafka.common.message.BrokerRegistrationRequestData;
+import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
+import org.apache.kafka.common.message.ControlledShutdownRequestData;
+import org.apache.kafka.common.message.CreateAclsRequestData;
+import org.apache.kafka.common.message.CreateDelegationTokenRequestData;
+import org.apache.kafka.common.message.CreatePartitionsRequestData;
+import org.apache.kafka.common.message.CreateTopicsRequestData;
+import org.apache.kafka.common.message.DeleteAclsRequestData;
+import org.apache.kafka.common.message.DeleteGroupsRequestData;
+import org.apache.kafka.common.message.DeleteRecordsRequestData;
+import org.apache.kafka.common.message.DeleteTopicsRequestData;
+import org.apache.kafka.common.message.DescribeAclsRequestData;
+import org.apache.kafka.common.message.DescribeClientQuotasRequestData;
+import org.apache.kafka.common.message.DescribeClusterRequestData;
+import org.apache.kafka.common.message.DescribeConfigsRequestData;
+import org.apache.kafka.common.message.DescribeDelegationTokenRequestData;
+import org.apache.kafka.common.message.DescribeGroupsRequestData;
+import org.apache.kafka.common.message.DescribeLogDirsRequestData;
+import org.apache.kafka.common.message.DescribeProducersRequestData;
+import org.apache.kafka.common.message.DescribeQuorumRequestData;
+import org.apache.kafka.common.message.DescribeTransactionsRequestData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsRequestData;
+import org.apache.kafka.common.message.ElectLeadersRequestData;
+import org.apache.kafka.common.message.EndQuorumEpochRequestData;
+import org.apache.kafka.common.message.EndTxnRequestData;
+import org.apache.kafka.common.message.EnvelopeRequestData;
+import org.apache.kafka.common.message.ExpireDelegationTokenRequestData;
+import org.apache.kafka.common.message.FetchRequestData;
+import org.apache.kafka.common.message.FetchSnapshotRequestData;
+import org.apache.kafka.common.message.FindCoordinatorRequestData;
+import org.apache.kafka.common.message.HeartbeatRequestData;
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData;
+import org.apache.kafka.common.message.InitProducerIdRequestData;
+import org.apache.kafka.common.message.JoinGroupRequestData;
+import org.apache.kafka.common.message.LeaderAndIsrRequestData;
+import org.apache.kafka.common.message.LeaveGroupRequestData;
+import org.apache.kafka.common.message.ListGroupsRequestData;
+import org.apache.kafka.common.message.ListOffsetsRequestData;
+import org.apache.kafka.common.message.ListPartitionReassignmentsRequestData;
+import org.apache.kafka.common.message.ListTransactionsRequestData;
+import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.OffsetCommitRequestData;
+import org.apache.kafka.common.message.OffsetDeleteRequestData;
+import org.apache.kafka.common.message.OffsetFetchRequestData;
+import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData;
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.RenewDelegationTokenRequestData;
+import org.apache.kafka.common.message.SaslAuthenticateRequestData;
+import org.apache.kafka.common.message.SaslHandshakeRequestData;
+import org.apache.kafka.common.message.StopReplicaRequestData;
+import org.apache.kafka.common.message.SyncGroupRequestData;
+import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
+import org.apache.kafka.common.message.UnregisterBrokerRequestData;
+import org.apache.kafka.common.message.UpdateFeaturesRequestData;
+import org.apache.kafka.common.message.UpdateMetadataRequestData;
+import org.apache.kafka.common.message.VoteRequestData;
+import org.apache.kafka.common.message.WriteTxnMarkersRequestData;
+import org.apache.kafka.common.protocol.ApiKeys;
+
+import io.kroxylicious.proxy.codec.DecodedRequestFrame;
+
+/**
+ * <p>Interface for {@code *RequestFilter}s.
+ * This interface can be implemented in two ways:
+ * <ul>
+ *     <li>filter classes can (multiply) implement one of the RPC-specific subinterfaces such as {@link ProduceRequestFilter} for a type-safe API</li>
+ *     <li>filter classes can extend {@link KrpcGenericRequestFilter}</li>
+ * </ul>
+ *
+ * <p>When implementing one or more of the {@code *RequestFilter} subinterfaces you need only implement
+ * the {@code on*Request} method(s), unless your filter can avoid deserialization in which case
+ * you can override {@link #shouldDeserializeRequest(ApiKeys, short)} as well.</p>
+ *
+ * <p>When extending {@link KrpcGenericRequestFilter} you need to override {@link #apply(DecodedRequestFrame, KrpcFilterContext)},
+ * and may override {@link #shouldDeserializeRequest(ApiKeys, short)} as well.</p>
+ *
+ * <h2>Guarantees</h2>
+ * <p>Implementors of this API may assume the following:</p>
+ * <ol>
+ *     <li>That each instance of the filter is associated with a single channel</li>
+ *     <li>That {@link #shouldDeserializeRequest(ApiKeys, short)} and
+ *     {@link #apply(DecodedRequestFrame, 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>
+ * <p>From 1. and 2. it follows that you can use member variables in your filter to
+ * store channel-local state.</p>
+ *
+ * <p>Implementors should <strong>not</strong> assume:</p>
+ * <ol>
+ *     <li>That filters in the same chain execute on the same thread. Thus inter-filter communication/state
+ *     transfer needs to be thread-safe</li>
+ * </ol>
+ */
+public /* sealed */ interface KrpcRequestFilter extends KrpcFilter /* TODO permits ... */ {
+
+    /**
+     * Apply the filter to the given {@code decodedFrame} using the given {@code filterContext}.
+     * @param decodedFrame The request frame.
+     * @param filterContext The filter context.
+     * @return The state of the filter.
+     */
+    public default KrpcFilterState apply(DecodedRequestFrame<?> decodedFrame,
+                                         KrpcFilterContext filterContext) {
+        KrpcFilterState state;
+        switch (decodedFrame.apiKey()) {
+            case ADD_OFFSETS_TO_TXN:
+                state = ((AddOffsetsToTxnRequestFilter) this).onAddOffsetsToTxnRequest((AddOffsetsToTxnRequestData) decodedFrame.body(), filterContext);
+                break;
+            case ADD_PARTITIONS_TO_TXN:
+                state = ((AddPartitionsToTxnRequestFilter) this).onAddPartitionsToTxnRequest((AddPartitionsToTxnRequestData) decodedFrame.body(), filterContext);
+                break;
+            case ALLOCATE_PRODUCER_IDS:
+                state = ((AllocateProducerIdsRequestFilter) this).onAllocateProducerIdsRequest((AllocateProducerIdsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case ALTER_CLIENT_QUOTAS:
+                state = ((AlterClientQuotasRequestFilter) this).onAlterClientQuotasRequest((AlterClientQuotasRequestData) decodedFrame.body(), filterContext);
+                break;
+            case ALTER_CONFIGS:
+                state = ((AlterConfigsRequestFilter) this).onAlterConfigsRequest((AlterConfigsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case ALTER_PARTITION_REASSIGNMENTS:
+                state = ((AlterPartitionReassignmentsRequestFilter) this).onAlterPartitionReassignmentsRequest((AlterPartitionReassignmentsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case ALTER_PARTITION:
+                state = ((AlterPartitionRequestFilter) this).onAlterPartitionRequest((AlterPartitionRequestData) decodedFrame.body(), filterContext);
+                break;
+            case ALTER_REPLICA_LOG_DIRS:
+                state = ((AlterReplicaLogDirsRequestFilter) this).onAlterReplicaLogDirsRequest((AlterReplicaLogDirsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case ALTER_USER_SCRAM_CREDENTIALS:
+                state = ((AlterUserScramCredentialsRequestFilter) this).onAlterUserScramCredentialsRequest((AlterUserScramCredentialsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case API_VERSIONS:
+                state = ((ApiVersionsRequestFilter) this).onApiVersionsRequest((ApiVersionsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case BEGIN_QUORUM_EPOCH:
+                state = ((BeginQuorumEpochRequestFilter) this).onBeginQuorumEpochRequest((BeginQuorumEpochRequestData) decodedFrame.body(), filterContext);
+                break;
+            case BROKER_HEARTBEAT:
+                state = ((BrokerHeartbeatRequestFilter) this).onBrokerHeartbeatRequest((BrokerHeartbeatRequestData) decodedFrame.body(), filterContext);
+                break;
+            case BROKER_REGISTRATION:
+                state = ((BrokerRegistrationRequestFilter) this).onBrokerRegistrationRequest((BrokerRegistrationRequestData) decodedFrame.body(), filterContext);
+                break;
+            case CONSUMER_GROUP_HEARTBEAT:
+                state = ((ConsumerGroupHeartbeatRequestFilter) this).onConsumerGroupHeartbeatRequest((ConsumerGroupHeartbeatRequestData) decodedFrame.body(), filterContext);
+                break;
+            case CONTROLLED_SHUTDOWN:
+                state = ((ControlledShutdownRequestFilter) this).onControlledShutdownRequest((ControlledShutdownRequestData) decodedFrame.body(), filterContext);
+                break;
+            case CREATE_ACLS:
+                state = ((CreateAclsRequestFilter) this).onCreateAclsRequest((CreateAclsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case CREATE_DELEGATION_TOKEN:
+                state = ((CreateDelegationTokenRequestFilter) this).onCreateDelegationTokenRequest((CreateDelegationTokenRequestData) decodedFrame.body(), filterContext);
+                break;
+            case CREATE_PARTITIONS:
+                state = ((CreatePartitionsRequestFilter) this).onCreatePartitionsRequest((CreatePartitionsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case CREATE_TOPICS:
+                state = ((CreateTopicsRequestFilter) this).onCreateTopicsRequest((CreateTopicsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DELETE_ACLS:
+                state = ((DeleteAclsRequestFilter) this).onDeleteAclsRequest((DeleteAclsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DELETE_GROUPS:
+                state = ((DeleteGroupsRequestFilter) this).onDeleteGroupsRequest((DeleteGroupsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DELETE_RECORDS:
+                state = ((DeleteRecordsRequestFilter) this).onDeleteRecordsRequest((DeleteRecordsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DELETE_TOPICS:
+                state = ((DeleteTopicsRequestFilter) this).onDeleteTopicsRequest((DeleteTopicsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DESCRIBE_ACLS:
+                state = ((DescribeAclsRequestFilter) this).onDescribeAclsRequest((DescribeAclsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DESCRIBE_CLIENT_QUOTAS:
+                state = ((DescribeClientQuotasRequestFilter) this).onDescribeClientQuotasRequest((DescribeClientQuotasRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DESCRIBE_CLUSTER:
+                state = ((DescribeClusterRequestFilter) this).onDescribeClusterRequest((DescribeClusterRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DESCRIBE_CONFIGS:
+                state = ((DescribeConfigsRequestFilter) this).onDescribeConfigsRequest((DescribeConfigsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DESCRIBE_DELEGATION_TOKEN:
+                state = ((DescribeDelegationTokenRequestFilter) this).onDescribeDelegationTokenRequest((DescribeDelegationTokenRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DESCRIBE_GROUPS:
+                state = ((DescribeGroupsRequestFilter) this).onDescribeGroupsRequest((DescribeGroupsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DESCRIBE_LOG_DIRS:
+                state = ((DescribeLogDirsRequestFilter) this).onDescribeLogDirsRequest((DescribeLogDirsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DESCRIBE_PRODUCERS:
+                state = ((DescribeProducersRequestFilter) this).onDescribeProducersRequest((DescribeProducersRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DESCRIBE_QUORUM:
+                state = ((DescribeQuorumRequestFilter) this).onDescribeQuorumRequest((DescribeQuorumRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DESCRIBE_TRANSACTIONS:
+                state = ((DescribeTransactionsRequestFilter) this).onDescribeTransactionsRequest((DescribeTransactionsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case DESCRIBE_USER_SCRAM_CREDENTIALS:
+                state = ((DescribeUserScramCredentialsRequestFilter) this).onDescribeUserScramCredentialsRequest((DescribeUserScramCredentialsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case ELECT_LEADERS:
+                state = ((ElectLeadersRequestFilter) this).onElectLeadersRequest((ElectLeadersRequestData) decodedFrame.body(), filterContext);
+                break;
+            case END_QUORUM_EPOCH:
+                state = ((EndQuorumEpochRequestFilter) this).onEndQuorumEpochRequest((EndQuorumEpochRequestData) decodedFrame.body(), filterContext);
+                break;
+            case END_TXN:
+                state = ((EndTxnRequestFilter) this).onEndTxnRequest((EndTxnRequestData) decodedFrame.body(), filterContext);
+                break;
+            case ENVELOPE:
+                state = ((EnvelopeRequestFilter) this).onEnvelopeRequest((EnvelopeRequestData) decodedFrame.body(), filterContext);
+                break;
+            case EXPIRE_DELEGATION_TOKEN:
+                state = ((ExpireDelegationTokenRequestFilter) this).onExpireDelegationTokenRequest((ExpireDelegationTokenRequestData) decodedFrame.body(), filterContext);
+                break;
+            case FETCH:
+                state = ((FetchRequestFilter) this).onFetchRequest((FetchRequestData) decodedFrame.body(), filterContext);
+                break;
+            case FETCH_SNAPSHOT:
+                state = ((FetchSnapshotRequestFilter) this).onFetchSnapshotRequest((FetchSnapshotRequestData) decodedFrame.body(), filterContext);
+                break;
+            case FIND_COORDINATOR:
+                state = ((FindCoordinatorRequestFilter) this).onFindCoordinatorRequest((FindCoordinatorRequestData) decodedFrame.body(), filterContext);
+                break;
+            case HEARTBEAT:
+                state = ((HeartbeatRequestFilter) this).onHeartbeatRequest((HeartbeatRequestData) decodedFrame.body(), filterContext);
+                break;
+            case INCREMENTAL_ALTER_CONFIGS:
+                state = ((IncrementalAlterConfigsRequestFilter) this).onIncrementalAlterConfigsRequest((IncrementalAlterConfigsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case INIT_PRODUCER_ID:
+                state = ((InitProducerIdRequestFilter) this).onInitProducerIdRequest((InitProducerIdRequestData) decodedFrame.body(), filterContext);
+                break;
+            case JOIN_GROUP:
+                state = ((JoinGroupRequestFilter) this).onJoinGroupRequest((JoinGroupRequestData) decodedFrame.body(), filterContext);
+                break;
+            case LEADER_AND_ISR:
+                state = ((LeaderAndIsrRequestFilter) this).onLeaderAndIsrRequest((LeaderAndIsrRequestData) decodedFrame.body(), filterContext);
+                break;
+            case LEAVE_GROUP:
+                state = ((LeaveGroupRequestFilter) this).onLeaveGroupRequest((LeaveGroupRequestData) decodedFrame.body(), filterContext);
+                break;
+            case LIST_GROUPS:
+                state = ((ListGroupsRequestFilter) this).onListGroupsRequest((ListGroupsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case LIST_OFFSETS:
+                state = ((ListOffsetsRequestFilter) this).onListOffsetsRequest((ListOffsetsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case LIST_PARTITION_REASSIGNMENTS:
+                state = ((ListPartitionReassignmentsRequestFilter) this).onListPartitionReassignmentsRequest((ListPartitionReassignmentsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case LIST_TRANSACTIONS:
+                state = ((ListTransactionsRequestFilter) this).onListTransactionsRequest((ListTransactionsRequestData) decodedFrame.body(), filterContext);
+                break;
+            case METADATA:
+                state = ((MetadataRequestFilter) this).onMetadataRequest((MetadataRequestData) decodedFrame.body(), filterContext);
+                break;
+            case OFFSET_COMMIT:
+                state = ((OffsetCommitRequestFilter) this).onOffsetCommitRequest((OffsetCommitRequestData) decodedFrame.body(), filterContext);
+                break;
+            case OFFSET_DELETE:
+                state = ((OffsetDeleteRequestFilter) this).onOffsetDeleteRequest((OffsetDeleteRequestData) decodedFrame.body(), filterContext);
+                break;
+            case OFFSET_FETCH:
+                state = ((OffsetFetchRequestFilter) this).onOffsetFetchRequest((OffsetFetchRequestData) decodedFrame.body(), filterContext);
+                break;
+            case OFFSET_FOR_LEADER_EPOCH:
+                state = ((OffsetForLeaderEpochRequestFilter) this).onOffsetForLeaderEpochRequest((OffsetForLeaderEpochRequestData) decodedFrame.body(), filterContext);
+                break;
+            case PRODUCE:
+                state = ((ProduceRequestFilter) this).onProduceRequest((ProduceRequestData) decodedFrame.body(), filterContext);
+                break;
+            case RENEW_DELEGATION_TOKEN:
+                state = ((RenewDelegationTokenRequestFilter) this).onRenewDelegationTokenRequest((RenewDelegationTokenRequestData) decodedFrame.body(), filterContext);
+                break;
+            case SASL_AUTHENTICATE:
+                state = ((SaslAuthenticateRequestFilter) this).onSaslAuthenticateRequest((SaslAuthenticateRequestData) decodedFrame.body(), filterContext);
+                break;
+            case SASL_HANDSHAKE:
+                state = ((SaslHandshakeRequestFilter) this).onSaslHandshakeRequest((SaslHandshakeRequestData) decodedFrame.body(), filterContext);
+                break;
+            case STOP_REPLICA:
+                state = ((StopReplicaRequestFilter) this).onStopReplicaRequest((StopReplicaRequestData) decodedFrame.body(), filterContext);
+                break;
+            case SYNC_GROUP:
+                state = ((SyncGroupRequestFilter) this).onSyncGroupRequest((SyncGroupRequestData) decodedFrame.body(), filterContext);
+                break;
+            case TXN_OFFSET_COMMIT:
+                state = ((TxnOffsetCommitRequestFilter) this).onTxnOffsetCommitRequest((TxnOffsetCommitRequestData) decodedFrame.body(), filterContext);
+                break;
+            case UNREGISTER_BROKER:
+                state = ((UnregisterBrokerRequestFilter) this).onUnregisterBrokerRequest((UnregisterBrokerRequestData) decodedFrame.body(), filterContext);
+                break;
+            case UPDATE_FEATURES:
+                state = ((UpdateFeaturesRequestFilter) this).onUpdateFeaturesRequest((UpdateFeaturesRequestData) decodedFrame.body(), filterContext);
+                break;
+            case UPDATE_METADATA:
+                state = ((UpdateMetadataRequestFilter) this).onUpdateMetadataRequest((UpdateMetadataRequestData) decodedFrame.body(), filterContext);
+                break;
+            case VOTE:
+                state = ((VoteRequestFilter) this).onVoteRequest((VoteRequestData) decodedFrame.body(), filterContext);
+                break;
+            case WRITE_TXN_MARKERS:
+                state = ((WriteTxnMarkersRequestFilter) this).onWriteTxnMarkersRequest((WriteTxnMarkersRequestData) decodedFrame.body(), filterContext);
+                break;
+            default:
+                throw new IllegalStateException("Unsupported RPC " + decodedFrame.apiKey());
+        }
+        return state;
+    }
+
+    /**
+     * <p>Determines whether a request with the given {@code apiKey} and {@code apiVersion} should be deserialized.
+     * Note that it is not guaranteed that this method will be called once per request,
+     * or that two consecutive calls refer to the same request.
+     * That is, the sequences of invocations like the following are allowed:</p>
+     * <ol>
+     *     <li>{@code shouldDeserializeRequest} on request A</li>
+     *     <li>{@code shouldDeserializeRequest} on request B</li>
+     *     <li>{@code shouldDeserializeRequest} on request A</li>
+     *     <li>{@code apply} on request A</li>
+     *     <li>{@code apply} on request B</li>
+     * </ol>
+     * @param apiKey The API key
+     * @param apiVersion The API version
+     * @return
+     */
+    default boolean shouldDeserializeRequest(ApiKeys apiKey, short apiVersion) {
+        switch (apiKey) {
+            case ADD_OFFSETS_TO_TXN:
+                return this instanceof AddOffsetsToTxnRequestFilter;
+            case ADD_PARTITIONS_TO_TXN:
+                return this instanceof AddPartitionsToTxnRequestFilter;
+            case ALLOCATE_PRODUCER_IDS:
+                return this instanceof AllocateProducerIdsRequestFilter;
+            case ALTER_CLIENT_QUOTAS:
+                return this instanceof AlterClientQuotasRequestFilter;
+            case ALTER_CONFIGS:
+                return this instanceof AlterConfigsRequestFilter;
+            case ALTER_PARTITION_REASSIGNMENTS:
+                return this instanceof AlterPartitionReassignmentsRequestFilter;
+            case ALTER_PARTITION:
+                return this instanceof AlterPartitionRequestFilter;
+            case ALTER_REPLICA_LOG_DIRS:
+                return this instanceof AlterReplicaLogDirsRequestFilter;
+            case ALTER_USER_SCRAM_CREDENTIALS:
+                return this instanceof AlterUserScramCredentialsRequestFilter;
+            case API_VERSIONS:
+                return this instanceof ApiVersionsRequestFilter;
+            case BEGIN_QUORUM_EPOCH:
+                return this instanceof BeginQuorumEpochRequestFilter;
+            case BROKER_HEARTBEAT:
+                return this instanceof BrokerHeartbeatRequestFilter;
+            case BROKER_REGISTRATION:
+                return this instanceof BrokerRegistrationRequestFilter;
+            case CONSUMER_GROUP_HEARTBEAT:
+                return this instanceof ConsumerGroupHeartbeatRequestFilter;
+            case CONTROLLED_SHUTDOWN:
+                return this instanceof ControlledShutdownRequestFilter;
+            case CREATE_ACLS:
+                return this instanceof CreateAclsRequestFilter;
+            case CREATE_DELEGATION_TOKEN:
+                return this instanceof CreateDelegationTokenRequestFilter;
+            case CREATE_PARTITIONS:
+                return this instanceof CreatePartitionsRequestFilter;
+            case CREATE_TOPICS:
+                return this instanceof CreateTopicsRequestFilter;
+            case DELETE_ACLS:
+                return this instanceof DeleteAclsRequestFilter;
+            case DELETE_GROUPS:
+                return this instanceof DeleteGroupsRequestFilter;
+            case DELETE_RECORDS:
+                return this instanceof DeleteRecordsRequestFilter;
+            case DELETE_TOPICS:
+                return this instanceof DeleteTopicsRequestFilter;
+            case DESCRIBE_ACLS:
+                return this instanceof DescribeAclsRequestFilter;
+            case DESCRIBE_CLIENT_QUOTAS:
+                return this instanceof DescribeClientQuotasRequestFilter;
+            case DESCRIBE_CLUSTER:
+                return this instanceof DescribeClusterRequestFilter;
+            case DESCRIBE_CONFIGS:
+                return this instanceof DescribeConfigsRequestFilter;
+            case DESCRIBE_DELEGATION_TOKEN:
+                return this instanceof DescribeDelegationTokenRequestFilter;
+            case DESCRIBE_GROUPS:
+                return this instanceof DescribeGroupsRequestFilter;
+            case DESCRIBE_LOG_DIRS:
+                return this instanceof DescribeLogDirsRequestFilter;
+            case DESCRIBE_PRODUCERS:
+                return this instanceof DescribeProducersRequestFilter;
+            case DESCRIBE_QUORUM:
+                return this instanceof DescribeQuorumRequestFilter;
+            case DESCRIBE_TRANSACTIONS:
+                return this instanceof DescribeTransactionsRequestFilter;
+            case DESCRIBE_USER_SCRAM_CREDENTIALS:
+                return this instanceof DescribeUserScramCredentialsRequestFilter;
+            case ELECT_LEADERS:
+                return this instanceof ElectLeadersRequestFilter;
+            case END_QUORUM_EPOCH:
+                return this instanceof EndQuorumEpochRequestFilter;
+            case END_TXN:
+                return this instanceof EndTxnRequestFilter;
+            case ENVELOPE:
+                return this instanceof EnvelopeRequestFilter;
+            case EXPIRE_DELEGATION_TOKEN:
+                return this instanceof ExpireDelegationTokenRequestFilter;
+            case FETCH:
+                return this instanceof FetchRequestFilter;
+            case FETCH_SNAPSHOT:
+                return this instanceof FetchSnapshotRequestFilter;
+            case FIND_COORDINATOR:
+                return this instanceof FindCoordinatorRequestFilter;
+            case HEARTBEAT:
+                return this instanceof HeartbeatRequestFilter;
+            case INCREMENTAL_ALTER_CONFIGS:
+                return this instanceof IncrementalAlterConfigsRequestFilter;
+            case INIT_PRODUCER_ID:
+                return this instanceof InitProducerIdRequestFilter;
+            case JOIN_GROUP:
+                return this instanceof JoinGroupRequestFilter;
+            case LEADER_AND_ISR:
+                return this instanceof LeaderAndIsrRequestFilter;
+            case LEAVE_GROUP:
+                return this instanceof LeaveGroupRequestFilter;
+            case LIST_GROUPS:
+                return this instanceof ListGroupsRequestFilter;
+            case LIST_OFFSETS:
+                return this instanceof ListOffsetsRequestFilter;
+            case LIST_PARTITION_REASSIGNMENTS:
+                return this instanceof ListPartitionReassignmentsRequestFilter;
+            case LIST_TRANSACTIONS:
+                return this instanceof ListTransactionsRequestFilter;
+            case METADATA:
+                return this instanceof MetadataRequestFilter;
+            case OFFSET_COMMIT:
+                return this instanceof OffsetCommitRequestFilter;
+            case OFFSET_DELETE:
+                return this instanceof OffsetDeleteRequestFilter;
+            case OFFSET_FETCH:
+                return this instanceof OffsetFetchRequestFilter;
+            case OFFSET_FOR_LEADER_EPOCH:
+                return this instanceof OffsetForLeaderEpochRequestFilter;
+            case PRODUCE:
+                return this instanceof ProduceRequestFilter;
+            case RENEW_DELEGATION_TOKEN:
+                return this instanceof RenewDelegationTokenRequestFilter;
+            case SASL_AUTHENTICATE:
+                return this instanceof SaslAuthenticateRequestFilter;
+            case SASL_HANDSHAKE:
+                return this instanceof SaslHandshakeRequestFilter;
+            case STOP_REPLICA:
+                return this instanceof StopReplicaRequestFilter;
+            case SYNC_GROUP:
+                return this instanceof SyncGroupRequestFilter;
+            case TXN_OFFSET_COMMIT:
+                return this instanceof TxnOffsetCommitRequestFilter;
+            case UNREGISTER_BROKER:
+                return this instanceof UnregisterBrokerRequestFilter;
+            case UPDATE_FEATURES:
+                return this instanceof UpdateFeaturesRequestFilter;
+            case UPDATE_METADATA:
+                return this instanceof UpdateMetadataRequestFilter;
+            case VOTE:
+                return this instanceof VoteRequestFilter;
+            case WRITE_TXN_MARKERS:
+                return this instanceof WriteTxnMarkersRequestFilter;
+            default:
+                throw new IllegalStateException("Unsupported API key " + apiKey);
+        }
+    }
+}
diff --git a/krpc-code-gen/src/test/resources/Kproxy/KrpcRequestFilter.ftl b/krpc-code-gen/src/test/resources/Kproxy/KrpcRequestFilter.ftl
index 3ef48d9c..3b018cf9 100644
--- a/krpc-code-gen/src/test/resources/Kproxy/KrpcRequestFilter.ftl
+++ b/krpc-code-gen/src/test/resources/Kproxy/KrpcRequestFilter.ftl
@@ -5,22 +5,12 @@
     Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 
 -->
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
+====
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+====
+
 package ${outputPackage};
 
 <#list messageSpecs as messageSpec>
diff --git a/krpc-code-gen/src/test/resources/hello-world/example-expected-FetchRequest.txt b/krpc-code-gen/src/test/resources/hello-world/example-expected-FetchRequest.txt
new file mode 100644
index 00000000..307354dc
--- /dev/null
+++ b/krpc-code-gen/src/test/resources/hello-world/example-expected-FetchRequest.txt
@@ -0,0 +1,86 @@
+====
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+====
+
+name: FetchRequest
+type: REQUEST
+apiKey: Optional[1]
+struct:
+  name: FetchRequest
+  hasKeys: no
+  versions: 0-15
+fields:
+    ClusterId
+    ReplicaId
+    ReplicaState
+    MaxWaitMs
+    MinBytes
+    MaxBytes
+    IsolationLevel
+    SessionId
+    SessionEpoch
+    Topics
+    ForgottenTopicsData
+    RackId
+validVersions: 0-15
+validVersionsString: 0-15
+flexibleVersions: 12+
+flexibleVersionsString: 12+
+dataClassName: FetchRequestData
+latestVersionUnstable: no
+fields:
+  name: ClusterId
+  versions: 12+
+  type: string
+  ignorable: yes
+  name: ReplicaId
+  versions: 0-14
+  type: int32
+  ignorable: no
+  name: ReplicaState
+  versions: 15+
+  type: ReplicaState
+  ignorable: no
+  name: MaxWaitMs
+  versions: 0+
+  type: int32
+  ignorable: no
+  name: MinBytes
+  versions: 0+
+  type: int32
+  ignorable: no
+  name: MaxBytes
+  versions: 3+
+  type: int32
+  ignorable: yes
+  name: IsolationLevel
+  versions: 4+
+  type: int8
+  ignorable: yes
+  name: SessionId
+  versions: 7+
+  type: int32
+  ignorable: yes
+  name: SessionEpoch
+  versions: 7+
+  type: int32
+  ignorable: yes
+  name: Topics
+  versions: 0+
+  type: []FetchTopic
+  ignorable: no
+  name: ForgottenTopicsData
+  versions: 7+
+  type: []ForgottenTopic
+  ignorable: no
+  name: RackId
+  versions: 11+
+  type: string
+  ignorable: yes
+commonStructs:
+listeners:
+  name: ZK_BROKER
+  name: BROKER
+  name: CONTROLLER
diff --git a/krpc-code-gen/src/test/resources/hello-world/example.ftl b/krpc-code-gen/src/test/resources/hello-world/example.ftl
index 3262932d..58bbc3e5 100644
--- a/krpc-code-gen/src/test/resources/hello-world/example.ftl
+++ b/krpc-code-gen/src/test/resources/hello-world/example.ftl
@@ -5,6 +5,12 @@
     Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 
 -->
+====
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+====
+
 name: ${messageSpec.name}
 type: ${messageSpec.type}
 apiKey: ${messageSpec.apiKey}
@@ -21,6 +27,7 @@ validVersionsString: ${messageSpec.validVersionsString}
 flexibleVersions: ${messageSpec.flexibleVersions}
 flexibleVersionsString: ${messageSpec.flexibleVersionsString}
 dataClassName: ${messageSpec.dataClassName}
+latestVersionUnstable: ${messageSpec.latestVersionUnstable.isPresent()?string('yes', 'no')}
 fields:
 <#list messageSpec.fields as field>
   name: ${field.name}