Repository: kroxylicious/kroxylicious
Pull request #243 (2023-03-30 03:21:41)

Add lower-level integration test checking kroxy can decode and proxy RPCs to a mock kafka broker
### Type of change

- Enhancement / new feature

### Description

Add `kroxylicious-test-tools`: a mock kafka server and client for testing kroxy

`MockServer` is a mock Kafka broker that can respond with a user defined `ApiMessage`. `MockServer` has a stripped down copy of the Kroxylicious codec internals for encoding and decoding requests/responses.
 
`KafkaClient` is a test client oriented around sending a single Kafka RPC to a Kafka broker, receiving a response and then closing its channel.

To test that Kroxy can decode and manipulate all RPCs we add a FixedClientIdFilter that changes the clientId in all request headers to some fixed value. This should demonstrate that kroxy can decode and modify any RPCs.

We add a `ProxyRPCTest` that executes scenarios for randomly generated examples of ApiMessage. Each scenario defines:
1. the response to serve from the mock
2. the request to send to kroxylicious from the client
3. the request expected to be received by the mock (from kroxylicious)
4. the response expected to be received by the client (from kroxylicious)

### Additional Context
This was spun off [a comment in another PR about testing](https://github.com/kroxylicious/kroxylicious/pull/201#discussion_r1148538541)

We want to (cheaply) test that kroxy is able to proxy and manipulate all RPCs, this is tricky with the high level clients as we are protected from the internals of sending kafka messages over the wire. Also with a real broker it's hard to verify what was sent over the wire from kroxy to broker.

One option is to replace the real broker with a mock. The mock broker should be Kafka API aware so it's possible to make it do more complex mocking later. Similarly with the client, we can use a simpler client that knows how to encode a request, send it, decode a response and shutdown it's channel to control of what is sent.

### design choices

I chose to copy the relevant codec classes rather than attempt to reuse them so that:
1. the mock code is independent of kroxy code, so we know which variable is changing when testing
2. the mock doesn't need some proxy code like correlation id rewriting, we can focus the code better.
3. prevents a circular dependency or the hassle of extracting a codec module, we could depend on these test-tools from kroxylicious

I explored making it possible to configure Kroxylicious in a "mock mode" with proxying disabled and Filters that return mock ApiMessage. This felt like fighting against the core proxying nature of kroxylicious and feels like it would make the code more complex to support a use-case that kroxy isn't built for. So I think an independent mock is the way to go.

robobario: I've implemented tests for four RPCs, generating the test data is a bit manual so I'll hold off doing a test for every RPC until this approach gets accepted or rejected
robobario: > I've implemented tests for four RPCs, generating the test data is a bit manual so I'll hold off doing a test for every RPC until this approach gets accepted or rejected

I've changed direction here after a conversation with @showuon. Instead of using a big YAML file for input we can generate sample request and response ApiMessages programmaticaly with reflection for all ApiKeys. So it's now running a test for all ApiKeys using randomly generated sample data, except for some special cases that aren't proxied/intercepted the same way (SASL requests, ApiVersions)
showuon: > I've changed direction here after a conversation with @showuon. Instead of using a big YAML file for input we can generate sample request and response ApiMessages programmaticaly with reflection for all ApiKeys. So it's now running a test for all ApiKeys using randomly generated sample data, except for some special cases that aren't proxied/intercepted the same way (SASL requests, ApiVersions)

Generated via a template, awesome!
robobario: And to give some direction @SamBarker 

main controversies to resolve:

- Independent code, should the mocks/client be totally independent despite having similar (but not identical) code to kroxy for encode/decode. I think the repetition is less evil than trying to bend kroxy towards supporting this other mocking case.
- JsonObject apis. The mock and client use JsonObject as their API with this idea of supporting tests that declare their expectations in YAML. But since shifting to generating the set of test ApiMessage programmatically it feels wrong. Maybe the mocks/client should operate on a Request and Response class and provide convenient ways to translate those classes to and from JsonObject.
robobario: > * YAML. But since shifting to generating the set of test ApiMessage programmatically it feels wrong. Maybe the mocks/client should operate on a Request and Response class and provide convenient ways to translate those classes to and from JsonNode.

I've changed it so MockServer and KafkaClient offer a Request and Response class instead of working with JsonNode. The integrationtests module is currently translating a few special cases from YAML to these Request/Response classes (when the broker address is manipulated by the BrokerAddress filter).
robobario: That was interesting, I guess the PR build runs against main with the branch merged in, so it had compile errors that weren't present in my branch.
k-wall: Couldn't we test the ability to encode/decode RPCs correctly with less code by adding unit tests around `Kafka(Request|Response)(En|De)coder`?
robobario: > I'm a bit uneasy about that amount of code this PR is proposing. Does it really bring sufficient benefit to warrant its existence? Couldn't we test the ability to encode/decode RPCs correctly with less code by adding unit tests around `Kafka(Request|Response)(En|De)coder`? I thinking about tests that ensured that every possible RPC could be decoded and recoded again, with fidelity.

I don't think of this mock and client as code that exists just to test RPC decode/encode. It's somewhere between unit tests and testing against a real kafka. We (Filter Authors) can test that our changes work against precise RPCs. We can test that outbound bytes to Kafka are modified as expected. All this can be done quickly and exercises kroxy's network code, filter encode/decode, proxying logic.

Thinking of the multi-tenant unit tests, if we can run the same tests in the context of a full kroxylicious without too much speed penalty, doesn't that give us more confidence that our filter will work? We can get earlier feedback that our Filter configuration works, we can check multiple Filters compose together if required.

Another example, if we wrote a SASL filter today using unit tests, our tests would pass. But Kroxy has special handling for SASL that is currently preventing decode of `SASL_HANDSHAKE` and `SASL_AUTHENTICATE` (aside, I don't know if this is intended when no SASL mechanism is configured). This level of testing catches that early.
robobario: No changes on that force-push, just rewrote the history down to 4 commits to remove fixups/flip-flopping between implementations.

Patch:
diff --git a/.github/workflows/performance.yml b/.github/workflows/performance.yml
index 81e34a7e..e720a8af 100644
--- a/.github/workflows/performance.yml
+++ b/.github/workflows/performance.yml
@@ -73,6 +73,24 @@ jobs:
 
       - name: 'Run kroxy'
         run: java -jar ${{ env.KROXY_JAR_FILE }} --config kroxylicious/example-proxy-config.yml &
+      
+      - name: 'await zookeeper'
+        uses: iFaxity/wait-on-action@v1
+        with:
+            resource: tcp:localhost:2181
+            timeout: 30000
+
+      - name: 'await kafka'
+        uses: iFaxity/wait-on-action@v1
+        with:
+            resource: tcp:localhost:9092
+            timeout: 30000
+      
+      - name: 'await kroxylicious'
+        uses: iFaxity/wait-on-action@v1
+        with:
+            resource: tcp:localhost:9192
+            timeout: 30000
 
       - name: 'Run warm-up'
         run: cd kafka_2.13; bin/kafka-producer-perf-test.sh --topic perf-test --throughput -1 --num-records 1000000 --record-size 1024 --producer-props acks=all bootstrap.servers=localhost:9192
@@ -144,6 +162,24 @@ jobs:
 
       - name: 'Run kroxy'
         run: java -jar ${{ env.KROXY_JAR_FILE }} --config kroxylicious/example-proxy-config.yml &
+      
+      - name: 'await zookeeper'
+        uses: iFaxity/wait-on-action@v1
+        with:
+            resource: tcp:localhost:2181
+            timeout: 30000
+
+      - name: 'await kafka'
+        uses: iFaxity/wait-on-action@v1
+        with:
+            resource: tcp:localhost:9092
+            timeout: 30000
+      
+      - name: 'await kroxylicious'
+        uses: iFaxity/wait-on-action@v1
+        with:
+            resource: tcp:localhost:9192
+            timeout: 30000
 
       - name: 'Run warm-up'
         run: cd kafka_2.13; bin/kafka-producer-perf-test.sh --topic perf-test --throughput -1 --num-records 1000000 --record-size 1024 --producer-props acks=all bootstrap.servers=localhost:9192
diff --git a/api/kroxylicious-api/kroxylicious-api.iml b/api/kroxylicious-api/kroxylicious-api.iml
new file mode 100644
index 00000000..9e3449c9
--- /dev/null
+++ b/api/kroxylicious-api/kroxylicious-api.iml
@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module version="4">
+  <component name="CheckStyle-IDEA-Module" serialisationVersion="2">
+    <option name="activeLocationsIds" />
+  </component>
+</module>
\ No newline at end of file
diff --git a/kroxylicious-parent.iml b/kroxylicious-parent.iml
new file mode 100644
index 00000000..9e3449c9
--- /dev/null
+++ b/kroxylicious-parent.iml
@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module version="4">
+  <component name="CheckStyle-IDEA-Module" serialisationVersion="2">
+    <option name="activeLocationsIds" />
+  </component>
+</module>
\ No newline at end of file
diff --git a/kroxylicious/kroxylicious.iml b/kroxylicious/kroxylicious.iml
new file mode 100644
index 00000000..9e3449c9
--- /dev/null
+++ b/kroxylicious/kroxylicious.iml
@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module version="4">
+  <component name="CheckStyle-IDEA-Module" serialisationVersion="2">
+    <option name="activeLocationsIds" />
+  </component>
+</module>
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index c9c5bf92..db4d6bec 100644
--- a/pom.xml
+++ b/pom.xml
@@ -221,6 +221,7 @@
         <module>api/kroxylicious-api</module>
         <module>api/kroxylicious-filter-api</module>
         <module>krpc-code-gen</module>
+        <module>kroxylicious-test-tools</module>
         <module>kroxylicious</module>
         <module>kroxylicious-multitenant</module>
         <module>integrationtests</module>


Test patch:
diff --git a/integrationtests/pom.xml b/integrationtests/pom.xml
index b5871409..5937e945 100644
--- a/integrationtests/pom.xml
+++ b/integrationtests/pom.xml
@@ -31,6 +31,12 @@
             <version>0.2.0-SNAPSHOT</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-test-tools</artifactId>
+            <version>0.2.0-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-multitenant</artifactId>
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java b/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
new file mode 100644
index 00000000..891f34f7
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
@@ -0,0 +1,144 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import org.apache.kafka.common.message.ApiMessageType;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import io.kroxylicious.test.ApiMessageSampleGenerator;
+import io.kroxylicious.test.ApiMessageSampleGenerator.ApiAndVersion;
+import io.kroxylicious.test.Request;
+import io.kroxylicious.test.Response;
+import io.kroxylicious.test.client.KafkaClient;
+import io.kroxylicious.test.server.MockServer;
+
+import static io.kroxylicious.proxy.Utils.startProxy;
+import static org.apache.kafka.common.protocol.ApiKeys.API_VERSIONS;
+import static org.apache.kafka.common.protocol.ApiKeys.CONTROLLED_SHUTDOWN;
+import static org.apache.kafka.common.protocol.ApiKeys.SASL_AUTHENTICATE;
+import static org.apache.kafka.common.protocol.ApiKeys.SASL_HANDSHAKE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.fail;
+
+public class ProxyRpcTest {
+
+    public static final int PROXY_PORT = 9192;
+    public static final String PROXY_ADDRESS = "localhost:" + PROXY_PORT;
+    private static MockServer mockServer;
+    private static KafkaClient kafkaClient;
+    private static KafkaProxy kafkaProxy;
+
+    public record Scenario(String name, Response givenMockResponse, Request whenSendRequest, Request thenMockReceivesRequest, Response thenResponseReceived) {
+    }
+
+    /**
+     * API_VERSIONS is not proxied, kroxy can respond to this itself
+     */
+    private static final Set<ApiKeys> NOT_PROXIED = Set.of(API_VERSIONS);
+
+    @BeforeAll
+    public static void beforeAll() throws InterruptedException {
+        mockServer = MockServer.startOnRandomPort();
+        kafkaClient = new KafkaClient();
+        String config = configure(mockServer);
+        kafkaProxy = startProxy(config);
+    }
+
+    @MethodSource("scenarios")
+    @ParameterizedTest
+    public void testKroxyliciousCanDecodeManipulateAndProxyRPC(Scenario scenario) throws Exception {
+        mockServer.clear();
+        mockServer.setResponse(scenario.givenMockResponse());
+        CompletableFuture<Response> future = kafkaClient.get("localhost", PROXY_PORT, scenario.whenSendRequest());
+        Response response = future.get(10, TimeUnit.SECONDS);
+        assertEquals(scenario.thenMockReceivesRequest(), onlyRequest(mockServer), "unexpected request received at mock for scenario: " + scenario.name());
+        assertEquals(scenario.thenResponseReceived(), response, "unexpected response received from kroxylicious for scenario: " + scenario.name());
+    }
+
+    @AfterAll
+    public static void afterAll() throws Exception {
+        kafkaProxy.close();
+        mockServer.close();
+        kafkaClient.close();
+    }
+
+    @NotNull
+    private static Stream<Scenario> scenarios() {
+        Map<ApiAndVersion, ApiMessage> requestSamples = ApiMessageSampleGenerator.createRequestSamples();
+        Map<ApiAndVersion, ApiMessage> responseSamples = ApiMessageSampleGenerator.createResponseSamples();
+        return Arrays.stream(ApiKeys.values()).filter(apiKeys -> !NOT_PROXIED.contains(apiKeys)).flatMap(apiKeys -> toScenario(requestSamples, responseSamples, apiKeys));
+    }
+
+    private static final ApiAndVersion v0HeaderVersion = new ApiAndVersion(CONTROLLED_SHUTDOWN, (short) 0);
+
+    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);
+        return supported.mapToObj(version -> new ApiAndVersion(apiKeys, (short) version)).map(apiAndVersion -> {
+            ApiMessage request = requestSamples.get(apiAndVersion);
+            ApiMessage response = responseSample.get(apiAndVersion);
+            Request clientRequest = createRequestDefinition(apiAndVersion, "mockClientId", request);
+            String expected;
+            if (v0HeaderVersion.equals(apiAndVersion)) {
+                // controlled shutdown is the only usage of a version 0 header schema which doesn't have clientId
+                expected = "";
+            }
+            else if (apiAndVersion.keys() == SASL_AUTHENTICATE || apiAndVersion.keys() == SASL_HANDSHAKE) {
+                // sasl requests have special logic in kroxy, and they do not get Filtered
+                expected = "mockClientId";
+            }
+            else {
+                expected = "fixed";
+            }
+            Request expectedAtMock = createRequestDefinition(apiAndVersion, expected, request);
+            Response responseJson = createResponseDefinition(apiAndVersion, response);
+            return new Scenario(apiKeys.name, responseJson, clientRequest, expectedAtMock, responseJson);
+        });
+    }
+
+    @NotNull
+    private static Response createResponseDefinition(ApiAndVersion apiAndVersion, ApiMessage message) {
+        return new Response(apiAndVersion.keys(), apiAndVersion.apiVersion(), message);
+    }
+
+    @NotNull
+    private static Request createRequestDefinition(ApiAndVersion apiKeys, String clientId, ApiMessage requestBody) {
+        return new Request(apiKeys.keys(), apiKeys.apiVersion(), clientId, requestBody);
+    }
+
+    private static String configure(MockServer mockServer) {
+        return KroxyConfig.builder().withNewProxy().withAddress(ProxyRpcTest.PROXY_ADDRESS).endProxy()
+                .addToClusters("demo", new ClusterBuilder().withBootstrapServers("localhost:" + mockServer.port()).build())
+                .addNewFilter().withType("ApiVersions").endFilter()
+                .addNewFilter().withType("FixedClientId").withConfig(Map.of("clientId", "fixed")).endFilter()
+                .build().toYaml();
+    }
+
+    private static Request onlyRequest(MockServer mockServer) {
+        List<Request> requests = mockServer.getReceivedRequests();
+        if (requests.size() != 1) {
+            fail("mock server does not have exactly one request recorded");
+        }
+        return requests.get(0);
+    }
+
+}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/FixedClientIdFilter.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/FixedClientIdFilter.java
new file mode 100644
index 00000000..55e884f8
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/FixedClientIdFilter.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.filter;
+
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+
+import io.kroxylicious.proxy.config.BaseConfig;
+
+public class FixedClientIdFilter implements KrpcFilter {
+
+    private final String clientId;
+
+    public static class FixedClientIdFilterConfig extends BaseConfig {
+
+        private final String clientId;
+
+        public FixedClientIdFilterConfig(String clientId) {
+            this.clientId = clientId;
+        }
+
+        public String getClientId() {
+            return clientId;
+        }
+    }
+
+    FixedClientIdFilter(FixedClientIdFilterConfig config) {
+        this.clientId = config.getClientId();
+    }
+
+    @Override
+    public boolean shouldDeserializeRequest(ApiKeys apiKey, short apiVersion) {
+        return true;
+    }
+
+    @Override
+    public boolean shouldDeserializeResponse(ApiKeys apiKey, short apiVersion) {
+        return true;
+    }
+
+    @Override
+    public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        header.setClientId(clientId);
+        filterContext.forwardRequest(body);
+    }
+
+    @Override
+    public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        filterContext.forwardResponse(body);
+    }
+}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java
new file mode 100644
index 00000000..45e28f80
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java
@@ -0,0 +1,18 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.proxy.filter;
+
+import io.kroxylicious.proxy.service.BaseContributor;
+
+public class TestFilterContributor extends BaseContributor<KrpcFilter> implements FilterContributor {
+
+    public static final BaseContributorBuilder<KrpcFilter> FILTERS = BaseContributor.<KrpcFilter> builder()
+            .add("FixedClientId", FixedClientIdFilter.FixedClientIdFilterConfig.class, FixedClientIdFilter::new);
+
+    public TestFilterContributor() {
+        super(FILTERS);
+    }
+}
diff --git a/integrationtests/src/test/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterContributor b/integrationtests/src/test/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterContributor
new file mode 100644
index 00000000..3d64e347
--- /dev/null
+++ b/integrationtests/src/test/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterContributor
@@ -0,0 +1,1 @@
+io.kroxylicious.proxy.filter.TestFilterContributor
diff --git a/kroxylicious-test-tools/kroxylicious-test-tools.iml b/kroxylicious-test-tools/kroxylicious-test-tools.iml
new file mode 100644
index 00000000..9e3449c9
--- /dev/null
+++ b/kroxylicious-test-tools/kroxylicious-test-tools.iml
@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module version="4">
+  <component name="CheckStyle-IDEA-Module" serialisationVersion="2">
+    <option name="activeLocationsIds" />
+  </component>
+</module>
\ No newline at end of file
diff --git a/kroxylicious-test-tools/pom.xml b/kroxylicious-test-tools/pom.xml
new file mode 100644
index 00000000..146e32b2
--- /dev/null
+++ b/kroxylicious-test-tools/pom.xml
@@ -0,0 +1,147 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/maven-v4_0_0.xsd">
+
+    <modelVersion>4.0.0</modelVersion>
+
+    <packaging>jar</packaging>
+    <parent>
+        <groupId>io.kroxylicious</groupId>
+        <artifactId>kroxylicious-parent</artifactId>
+        <version>0.2.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+    <artifactId>kroxylicious-test-tools</artifactId>
+    <name>kroxylicious-test-tools</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-common</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-buffer</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-transport</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-codec</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-handler</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>unpack-message-specs</id>
+                        <goals>
+                            <goal>unpack</goal>
+                        </goals>
+                        <phase>generate-sources</phase>
+                        <configuration>
+                            <artifact>org.apache.kafka:kafka-clients:${kafka.version}</artifact>
+                            <includes>common/message/*.json</includes>
+                            <outputDirectory>${project.build.directory}/message-specs</outputDirectory>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>io.kroxylicious</groupId>
+                <artifactId>kroxylicious-krpc-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>generate-request-decoder</id>
+                        <goals>
+                            <goal>generate-multi</goal>
+                        </goals>
+                        <phase>generate-sources</phase>
+                        <configuration>
+                            <messageSpecDirectory>${project.build.directory}/message-specs/common/message
+                            </messageSpecDirectory>
+                            <messageSpecFilter>*{Request,Response}.json</messageSpecFilter>
+                            <templateDirectory>${project.basedir}/src/main/templates</templateDirectory>
+                            <templateNames>BodyDecoder.ftl</templateNames>
+                            <!--suppress UnresolvedMavenProperty -->
+                            <outputFilePattern>${templateName}.java</outputFilePattern>
+                            <outputPackage>io.kroxylicious.test.codec</outputPackage>
+                            <outputDirectory>${project.build.directory}/generated-sources/krpc</outputDirectory>
+                        </configuration>
+                    </execution>
+                    <execution>
+                        <id>generate-data-classes</id>
+                        <goals>
+                            <goal>generate-multi</goal>
+                        </goals>
+                        <phase>generate-sources</phase>
+                        <configuration>
+                            <messageSpecDirectory>${project.build.directory}/message-specs/common/message
+                            </messageSpecDirectory>
+                            <messageSpecFilter>*{Request,Response}.json</messageSpecFilter>
+                            <templateDirectory>${project.basedir}/src/main/templates</templateDirectory>
+                            <templateNames>DataClasses.ftl</templateNames>
+                            <!--suppress UnresolvedMavenProperty -->
+                            <outputFilePattern>${templateName}.java</outputFilePattern>
+                            <outputPackage>io.kroxylicious.test</outputPackage>
+                            <outputDirectory>${project.build.directory}/generated-sources/krpc</outputDirectory>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>
+
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
new file mode 100644
index 00000000..4298f3cf
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/ApiMessageSampleGenerator.java
@@ -0,0 +1,273 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.test;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.Message;
+import org.apache.kafka.common.protocol.types.BoundField;
+import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.record.BaseRecords;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.MemoryRecordsBuilder;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.record.SimpleRecord;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.utils.ImplicitLinkedHashCollection;
+
+/**
+ * Generates an ApiMessage instance per ApiKey. The message is built
+ * with reflection and populated with randomised data. The data is built
+ * from the same random seed and the ApiKeys and methods are iterated over
+ * in a deterministic order so the output of `createRequestSamples` should
+ * be consistent.
+ */
+public class ApiMessageSampleGenerator {
+
+    public record ApiAndVersion(ApiKeys keys, short apiVersion) {
+
+    }
+
+    public static final int RANGE_MIN = 0;
+    public static final int RANGE_MAX = 1024;
+
+    private ApiMessageSampleGenerator() {
+
+    }
+
+    /**
+     * Generates a sample request ApiMessage for all ApiKeys
+     * @return ApiKeys to message
+     */
+    static public Map<ApiAndVersion, ApiMessage> createRequestSamples() {
+        Random random = new Random(0);
+        return instantiateAll(DataClasses.getRequestClasses(), random);
+    }
+
+    /**
+     * Generates a sample response ApiMessage for all ApiKeys
+     * @return ApiKeys to message
+     */
+    static public Map<ApiAndVersion, ApiMessage> createResponseSamples() {
+        Random random = new Random(0);
+        return instantiateAll(DataClasses.getResponseClasses(), random);
+    }
+
+    private static Map<ApiAndVersion, ApiMessage> instantiateAll(Map<ApiKeys, Class<? extends ApiMessage>> messages, Random random) {
+        return messages.entrySet().stream().sorted(Comparator.comparing(apiKeysClassEntry -> apiKeysClassEntry.getKey().name))
+                .flatMap(ApiMessageSampleGenerator::getSupportedVersions)
+                .flatMap(entry -> instantiateSample(entry, random)).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    }
+
+    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();
+        return IntStream.range(lowestSupportedVersion, highestSupportedVersion + 1).mapToObj(version -> new AbstractMap.SimpleEntry<>(
+                new ApiAndVersion(apiKeysClassEntry.getKey(), (short) version), apiKeysClassEntry.getValue()));
+    }
+
+    private static Stream<Map.Entry<ApiAndVersion, ApiMessage>> instantiateSample(Map.Entry<ApiAndVersion, Class<? extends ApiMessage>> entry, Random random) {
+        try {
+            Field schemasField = entry.getValue().getDeclaredField("SCHEMAS");
+            Schema[] schemas = (Schema[]) schemasField.get(null);
+            Field lowestSupportedVersion = entry.getValue().getDeclaredField("LOWEST_SUPPORTED_VERSION");
+            short lowestVersion = (short) lowestSupportedVersion.get(null);
+            Field highestSupportedVersion = entry.getValue().getDeclaredField("HIGHEST_SUPPORTED_VERSION");
+            short highestVersion = (short) highestSupportedVersion.get(null);
+            short apiVersion = entry.getKey().apiVersion;
+            if (apiVersion < lowestVersion || apiVersion > highestVersion) {
+                return Stream.of();
+            }
+            Schema schema = schemas[apiVersion];
+            ApiMessage message = (ApiMessage) instantiate(entry.getValue(), random, schema);
+            return Stream.of(new AbstractMap.SimpleEntry<>(entry.getKey(), message));
+        }
+        catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    static private Message instantiate(Class<? extends Message> clazz, Random random, Schema schema) {
+        try {
+            Message instance = clazz.getConstructor().newInstance();
+            Map<String, org.apache.kafka.common.protocol.types.Field> fieldsForVersion = Arrays.stream(schema.fields())
+                    .filter(boundField -> !boundField.def.name.startsWith("_"))
+                    .collect(Collectors.toMap(ApiMessageSampleGenerator::toSetterName, boundField -> boundField.def));
+
+            Stream<Method> sortedMethods = Arrays.stream(clazz.getMethods())
+                    .filter(method -> method.getName().startsWith("set"))
+                    .filter(method -> fieldsForVersion.containsKey(method.getName()))
+                    .sorted(Comparator.comparing(Method::getName));
+            sortedMethods.forEach(method -> {
+                try {
+                    Object o = instantiateArg(method, random, fieldsForVersion.get(method.getName()).type);
+                    method.invoke(instance, o);
+                }
+                catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            });
+            return instance;
+        }
+        catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static String toSetterName(BoundField boundField) {
+        return "set" + Arrays.stream(boundField.def.name.split("_")).map(s -> s.substring(0, 1).toUpperCase() + s.substring(1)).collect(Collectors.joining());
+    }
+
+    static private Object instantiateArg(Class<?> paramClass, Type paramGenericType, Random random, org.apache.kafka.common.protocol.types.Type type) {
+        if (paramClass == long.class || paramClass == Long.class) {
+            return random.nextLong(RANGE_MIN, RANGE_MAX);
+        }
+        else if (paramClass == int.class || paramClass == Integer.class) {
+            return random.nextInt(RANGE_MIN, RANGE_MAX);
+        }
+        else if (paramClass == short.class || paramClass == Short.class) {
+            return (short) random.nextInt(RANGE_MIN, RANGE_MAX);
+        }
+        else if (paramClass == double.class || paramClass == Double.class) {
+            return random.nextDouble(RANGE_MIN, RANGE_MAX);
+        }
+        else if (paramClass == String.class) {
+            return "random-string-" + random.nextInt(RANGE_MIN, RANGE_MAX);
+        }
+        else if (paramClass == byte.class || paramClass == Byte.class) {
+            return (byte) random.nextInt(RANGE_MIN, 127);
+        }
+        else if (paramClass == byte[].class) {
+            byte[] bytes = new byte[5];
+            random.nextBytes(bytes);
+            return bytes;
+        }
+        else if (paramClass == ByteBuffer.class) {
+            byte[] bytes = new byte[5];
+            random.nextBytes(bytes);
+            return ByteBuffer.wrap(bytes).flip();
+        }
+        else if (paramClass == boolean.class) {
+            return random.nextBoolean();
+        }
+        else if (paramClass == Uuid.class) {
+            return randomUuid(random);
+        }
+        else if (paramClass == BaseRecords.class) {
+            return randomMemoryRecords(random);
+        }
+        else if (Message.class.isAssignableFrom(paramClass)) {
+            return instantiate(paramClass.asSubclass(Message.class), random, (Schema) type);
+        }
+        else if (List.class.isAssignableFrom(paramClass)) {
+            return instantiateList(paramGenericType, random, type);
+        }
+        else if (ImplicitLinkedHashCollection.class.isAssignableFrom(paramClass)) {
+            return instantiateMessageCollection(paramClass.asSubclass(ImplicitLinkedHashCollection.class), random, type);
+        }
+        else {
+            throw new IllegalArgumentException("unexpected type " + paramClass.getSimpleName());
+        }
+    }
+
+    private static MemoryRecords randomMemoryRecords(Random random) {
+        String key = "key-" + random.nextInt(RANGE_MIN, RANGE_MAX);
+        String value = "value-" + random.nextInt(RANGE_MIN, RANGE_MAX);
+        long baseOffset = random.nextInt(RANGE_MIN, RANGE_MAX);
+        long logAppendTime = RecordBatch.NO_TIMESTAMP;
+        try (MemoryRecordsBuilder builder = MemoryRecords.builder(ByteBuffer.allocate(1), RecordBatch.CURRENT_MAGIC_VALUE, CompressionType.NONE,
+                TimestampType.CREATE_TIME, baseOffset,
+                logAppendTime,
+                RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, false,
+                RecordBatch.NO_PARTITION_LEADER_EPOCH)) {
+            builder.append(new SimpleRecord(random.nextLong(RANGE_MIN, RANGE_MAX), key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8)));
+            return builder.build();
+        }
+    }
+
+    /**
+     * Takes the guts of the JVM randomUUID and byte[] constructor
+     * so that we can create a random UUID using our random
+     * implementation. This is so that all random data used in generation
+     * comes from the same random seed to generate deterministic samples.
+     * @param random the random to use
+     * @return a random Uuid
+     */
+    private static Uuid randomUuid(Random random) {
+        byte[] randomBytes = new byte[16];
+        random.nextBytes(randomBytes);
+        randomBytes[6] &= 0x0f; /* clear version */
+        randomBytes[6] |= 0x40; /* set to version 4 */
+        randomBytes[8] &= 0x3f; /* clear variant */
+        randomBytes[8] |= 0x80; /* set to IETF variant */
+        long mostSignificantBits = 0;
+        long leastSignificantBits = 0;
+        for (int i = 0; i < 8; i++) {
+            mostSignificantBits = (mostSignificantBits << 8) | (randomBytes[i] & 0xff);
+        }
+        for (int i = 8; i < 16; i++) {
+            leastSignificantBits = (leastSignificantBits << 8) | (randomBytes[i] & 0xff);
+        }
+        return new Uuid(mostSignificantBits, leastSignificantBits);
+    }
+
+    private static Object instantiateMessageCollection(Class<? extends ImplicitLinkedHashCollection> genericType, Random random,
+                                                       org.apache.kafka.common.protocol.types.Type field) {
+        try {
+            ImplicitLinkedHashCollection<ImplicitLinkedHashCollection.Element> collection = genericType.getConstructor().newInstance();
+            ParameterizedType paramClass1 = (ParameterizedType) genericType.getGenericSuperclass();
+            Type actualTypeArgument = paramClass1.getActualTypeArguments()[0];
+            Class<?> typeArgument = (Class<?>) actualTypeArgument;
+            Object o = instantiateArg(typeArgument, actualTypeArgument, random, field.arrayElementType().get());
+            collection.add((ImplicitLinkedHashCollection.Element) o);
+            return collection;
+        }
+        catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static Object instantiateList(Type paramClass, Random random, org.apache.kafka.common.protocol.types.Type type) {
+        ArrayList<Object> objects = new ArrayList<>();
+        ParameterizedType paramClass1 = (ParameterizedType) paramClass;
+        Type actualTypeArgument = paramClass1.getActualTypeArguments()[0];
+        Class<?> typeArgument = (Class<?>) actualTypeArgument;
+        objects.add(instantiateArg(typeArgument, actualTypeArgument, random, type.arrayElementType().get()));
+        return objects;
+    }
+
+    static private Object instantiateArg(Method method, Random random, org.apache.kafka.common.protocol.types.Type type) {
+        int parameterCount = method.getParameterCount();
+        if (parameterCount != 1) {
+            throw new IllegalArgumentException("setter takes more than one arg!");
+        }
+        Class<?> parameterType = method.getParameterTypes()[0];
+        Type genericParameterType = method.getGenericParameterTypes()[0];
+        return instantiateArg(parameterType, genericParameterType, random, type);
+    }
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/Request.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/Request.java
new file mode 100644
index 00000000..2ee07b18
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/Request.java
@@ -0,0 +1,17 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.test;
+
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+
+public record Request(ApiKeys apiKeys,
+                      short apiVersion,
+                      String clientIdHeader,
+                      ApiMessage message) {
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/Response.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/Response.java
new file mode 100644
index 00000000..2339d0a9
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/Response.java
@@ -0,0 +1,16 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.test;
+
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+
+public record Response(ApiKeys apiKeys,
+                       short apiVersion,
+                       ApiMessage message) {
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/client/CorrelationManager.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/client/CorrelationManager.java
new file mode 100644
index 00000000..bf94f455
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/client/CorrelationManager.java
@@ -0,0 +1,112 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.client;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tracks api version for requests
+ */
+public class CorrelationManager {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(CorrelationManager.class);
+
+    final Map<Integer, Correlation> brokerRequests = new HashMap<>();
+
+    /**
+     * Creates an empty CorrelationManager
+     */
+    public CorrelationManager() {
+    }
+
+    /**
+     * Allocate and return a correlation id for an outgoing request to the broker.
+     *
+     * @param apiKey                  The API key.
+     * @param apiVersion              The API version.
+     * @param correlationId The request's correlation id.
+     */
+    public void putBrokerRequest(short apiKey,
+                                 short apiVersion,
+                                 int correlationId) {
+        Correlation existing = this.brokerRequests.put(correlationId, new Correlation(apiKey, apiVersion));
+        if (existing != null) {
+            LOGGER.error("Duplicate upstream correlation id {}", correlationId);
+        }
+    }
+
+    /**
+     * Find (and remove) the Correlation for an incoming response from the broker
+     * @param correlationId The correlation id in the response.
+     * @return the correlation
+     */
+    public Correlation getBrokerCorrelation(int correlationId) {
+        return brokerRequests.remove(correlationId);
+    }
+
+    /**
+     * A record for which responses should be decoded, together with their
+     * API key and version.
+     */
+    // TODO a perfect value type
+    public static class Correlation {
+        private final short apiKey;
+        private final short apiVersion;
+
+        private Correlation(short apiKey,
+                            short apiVersion) {
+            this.apiKey = apiKey;
+            this.apiVersion = apiVersion;
+        }
+
+        @Override
+        public String toString() {
+            return "Correlation(" +
+                    "apiKey=" + ApiKeys.forId(apiKey) +
+                    ", apiVersion=" + apiVersion +
+                    ')';
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) {
+                return true;
+            }
+            if (o == null || getClass() != o.getClass()) {
+                return false;
+            }
+            Correlation that = (Correlation) o;
+            return apiKey == that.apiKey && apiVersion == that.apiVersion;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(apiKey, apiVersion);
+        }
+
+        /**
+         * get the api key of the request
+         * @return api key
+         */
+        public short apiKey() {
+            return apiKey;
+        }
+
+        /**
+         * get the api version of the request
+         * @return api version
+         */
+        public short apiVersion() {
+            return apiVersion;
+        }
+    }
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/client/KafkaClient.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/client/KafkaClient.java
new file mode 100644
index 00000000..1a0c3d74
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/client/KafkaClient.java
@@ -0,0 +1,104 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.test.client;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.kafka.common.message.RequestHeaderData;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+
+import io.kroxylicious.test.Request;
+import io.kroxylicious.test.Response;
+import io.kroxylicious.test.codec.DecodedRequestFrame;
+import io.kroxylicious.test.codec.DecodedResponseFrame;
+import io.kroxylicious.test.codec.KafkaRequestEncoder;
+import io.kroxylicious.test.codec.KafkaResponseDecoder;
+
+/**
+ * KafkaClient for testing.
+ * <p>
+ * The kafka client closes its channel after it gets a response. A new
+ * connection is bootstrapped on every call to `get`.
+ * </p>
+ * <p>
+ * The intention is that it should be used to fire a single RPC at a server
+ * offering the Kafka protocol, read a response and inform the client of that
+ * response. It currently translates the response to a normalised JsonObject.
+ * </p>
+ */
+public final class KafkaClient implements AutoCloseable {
+
+    /**
+     * create empty kafkaClient
+     */
+    public KafkaClient() {
+    }
+
+    private final EventLoopGroup group = new NioEventLoopGroup();
+
+    private static final AtomicInteger correlationId = new AtomicInteger(1);
+
+    private static DecodedRequestFrame<?> toApiRequest(Request request) {
+        var messageType = request.apiKeys().messageType;
+        var header = new RequestHeaderData().setRequestApiKey(messageType.apiKey()).setRequestApiVersion(request.apiVersion());
+        header.setClientId(request.clientIdHeader());
+        header.setCorrelationId(correlationId.incrementAndGet());
+        return new DecodedRequestFrame<>(header.requestApiVersion(), header.correlationId(), header, request.message());
+    }
+
+    // TODO return a Response class with jsonObject() and frame() methods
+
+    /**
+     * Bootstrap and connect to a Kafka broker on a given host and port. Send
+     * the request to it and inform the client when we have received a response.
+     * The channel is closed after we have received the message.
+     * @param host kafka broker host
+     * @param port kafka broker port
+     * @param request request to send to kafka
+     * @return a future that will be completed with the response from the kafka broker (translated to JsonNode)
+     */
+    public CompletableFuture<Response> get(String host, int port, Request request) {
+        DecodedRequestFrame<?> decodedRequestFrame = toApiRequest(request);
+        CorrelationManager correlationManager = new CorrelationManager();
+        KafkaClientHandler kafkaClientHandler = new KafkaClientHandler(decodedRequestFrame);
+        Bootstrap b = new Bootstrap();
+        b.group(group)
+                .channel(NioSocketChannel.class)
+                .option(ChannelOption.TCP_NODELAY, true)
+                .handler(new ChannelInitializer<SocketChannel>() {
+                    @Override
+                    public void initChannel(SocketChannel ch) {
+                        ChannelPipeline p = ch.pipeline();
+                        p.addLast(new KafkaRequestEncoder(correlationManager));
+                        p.addLast(new KafkaResponseDecoder(correlationManager));
+                        p.addLast(kafkaClientHandler);
+                    }
+                });
+
+        b.connect(host, port);
+        CompletableFuture<DecodedResponseFrame<?>> onResponseFuture = kafkaClientHandler.getOnResponseFuture();
+        return onResponseFuture.thenApply(KafkaClient::toResponse);
+    }
+
+    private static Response toResponse(DecodedResponseFrame<?> decodedResponseFrame) {
+        return new Response(decodedResponseFrame.apiKey(), decodedResponseFrame.apiVersion(), decodedResponseFrame.body());
+    }
+
+    @Override
+    public void close() {
+        group.shutdownGracefully();
+    }
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/client/KafkaClientHandler.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/client/KafkaClientHandler.java
new file mode 100644
index 00000000..2c91b59e
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/client/KafkaClientHandler.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.client;
+
+import java.util.concurrent.CompletableFuture;
+
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+import io.kroxylicious.test.codec.DecodedRequestFrame;
+import io.kroxylicious.test.codec.DecodedResponseFrame;
+
+/**
+ * Sends a single request frame, waits for a response then closes the channel
+ */
+public class KafkaClientHandler extends ChannelInboundHandlerAdapter {
+    private final DecodedRequestFrame<?> decodedRequestFrame;
+    private final CompletableFuture<DecodedResponseFrame<?>> onResponse = new CompletableFuture<>();
+
+    /**
+     * Creates a KafkaClientHandler
+     * @param decodedRequestFrame the single request to send
+     */
+    public KafkaClientHandler(DecodedRequestFrame<?> decodedRequestFrame) {
+        this.decodedRequestFrame = decodedRequestFrame;
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) {
+        ctx.writeAndFlush(decodedRequestFrame);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) {
+        onResponse.complete((DecodedResponseFrame<?>) msg);
+        ctx.write(msg).addListener(ChannelFutureListener.CLOSE);
+    }
+
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) {
+        ctx.flush();
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        cause.printStackTrace();
+        ctx.close();
+    }
+
+    /**
+     * A future that is completed when the response is received
+     * @return on response future
+     */
+    public CompletableFuture<DecodedResponseFrame<?>> getOnResponseFuture() {
+        return onResponse;
+    }
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/ByteBufAccessor.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/ByteBufAccessor.java
new file mode 100644
index 00000000..ea817512
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/ByteBufAccessor.java
@@ -0,0 +1,59 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.codec;
+
+import java.nio.ByteBuffer;
+
+import org.apache.kafka.common.protocol.Writable;
+
+/**
+ * Provides write access to byte buffer for serializing frames.
+ */
+public interface ByteBufAccessor extends Writable {
+
+    @Override
+    void writeByte(byte val);
+
+    @Override
+    void writeShort(short val);
+
+    @Override
+    void writeInt(int val);
+
+    @Override
+    void writeLong(long val);
+
+    @Override
+    void writeDouble(double val);
+
+    @Override
+    void writeByteArray(byte[] arr);
+
+    @Override
+    void writeUnsignedVarint(int i);
+
+    @Override
+    void writeByteBuffer(ByteBuffer byteBuffer);
+
+    @Override
+    void writeVarint(int i);
+
+    @Override
+    void writeVarlong(long i);
+
+    /**
+     * Ensure underlying buffer is sized ready to be written to
+     * @param encodedSize size we are about to write
+     */
+    void ensureWritable(int encodedSize);
+
+    /**
+     * Get writer index
+     * @return writerIndex of underlying buffer
+     */
+    int writerIndex();
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/ByteBufAccessorImpl.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/ByteBufAccessorImpl.java
new file mode 100644
index 00000000..092ec0a5
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/ByteBufAccessorImpl.java
@@ -0,0 +1,245 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.codec;
+
+import java.nio.ByteBuffer;
+
+import org.apache.kafka.common.protocol.Readable;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+
+/**
+ * An implementation of Kafka's Readable and Writable abstraction in terms of
+ * a Netty ByteBuf.
+ * This allows us to re-use Kafka's generated {@code *RequestData} and
+ * {@code *ResponseData} classes as-is.
+ * This isn't completely ideal because the Kafka APIs for decoding of Records
+ * depends on NIO ByteBuffer, so copying between ByteBuffer and ByteBuf cannot
+ * always be avoided.
+ */
+public class ByteBufAccessorImpl implements ByteBufAccessor, Readable {
+
+    private final ByteBuf buf;
+
+    /**
+     * Create accessor
+     * @param buf underlying buffer to access
+     */
+    public ByteBufAccessorImpl(ByteBuf buf) {
+        this.buf = buf;
+    }
+
+    private static IllegalArgumentException illegalVarintException(int value) {
+        throw new IllegalArgumentException("Varint is too long, the most significant bit in the 5th byte is set, " +
+                "converted value: " + Integer.toHexString(value));
+    }
+
+    private static IllegalArgumentException illegalVarlongException(long value) {
+        throw new IllegalArgumentException("Varlong is too long, most significant bit in the 10th byte is set, " +
+                "converted value: " + Long.toHexString(value));
+    }
+
+    private static IllegalArgumentException illegalReadException(int size, int remaining) {
+        throw new IllegalArgumentException("Error reading byte array of " + size + " byte(s): only " + remaining +
+                " byte(s) available");
+    }
+
+    /**
+     * Read a long stored in variable-length format using zig-zag decoding from
+     * <a href="http://code.google.com/apis/protocolbuffers/docs/encoding.html"> Google Protocol Buffers</a>.
+     *
+     * @param buffer The buffer to read from
+     * @return The long value read
+     *
+     * @throws IllegalArgumentException if variable-length value does not terminate after 10 bytes have been read
+     */
+    private static long readVarlong(ByteBuf buffer) {
+        long value = 0L;
+        int i = 0;
+        long b;
+        while (((b = buffer.readByte()) & 0x80) != 0) {
+            value |= (b & 0x7f) << i;
+            i += 7;
+            if (i > 63) {
+                throw illegalVarlongException(value);
+            }
+        }
+        value |= b << i;
+        return (value >>> 1) ^ -(value & 1);
+    }
+
+    private static int readUnsignedVarint(ByteBuf buffer) {
+        int value = 0;
+        int i = 0;
+        int b;
+        while (((b = buffer.readByte()) & 0x80) != 0) {
+            value |= (b & 0x7f) << i;
+            i += 7;
+            if (i > 28) {
+                throw illegalVarintException(value);
+            }
+        }
+        value |= b << i;
+        return value;
+    }
+
+    private static int readVarint(ByteBuf buffer) {
+        int value = readUnsignedVarint(buffer);
+        return (value >>> 1) ^ -(value & 1);
+    }
+
+    private static void writeVarlong(long value, ByteBuf buffer) {
+        long v = (value << 1) ^ (value >> 63);
+        while ((v & 0xffffffffffffff80L) != 0L) {
+            byte b = (byte) ((v & 0x7f) | 0x80);
+            buffer.writeByte(b);
+            v >>>= 7;
+        }
+        buffer.writeByte((byte) v);
+    }
+
+    private static void writeVarint(int value, ByteBuf buffer) {
+        writeUnsignedVarint((value << 1) ^ (value >> 31), buffer);
+    }
+
+    private static void writeUnsignedVarint(int value, ByteBuf buffer) {
+        while ((value & 0xffffff80) != 0L) {
+            byte b = (byte) ((value & 0x7f) | 0x80);
+            buffer.writeByte(b);
+            value >>>= 7;
+        }
+        buffer.writeByte((byte) value);
+    }
+
+    @Override
+    public byte readByte() {
+        return buf.readByte();
+    }
+
+    @Override
+    public short readShort() {
+        return buf.readShort();
+    }
+
+    @Override
+    public int readInt() {
+        return buf.readInt();
+    }
+
+    @Override
+    public long readLong() {
+        return buf.readLong();
+    }
+
+    @Override
+    public double readDouble() {
+        return buf.readDouble();
+    }
+
+    @Override
+    public byte[] readArray(int size) {
+        int remaining = buf.readableBytes();
+        if (size > remaining) {
+            throw illegalReadException(size, remaining);
+        }
+        byte[] dst = new byte[size];
+        buf.readBytes(dst, 0, size);
+        return dst;
+    }
+
+    @Override
+    public int readUnsignedVarint() {
+        return readUnsignedVarint(buf);
+    }
+
+    @Override
+    public ByteBuffer readByteBuffer(int length) {
+        // TODO use buf.nioBufferCount() and buf.nioBuffers() to avoid the copy if possible
+        ByteBuffer wrap = ByteBuffer.wrap(ByteBufUtil.getBytes(buf, buf.readerIndex(), length, false));
+        buf.readerIndex(buf.readerIndex() + length);
+        return wrap;
+
+    }
+
+    @Override
+    public int readVarint() {
+        return readVarint(buf);
+    }
+
+    @Override
+    public long readVarlong() {
+        return readVarlong(buf);
+    }
+
+    @Override
+    public int remaining() {
+        return buf.writerIndex() - buf.readerIndex();
+    }
+
+    @Override
+    public void writeByte(byte val) {
+        buf.writeByte(val);
+    }
+
+    @Override
+    public void writeShort(short val) {
+        buf.writeShort(val);
+    }
+
+    @Override
+    public void writeInt(int val) {
+        buf.writeInt(val);
+    }
+
+    @Override
+    public void writeLong(long val) {
+        buf.writeLong(val);
+    }
+
+    @Override
+    public void writeDouble(double val) {
+        buf.writeDouble(val);
+    }
+
+    @Override
+    public void writeByteArray(byte[] arr) {
+        buf.writeBytes(arr);
+    }
+
+    @Override
+    public void writeUnsignedVarint(int i) {
+        writeUnsignedVarint(i, buf);
+    }
+
+    @Override
+    public void writeByteBuffer(ByteBuffer byteBuffer) {
+        while (byteBuffer.hasRemaining()) {
+            buf.writeByte(byteBuffer.get());
+        }
+    }
+
+    @Override
+    public void writeVarint(int i) {
+        writeVarint(i, buf);
+    }
+
+    @Override
+    public void writeVarlong(long i) {
+        writeVarlong(i, buf);
+    }
+
+    @Override
+    public void ensureWritable(int encodedSize) {
+        buf.ensureWritable(encodedSize);
+    }
+
+    @Override
+    public int writerIndex() {
+        return buf.writerIndex();
+    }
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/DecodedFrame.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/DecodedFrame.java
new file mode 100644
index 00000000..9ec8fd48
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/DecodedFrame.java
@@ -0,0 +1,149 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.codec;
+
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.MessageSizeAccumulator;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A frame that has been decoded.
+ *
+ * @param <H> The header type
+ * @param <B> The body type
+ *
+ */
+public abstract class DecodedFrame<H extends ApiMessage, B extends ApiMessage>
+        implements Frame {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(DecodedFrame.class);
+
+    /**
+     * Number of bytes required for storing the frame length.
+     */
+    private static final int FRAME_SIZE_LENGTH = Integer.BYTES;
+
+    /**
+     * apiVersion of the frame
+     */
+    protected final short apiVersion;
+
+    /**
+     * correlationId of the frame
+     */
+    protected final int correlationId;
+
+    /**
+     * header of the frame
+     */
+    protected final H header;
+
+    /**
+     * body of the frame
+     */
+    protected final B body;
+    private int headerAndBodyEncodedLength;
+    private ObjectSerializationCache serializationCache;
+
+    DecodedFrame(short apiVersion, int correlationId, H header, B body) {
+        this.apiVersion = apiVersion;
+        this.correlationId = correlationId;
+        this.header = header;
+        this.body = body;
+        this.headerAndBodyEncodedLength = -1;
+    }
+
+    @Override
+    public int correlationId() {
+        return correlationId;
+    }
+
+    /**
+     * header version of frame
+     * @return header version
+     */
+    protected abstract short headerVersion();
+
+    /**
+     * Header of the frame
+     * @return header
+     */
+    public H header() {
+        return header;
+    }
+
+    /**
+     * Body of the frame
+     * @return body
+     */
+    public B body() {
+        return body;
+    }
+
+    /**
+     * Get apiKey of body
+     * @return apiKey
+     */
+    public ApiKeys apiKey() {
+        return ApiKeys.forId(body.apiKey());
+    }
+
+    /**
+     * Get apiVersion of frame
+     * @return apiKey
+     */
+    public short apiVersion() {
+        return apiVersion;
+    }
+
+    @Override
+    public final int estimateEncodedSize() {
+        if (headerAndBodyEncodedLength != -1) {
+            assert serializationCache != null;
+            return FRAME_SIZE_LENGTH + headerAndBodyEncodedLength;
+        }
+        var headerVersion = headerVersion();
+        MessageSizeAccumulator sizer = new MessageSizeAccumulator();
+        ObjectSerializationCache cache = new ObjectSerializationCache();
+        header().addSize(sizer, cache, headerVersion);
+        body().addSize(sizer, cache, apiVersion());
+        headerAndBodyEncodedLength = sizer.totalSize();
+        serializationCache = cache;
+        return FRAME_SIZE_LENGTH + headerAndBodyEncodedLength;
+    }
+
+    @Override
+    public final void encode(ByteBufAccessor out) {
+        if (headerAndBodyEncodedLength < 0) {
+            LOGGER.warn("Encoding estimation should happen before encoding, if possible");
+        }
+        final int encodedSize = estimateEncodedSize();
+        if (LOGGER.isTraceEnabled()) {
+            LOGGER.trace("Writing {} with 4 byte length ({}) plus bytes of header {}, and body {} to {}",
+                    getClass().getSimpleName(), encodedSize, header, body, out);
+        }
+        out.ensureWritable(encodedSize);
+        final int initialIndex = out.writerIndex();
+        out.writeInt(headerAndBodyEncodedLength);
+        final ObjectSerializationCache cache = serializationCache;
+        header.write(out, cache, headerVersion());
+        body.write(out, cache, apiVersion());
+        assert (out.writerIndex() - initialIndex) == encodedSize;
+    }
+
+    @Override
+    public String toString() {
+        return getClass().getSimpleName() + "(" +
+                apiKey() + "(" + apiVersion + ")v" + apiVersion +
+                ", header=" + header +
+                ", body=" + body +
+                ')';
+    }
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/DecodedRequestFrame.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/DecodedRequestFrame.java
new file mode 100644
index 00000000..96c10369
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/DecodedRequestFrame.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.codec;
+
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.protocol.ApiMessage;
+
+/**
+ * A decoded request frame.
+ * @param <B> type of api message in decoded frame
+ */
+public class DecodedRequestFrame<B extends ApiMessage>
+        extends DecodedFrame<RequestHeaderData, B>
+        implements Frame {
+
+    /**
+     * Create a decoded request frame
+     * @param apiVersion apiVersion
+     * @param correlationId correlationId
+     * @param header header
+     * @param body body
+     */
+    public DecodedRequestFrame(short apiVersion,
+                               int correlationId,
+                               RequestHeaderData header,
+                               B body) {
+        super(apiVersion, correlationId, header, body);
+    }
+
+    @Override
+    public short headerVersion() {
+        return apiKey().messageType.requestHeaderVersion(apiVersion);
+    }
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/DecodedResponseFrame.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/DecodedResponseFrame.java
new file mode 100644
index 00000000..6993a486
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/DecodedResponseFrame.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.codec;
+
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.protocol.ApiMessage;
+
+/**
+ * A decoded response frame.
+ * @param <B> the decoded ApiMessage type
+ */
+public class DecodedResponseFrame<B extends ApiMessage>
+        extends DecodedFrame<ResponseHeaderData, B>
+        implements Frame {
+
+    /**
+     * Create a decoded response frame
+     * @param apiVersion apiVersion
+     * @param correlationId correlationId
+     * @param header header
+     * @param body body
+     */
+    public DecodedResponseFrame(short apiVersion, int correlationId, ResponseHeaderData header, B body) {
+        super(apiVersion, correlationId, header, body);
+    }
+
+    public short headerVersion() {
+        return apiKey().messageType.responseHeaderVersion(apiVersion);
+    }
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/Frame.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/Frame.java
new file mode 100644
index 00000000..0035d87b
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/Frame.java
@@ -0,0 +1,32 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.codec;
+
+/**
+ * A frame in the Kafka protocol, which may or may not be fully decoded.
+ */
+public interface Frame {
+
+    /**
+     * Estimate the expected encoded size of this {@code Frame}.<br>
+     * In particular, written data by {@link #encode(ByteBufAccessor)} should be the same as reported by this method.
+     * @return the expected encoded size in bytes
+     */
+    int estimateEncodedSize();
+
+    /**
+     * Write the frame, including the size prefix, to the given buffer
+     * @param out The output buffer
+     */
+    void encode(ByteBufAccessor out);
+
+    /**
+     * The correlation id.
+     * @return The correlation id.
+     */
+    int correlationId();
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaMessageDecoder.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaMessageDecoder.java
new file mode 100644
index 00000000..389ac130
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaMessageDecoder.java
@@ -0,0 +1,75 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.codec;
+
+import java.util.List;
+
+import org.slf4j.Logger;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+
+/**
+ * Abstraction for request and response decoders.
+ */
+public abstract class KafkaMessageDecoder extends ByteToMessageDecoder {
+
+    /**
+     * Get the logger to use
+     * @return logger
+     */
+    protected abstract Logger log();
+
+    /**
+     * Create a KafkaMessageDecoder
+     */
+    public KafkaMessageDecoder() {
+    }
+
+    @Override
+    public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
+        while (in.readableBytes() > 4) {
+            try {
+                int sof = in.readerIndex();
+                int frameSize = in.readInt();
+                int readable = in.readableBytes();
+                if (log().isTraceEnabled()) { // avoid boxing
+                    log().trace("{}: Frame of {} bytes ({} readable)", ctx, frameSize, readable);
+                }
+                // TODO handle too-large frames
+                if (readable >= frameSize) { // We can read the whole frame
+                    var idx = in.readerIndex();
+                    out.add(decodeHeaderAndBody(ctx,
+                            in.readSlice(frameSize), // Prevent decodeHeaderAndBody() from reading beyond the frame
+                            frameSize));
+                    log().trace("{}: readable: {}, having read {}", ctx, in.readableBytes(), in.readerIndex() - idx);
+                    if (in.readerIndex() - idx != frameSize) {
+                        throw new RuntimeException("decodeHeaderAndBody did not read all of the buffer " + in);
+                    }
+                }
+                else {
+                    in.readerIndex(sof);
+                    break;
+                }
+            }
+            catch (Exception e) {
+                log().error("{}: Error in decoder", ctx, e);
+                throw e;
+            }
+        }
+    }
+
+    /**
+     * Decode the header and body
+     * @param ctx handler context
+     * @param in buffer slice to decode from
+     * @param length size of the slice
+     * @return a Frame
+     */
+    protected abstract Frame decodeHeaderAndBody(ChannelHandlerContext ctx, ByteBuf in, int length);
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaMessageEncoder.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaMessageEncoder.java
new file mode 100644
index 00000000..8c2ea807
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaMessageEncoder.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.codec;
+
+import org.slf4j.Logger;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+
+/**
+ * Abstraction for request and response encoders.
+ * @param <F> type of frame to encode
+ */
+public abstract class KafkaMessageEncoder<F extends Frame> extends MessageToByteEncoder<F> {
+
+    /**
+     * Create KafkaMessageEncoder
+     */
+    public KafkaMessageEncoder() {
+    }
+
+    /**
+     * Logger
+     * @return logger
+     */
+    protected abstract Logger log();
+
+    /**
+     * This has been overridden in order to use {@link Frame#estimateEncodedSize()} to pre-size correctly the holding buffer
+     * and save expensive enlarging to happen under the hood, during the encoding process.
+     */
+    @Override
+    protected ByteBuf allocateBuffer(final ChannelHandlerContext ctx, final F msg, final boolean preferDirect) {
+        final int bytes = msg.estimateEncodedSize();
+        if (preferDirect) {
+            return ctx.alloc().ioBuffer(bytes);
+        }
+        return ctx.alloc().heapBuffer(bytes);
+    }
+
+    @Override
+    protected void encode(ChannelHandlerContext ctx, F frame, ByteBuf out) throws Exception {
+        log().trace("{}: Encoding {} to buffer {}", ctx, frame, out);
+        frame.encode(new ByteBufAccessorImpl(out));
+    }
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaRequestDecoder.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaRequestDecoder.java
new file mode 100644
index 00000000..95c4a6cf
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaRequestDecoder.java
@@ -0,0 +1,87 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.codec;
+
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.Readable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+
+/**
+ * Decoder for Kafka Requests
+ */
+public class KafkaRequestDecoder extends KafkaMessageDecoder {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaRequestDecoder.class);
+
+    /**
+     * Create KafkaRequestDecoder
+     */
+    public KafkaRequestDecoder() {
+        super();
+    }
+
+    @Override
+    protected Logger log() {
+        return LOGGER;
+    }
+
+    @Override
+    protected Frame decodeHeaderAndBody(ChannelHandlerContext ctx, ByteBuf in, final int length) {
+        // Read the api key and version to determine the header api version
+        final int sof = in.readerIndex();
+        var apiId = in.readShort();
+        ApiKeys apiKey = ApiKeys.forId(apiId);
+        if (log().isTraceEnabled()) { // avoid boxing
+            log().trace("{}: apiKey: {} {}", ctx, apiId, apiKey);
+        }
+        short apiVersion = in.readShort();
+        if (log().isTraceEnabled()) { // avoid boxing
+            log().trace("{}: apiVersion: {}", ctx, apiVersion);
+        }
+        int correlationId = in.readInt();
+        LOGGER.debug("{}: {} downstream correlation id: {}", ctx, apiKey, correlationId);
+
+        RequestHeaderData header;
+        final ByteBufAccessorImpl accessor;
+        var decodeRequest = true;
+        LOGGER.debug("Decode {}/v{} request? {}", apiKey, apiVersion, decodeRequest);
+        boolean decodeResponse = true;
+        LOGGER.debug("Decode {}/v{} response? {}", apiKey, apiVersion, decodeResponse);
+        short headerVersion = apiKey.requestHeaderVersion(apiVersion);
+        if (log().isTraceEnabled()) { // avoid boxing
+            log().trace("{}: headerVersion {}", ctx, headerVersion);
+        }
+        in.readerIndex(sof);
+        accessor = new ByteBufAccessorImpl(in);
+        header = readHeader(headerVersion, accessor);
+        if (log().isTraceEnabled()) {
+            log().trace("{}: header: {}", ctx, header);
+        }
+        final Frame frame;
+        ApiMessage body = BodyDecoder.decodeRequest(apiKey, apiVersion, accessor);
+        if (log().isTraceEnabled()) {
+            log().trace("{}: body {}", ctx, body);
+        }
+
+        frame = new DecodedRequestFrame<>(apiVersion, correlationId, header, body);
+        if (log().isTraceEnabled()) {
+            log().trace("{}: frame {}", ctx, frame);
+        }
+
+        return frame;
+    }
+
+    private RequestHeaderData readHeader(short headerVersion, Readable accessor) {
+        return new RequestHeaderData(accessor, headerVersion);
+    }
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaRequestEncoder.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaRequestEncoder.java
new file mode 100644
index 00000000..50204d18
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaRequestEncoder.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.codec;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+
+import io.kroxylicious.test.client.CorrelationManager;
+
+/**
+ * Kafka Request Encoder
+ */
+public class KafkaRequestEncoder extends KafkaMessageEncoder<DecodedRequestFrame<?>> {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaRequestEncoder.class);
+
+    private final CorrelationManager correlationManager;
+
+    /**
+     * Create KafkaRequestEncoder
+     * @param correlationManager manager for tracking the apiKey and apiVersion per correlationId
+     */
+    public KafkaRequestEncoder(CorrelationManager correlationManager) {
+        this.correlationManager = correlationManager;
+    }
+
+    @Override
+    protected Logger log() {
+        return LOGGER;
+    }
+
+    @Override
+    protected void encode(ChannelHandlerContext ctx, DecodedRequestFrame frame, ByteBuf out) throws Exception {
+        super.encode(ctx, frame, out);
+        // not sure if this testing client needs to know that acks=0 produce requests don't get responses
+        correlationManager.putBrokerRequest(frame.apiKey().id, frame.apiVersion(), frame.correlationId());
+    }
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaResponseDecoder.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaResponseDecoder.java
new file mode 100644
index 00000000..aa45d524
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaResponseDecoder.java
@@ -0,0 +1,75 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.codec;
+
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.Readable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+
+import io.kroxylicious.test.client.CorrelationManager;
+
+/**
+ * KafkaResponseDecoder
+ */
+public class KafkaResponseDecoder extends KafkaMessageDecoder {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaResponseDecoder.class);
+
+    private final CorrelationManager correlationManager;
+
+    /**
+     * Creates a response decoder
+     * @param correlationManager the manager used to retrieve apiKey and apiVersion for the response
+     */
+    public KafkaResponseDecoder(CorrelationManager correlationManager) {
+        super();
+        this.correlationManager = correlationManager;
+    }
+
+    @Override
+    protected Logger log() {
+        return LOGGER;
+    }
+
+    @Override
+    protected Frame decodeHeaderAndBody(ChannelHandlerContext ctx, ByteBuf in, int length) {
+        var ri = in.readerIndex();
+        var correlationId = in.readInt();
+        in.readerIndex(ri);
+
+        CorrelationManager.Correlation correlation = this.correlationManager.getBrokerCorrelation(correlationId);
+        if (correlation == null) {
+            throw new AssertionError("Missing correlation id " + correlationId);
+        }
+        else if (LOGGER.isDebugEnabled()) {
+            LOGGER.debug("{}: Recovered correlation {} for upstream correlation id {}", ctx, correlation, correlationId);
+        }
+
+        final Frame frame;
+        ApiKeys apiKey = ApiKeys.forId(correlation.apiKey());
+        short apiVersion = correlation.apiVersion();
+        var accessor = new ByteBufAccessorImpl(in);
+        short headerVersion = apiKey.responseHeaderVersion(apiVersion);
+        log().trace("{}: Header version: {}", ctx, headerVersion);
+        ResponseHeaderData header = readHeader(headerVersion, accessor);
+        log().trace("{}: Header: {}", ctx, header);
+        ApiMessage body = BodyDecoder.decodeResponse(apiKey, apiVersion, accessor);
+        log().trace("{}: Body: {}", ctx, body);
+        frame = new DecodedResponseFrame<>(apiVersion, correlationId, header, body);
+        return frame;
+    }
+
+    private ResponseHeaderData readHeader(short headerVersion, Readable accessor) {
+        return new ResponseHeaderData(accessor, headerVersion);
+    }
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaResponseEncoder.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaResponseEncoder.java
new file mode 100644
index 00000000..90a47fec
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/codec/KafkaResponseEncoder.java
@@ -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 io.kroxylicious.test.codec;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * KafkaResponseEncoder
+ */
+public class KafkaResponseEncoder extends KafkaMessageEncoder<Frame> {
+
+    /**
+     * Create KafkaResponseEncoder
+     */
+    public KafkaResponseEncoder() {
+    }
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaResponseEncoder.class);
+
+    @Override
+    protected Logger log() {
+        return LOGGER;
+    }
+
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/server/MockHandler.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/server/MockHandler.java
new file mode 100644
index 00000000..8a432ef6
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/server/MockHandler.java
@@ -0,0 +1,98 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.server;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.protocol.ApiMessage;
+
+import io.netty.channel.ChannelHandler.Sharable;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+import io.kroxylicious.test.codec.DecodedRequestFrame;
+import io.kroxylicious.test.codec.DecodedResponseFrame;
+
+/**
+ * MockHandler is responsible for:
+ * <ol>
+ *     <li>Serves a single response for any requests it receives. The response can be modified
+ *  * using setResponse.</li>
+ *     <li>Records requests it receives so they can be retrieved and verified</li>
+ *     <li>Can be cleared, making it forget received requests</li>
+ * </ol>
+ */
+@Sharable
+public class MockHandler extends ChannelInboundHandlerAdapter {
+
+    private ApiMessage message;
+
+    private final List<DecodedRequestFrame<?>> requests = new ArrayList<>();
+
+    /**
+     * Create mockhandler with initial message to serve
+     * @param message message to respond with, nullable
+     */
+    public MockHandler(ApiMessage message) {
+        this.message = message;
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) {
+        DecodedRequestFrame<?> msg1 = (DecodedRequestFrame<?>) msg;
+        respond(ctx, msg1);
+    }
+
+    private void respond(ChannelHandlerContext ctx, DecodedRequestFrame<?> frame) {
+        requests.add(frame);
+        if (message == null) {
+            // we allow a null message to enable tests to start the mock server
+            // before they set the expectation.
+            throw new RuntimeException("response message not set");
+        }
+        DecodedResponseFrame<?> responseFrame = new DecodedResponseFrame<>(frame.apiVersion(),
+                frame.correlationId(), new ResponseHeaderData().setCorrelationId(frame.correlationId()), message);
+        ctx.write(responseFrame);
+    }
+
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) {
+        ctx.flush();
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        // Close the connection when an exception is raised.
+        cause.printStackTrace();
+        ctx.close();
+    }
+
+    /**
+     * Set the response
+     * @param response response
+     */
+    public void setResponse(ApiMessage response) {
+        message = response;
+    }
+
+    /**
+     * Get requests
+     * @return get received requests
+     */
+    public List<DecodedRequestFrame<?>> getRequests() {
+        return Collections.unmodifiableList(requests);
+    }
+
+    /**
+     * Clear recorded requests
+     */
+    public void clear() {
+        requests.clear();
+    }
+}
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/server/MockServer.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/server/MockServer.java
new file mode 100644
index 00000000..4886010c
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/server/MockServer.java
@@ -0,0 +1,151 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.test.server;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+
+import io.kroxylicious.test.Request;
+import io.kroxylicious.test.Response;
+import io.kroxylicious.test.codec.DecodedRequestFrame;
+import io.kroxylicious.test.codec.KafkaRequestDecoder;
+import io.kroxylicious.test.codec.KafkaResponseEncoder;
+
+/**
+ * MockServer. Provides a mock kafka broker that can respond with a single
+ * fixed ApiMessage at a time. Intended for per-RPC testing of kroxylicious where
+ * we fire one RPC through the proxy, respond with some known message and then
+ * check the output from the proxy.
+ */
+public final class MockServer implements AutoCloseable {
+    private Channel channel;
+    private EventLoopGroup bossGroup;
+    private EventLoopGroup workerGroup;
+    private final int port;
+    private MockHandler serverHandler;
+
+    private MockServer(Response response, int port) {
+        this.port = start(port, response);
+    }
+
+    /**
+     * Set the response to be served by the MockServer
+     * @param response the response (nullable)
+     */
+    public void setResponse(Response response) {
+        serverHandler.setResponse(response == null ? null : response.message());
+    }
+
+    /**
+     * Get the requests received by the mock.
+     * @return list of requests converted into a normalised JsonNode
+     */
+    public List<Request> getReceivedRequests() {
+        return serverHandler.getRequests().stream().map(MockServer::toRequest).toList();
+    }
+
+    private static Request toRequest(DecodedRequestFrame<?> decodedRequestFrame) {
+        return new Request(decodedRequestFrame.apiKey(), decodedRequestFrame.apiVersion(), decodedRequestFrame.header().clientId(), decodedRequestFrame.body());
+    }
+
+    /**
+     * Start mock server on a random port. Note a response must be set on it before
+     * it receives any requests or mocking will fail.
+     * @return the created server
+     */
+    public static MockServer startOnRandomPort() {
+        return new MockServer(null, 0);
+    }
+
+    /**
+     * Start mock server on a random port and serve this response
+     * @param response response to serve
+     * @return the created server
+     */
+    public static MockServer startOnRandomPort(Response response) {
+        return new MockServer(response, 0);
+    }
+
+    /**
+     * Start the server
+     * @param port port to bind to (0 to bind to an ephemeral)
+     * @param response response to serve (nullable)
+     * @return the port bound to
+     */
+    public int start(int port, Response response) {
+        // Configure the server.
+        bossGroup = new NioEventLoopGroup(1);
+        workerGroup = new NioEventLoopGroup();
+        serverHandler = new MockHandler(response == null ? null : response.message());
+        ServerBootstrap b = new ServerBootstrap();
+        b.group(bossGroup, workerGroup)
+                .channel(NioServerSocketChannel.class)
+                .option(ChannelOption.SO_BACKLOG, 100)
+                .childHandler(new ChannelInitializer<SocketChannel>() {
+                    @Override
+                    public void initChannel(SocketChannel ch) {
+                        ChannelPipeline p = ch.pipeline();
+                        p.addLast(new KafkaRequestDecoder());
+                        p.addLast(new KafkaResponseEncoder());
+                        p.addLast(serverHandler);
+                    }
+                });
+
+        // Start the server.
+        ChannelFuture f;
+        try {
+            f = b.bind(port).sync();
+        }
+        catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+
+        // Wait until the server socket is closed.
+        channel = f.channel();
+        InetSocketAddress localAddress = (InetSocketAddress) channel.localAddress();
+        return localAddress.getPort();
+    }
+
+    @Override
+    public void close() {
+        ChannelFuture channelFuture = channel.close();
+        try {
+            channelFuture.sync();
+            bossGroup.shutdownGracefully();
+            workerGroup.shutdownGracefully();
+        }
+        catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    /**
+     * port mock server is listening on
+     * @return port
+     */
+    public int port() {
+        return port;
+    }
+
+    /**
+     * Clear the response and tell the serverHandler to clear its collection of received requests.
+     */
+    public void clear() {
+        setResponse(null);
+        serverHandler.clear();
+    }
+}
diff --git a/kroxylicious-test-tools/src/main/templates/BodyDecoder.ftl b/kroxylicious-test-tools/src/main/templates/BodyDecoder.ftl
new file mode 100644
index 00000000..2c577fc5
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/templates/BodyDecoder.ftl
@@ -0,0 +1,87 @@
+<#--
+
+    Copyright Kroxylicious Authors.
+
+    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.
+ */
+package ${outputPackage};
+
+<#list messageSpecs as messageSpec>
+import org.apache.kafka.common.message.${messageSpec.name}Data;
+</#list>
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.Readable;
+
+/**
+* Decodes Kafka Readable into an ApiMessage
+* <p>Note: this class is automatically generated from a template</p>
+*/
+public class BodyDecoder {
+
+    /**
+    * Creates a BodyDecoder
+    */
+    private BodyDecoder() {
+    }
+
+    /**
+    * Decodes Kafka request Readable into an ApiMessage
+    * @param apiKey the api key of the message
+    * @param apiVersion the api version of the message
+    * @param accessor the accessor for the message bytes
+    * @return the ApiMessage
+    * @throws IllegalArgumentException if an unhandled ApiKey is encountered
+    */
+    static ApiMessage decodeRequest(ApiKeys apiKey, short apiVersion, Readable accessor) {
+        switch (apiKey) {
+<#list messageSpecs as messageSpec>
+    <#if messageSpec.type?lower_case == 'request'>
+            case ${retrieveApiKey(messageSpec)}:
+                return new ${messageSpec.name}Data(accessor, apiVersion);
+    </#if>
+</#list>
+            default:
+                throw new IllegalArgumentException("Unsupported RPC " + apiKey);
+        }
+    }
+
+    /**
+    * Decodes Kafka response Readable into an ApiMessage
+    * @param apiKey the api key of the message
+    * @param apiVersion the api version of the message
+    * @param accessor the accessor for the message bytes
+    * @return the ApiMessage
+    * @throws IllegalArgumentException if an unhandled ApiKey is encountered
+    */
+    static ApiMessage decodeResponse(ApiKeys apiKey, short apiVersion, Readable accessor) {
+        switch (apiKey) {
+<#list messageSpecs as messageSpec>
+    <#if messageSpec.type?lower_case == 'response'>
+            case ${retrieveApiKey(messageSpec)}:
+                return new ${messageSpec.name}Data(accessor, apiVersion);
+    </#if>
+</#list>
+            default:
+                throw new IllegalArgumentException("Unsupported RPC " + apiKey);
+        }
+    }
+
+}
diff --git a/kroxylicious-test-tools/src/main/templates/DataClasses.ftl b/kroxylicious-test-tools/src/main/templates/DataClasses.ftl
new file mode 100644
index 00000000..7e35e9f2
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/templates/DataClasses.ftl
@@ -0,0 +1,68 @@
+<#--
+
+    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 java.util.Collections;
+import java.util.Map;
+import java.util.HashMap;
+
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+
+<#list messageSpecs as messageSpec>
+import org.apache.kafka.common.message.${messageSpec.name}Data;
+</#list>
+
+
+/**
+* Enumerates all DataClasses
+*/
+public class DataClasses {
+
+    /**
+    * Create an empty DataClasses
+    */
+    private DataClasses() {
+
+    }
+
+    private static final Map<ApiKeys, Class<? extends ApiMessage>> requestClasses;
+    private static final Map<ApiKeys, Class<? extends ApiMessage>> responseClasses;
+
+    static {
+        requestClasses = new HashMap<ApiKeys, Class<? extends ApiMessage>>();
+        responseClasses = new HashMap<ApiKeys, Class<? extends ApiMessage>>();
+
+<#list messageSpecs as messageSpec>
+    <#if messageSpec.type?lower_case == 'request'>
+        requestClasses.put(ApiKeys.${retrieveApiKey(messageSpec)}, ${messageSpec.name}Data.class);
+    </#if>
+    <#if messageSpec.type?lower_case == 'response'>
+        responseClasses.put(ApiKeys.${retrieveApiKey(messageSpec)}, ${messageSpec.name}Data.class);
+    </#if>
+</#list>
+    }
+
+    /**
+    * Get the ApiMessage class per ApiKey for request messages
+    * @return ApiKeys to ApiMessage class mappings
+    */
+    public static Map<ApiKeys, Class<? extends ApiMessage>> getRequestClasses() {
+        return Collections.unmodifiableMap(requestClasses);
+    }
+
+    /**
+    * Get the ApiMessage class per ApiKey for response messages
+    * @return ApiKeys to ApiMessage class mappings
+    */
+    public static Map<ApiKeys, Class<? extends ApiMessage>> getResponseClasses() {
+        return Collections.unmodifiableMap(responseClasses);
+    }
+}
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
new file mode 100644
index 00000000..694bd69a
--- /dev/null
+++ b/kroxylicious-test-tools/src/test/java/io/kroxylicious/mock/MockServerTest.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.mock;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import org.apache.kafka.common.message.ApiMessageType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import io.kroxylicious.test.ApiMessageSampleGenerator;
+import io.kroxylicious.test.ApiMessageSampleGenerator.ApiAndVersion;
+import io.kroxylicious.test.DataClasses;
+import io.kroxylicious.test.Request;
+import io.kroxylicious.test.Response;
+import io.kroxylicious.test.client.KafkaClient;
+import io.kroxylicious.test.server.MockServer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class MockServerTest {
+    private static final Map<ApiAndVersion, ApiMessage> responseSamples = ApiMessageSampleGenerator.createResponseSamples();
+    private static final Map<ApiAndVersion, ApiMessage> requestSamples = ApiMessageSampleGenerator.createRequestSamples();
+
+    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);
+            return supported.mapToObj(version -> new ApiAndVersion(apiKeys, (short) version));
+        });
+    }
+
+    @ParameterizedTest
+    @MethodSource("allSupportedApiVersions")
+    public void testClientCanSendAndReceiveRPCToMock(ApiAndVersion apiKey) throws Exception {
+        Response mockResponse = getResponse(apiKey);
+        try (MockServer mockServer = MockServer.startOnRandomPort(mockResponse); KafkaClient kafkaClient = new KafkaClient()) {
+            CompletableFuture<Response> future = kafkaClient.get("127.0.0.1", mockServer.port(), getRequest(apiKey));
+            Response clientResponse = future.get(10, TimeUnit.SECONDS);
+            assertEquals(mockResponse, clientResponse);
+        }
+    }
+
+    private Response getResponse(ApiAndVersion apiAndVersion) {
+        return new Response(apiAndVersion.keys(), apiAndVersion.apiVersion(), responseSamples.get(apiAndVersion));
+    }
+
+    private Request getRequest(ApiAndVersion apiAndVersion) {
+        short apiVersion = apiAndVersion.apiVersion();
+        ApiMessage message = requestSamples.get(apiAndVersion);
+        return new Request(apiAndVersion.keys(), apiVersion, "clientId", message);
+    }
+
+}
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
new file mode 100644
index 00000000..83754d22
--- /dev/null
+++ b/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/ApiMessageSampleGeneratorTest.java
@@ -0,0 +1,98 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import org.apache.kafka.common.message.ApiMessageType;
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.record.BaseRecords;
+import org.apache.kafka.common.record.DefaultRecord;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.MutableRecordBatch;
+import org.apache.kafka.common.record.Record;
+import org.apache.kafka.common.utils.AbstractIterator;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.kafka.common.protocol.ApiKeys.PRODUCE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ApiMessageSampleGeneratorTest {
+
+    @Test
+    void testResponseSampleGenerationDeterministic() {
+        Map<ApiMessageSampleGenerator.ApiAndVersion, ApiMessage> a = ApiMessageSampleGenerator.createResponseSamples();
+        Map<ApiMessageSampleGenerator.ApiAndVersion, ApiMessage> b = ApiMessageSampleGenerator.createResponseSamples();
+
+        allSupportedApiVersions().forEach(apiAndVersion -> {
+            assertEquals(a.get(apiAndVersion), b.get(apiAndVersion));
+        });
+    }
+
+    @Test
+    void testRequestSampleGenerationDeterministic() {
+        Map<ApiMessageSampleGenerator.ApiAndVersion, ApiMessage> a = ApiMessageSampleGenerator.createRequestSamples();
+        Map<ApiMessageSampleGenerator.ApiAndVersion, ApiMessage> b = ApiMessageSampleGenerator.createRequestSamples();
+
+        allSupportedApiVersions().forEach(apiAndVersion -> {
+            assertEquals(a.get(apiAndVersion), b.get(apiAndVersion));
+        });
+    }
+
+    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);
+            return supported.mapToObj(version -> new ApiMessageSampleGenerator.ApiAndVersion(apiKeys, (short) version));
+        });
+    }
+
+    /**
+     * Go deep on one type that involves nested collections and MemoryRecords which have
+     * been the most fraught area to generate
+     */
+    @Test
+    void testProduceCollectionData() {
+        Map<ApiMessageSampleGenerator.ApiAndVersion, ApiMessage> messages = ApiMessageSampleGenerator.createRequestSamples();
+        ProduceRequestData message = (ProduceRequestData) messages.get(new ApiMessageSampleGenerator.ApiAndVersion(PRODUCE, PRODUCE.latestVersion()));
+        assertEquals(1, message.topicData().size());
+        ProduceRequestData.TopicProduceDataCollection topicProduceData = message.topicData();
+        Optional<ProduceRequestData.TopicProduceData> first = topicProduceData.stream().findFirst();
+        assertTrue(first.isPresent());
+        ProduceRequestData.TopicProduceData data = first.get();
+        List<ProduceRequestData.PartitionProduceData> partitionProduceData = data.partitionData();
+        assertNotNull(partitionProduceData);
+        assertEquals(1, partitionProduceData.size());
+        ProduceRequestData.PartitionProduceData element = partitionProduceData.get(0);
+        BaseRecords records = element.records();
+        assertNotNull(records);
+        AbstractIterator<MutableRecordBatch> iterator = ((MemoryRecords) records).batchIterator();
+        MutableRecordBatch batch = iterator.next();
+        assertFalse(iterator.hasNext());
+        Iterator<Record> batchIterator = batch.iterator();
+        DefaultRecord record = (DefaultRecord) batchIterator.next();
+        assertNotNull(record);
+        assertFalse(batchIterator.hasNext());
+        byte[] keyBytes = new byte[record.keySize()];
+        record.key().get(keyBytes);
+        byte[] valueBytes = new byte[record.valueSize()];
+        record.value().get(valueBytes);
+        assertTrue(new String(keyBytes, StandardCharsets.UTF_8).startsWith("key-"));
+        assertTrue(new String(valueBytes, StandardCharsets.UTF_8).startsWith("value-"));
+    }
+
+}