Repository: kroxylicious/kroxylicious
Pull request #197 (2023-03-15 20:47:44)

Set language level to 17, 19 for integration tests
### Type of change

_Select the type of your PR_

- Enhancement / new feature

### Description

Sets the language level to 17 for `main` and `test` in all modules except `integrationtests`. `integrationtests` relies on an 18+ feature to configure in a custom `InetAddressResolver` so test clients can resolve `foo.multitenant.kafka:9192` and `bar.multitenant.kafka:9192` to a localhost ip.

### Additional Context

The split language levels across test and main sources has been a pain for development, several of us use IDEA and it doesn't work smoothly having different language levels for `main` and `test`. It hungs together in an experimental maven import mode but recently we [found an issue](https://github.com/kroxylicious/kroxylicious/pull/171#issuecomment-1470346027) where test resources were not on the classpath when running some new unit tests. Moving to a more conventional setup with the same language level for test/main per maven module should make it smoother.

We want to target a LTS java version as a minimum runtime version.


Patch:
diff --git a/.idea/copyright/profiles_settings.xml b/.idea/copyright/profiles_settings.xml
new file mode 100644
index 00000000..e5edfa8d
--- /dev/null
+++ b/.idea/copyright/profiles_settings.xml
@@ -0,0 +1,7 @@
+<component name="CopyrightManager">
+  <settings>
+    <module2copyright>
+      <element module="All" copyright="Kroxylicious" />
+    </module2copyright>
+  </settings>
+</component>
\ No newline at end of file
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterContributor.java b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterContributor.java
index 2794e6ae..be5b1e45 100644
--- a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterContributor.java
+++ b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterContributor.java
@@ -7,5 +7,9 @@ package io.kroxylicious.proxy.filter;
 
 import io.kroxylicious.proxy.service.Contributor;
 
+/**
+ * FilterContributor is a pluggable source of Kroxylicious filter implementations.
+ * @see Contributor
+ */
 public interface FilterContributor extends Contributor<KrpcFilter> {
 }
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
index 8c004d81..24f87ef9 100644
--- a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
+++ b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
@@ -15,6 +15,7 @@ import org.apache.kafka.common.utils.ByteBufferOutputStream;
  */
 public interface KrpcFilterContext {
     /**
+     * A description of this channel.
      * @return A description of this channel (typically used for logging).
      */
     String channelDescriptor();
@@ -28,6 +29,7 @@ public interface KrpcFilterContext {
     ByteBufferOutputStream createByteBufferOutputStream(int initialCapacity);
 
     /**
+     * The SNI hostname provided by the client, if any.
      * @return the SNI hostname provided by the client.  Will be null if the client is
      * using a non-TLS connection or the TLS client hello didn't provide one.
      */
@@ -49,6 +51,7 @@ public interface KrpcFilterContext {
      * @param apiVersion The version of the request to use
      * @param request The request to send.
      * @param <T> The type of the response
+     * @return CompletionStage providing the response.
      */
     <T extends ApiMessage> CompletionStage<T> sendRequest(short apiVersion, ApiMessage request);
 
diff --git a/api/kroxylicious-filter-api/src/main/templates/Kproxy/Filter.ftl b/api/kroxylicious-filter-api/src/main/templates/Kproxy/Filter.ftl
index 0f70032d..c862d022 100644
--- a/api/kroxylicious-filter-api/src/main/templates/Kproxy/Filter.ftl
+++ b/api/kroxylicious-filter-api/src/main/templates/Kproxy/Filter.ftl
@@ -47,6 +47,7 @@ public interface ${filterClass} extends KrpcFilter {
      * The implementation may modify the given {@code data} in-place and return it,
      * or instantiate a new one.
      *
+     * @param header <#if messageSpec.type?lower_case == 'response'>response<#else>request</#if> header.
      * @param ${msgType} The KRPC message to handle.
      * @param context The context.
      */
diff --git a/api/kroxylicious-filter-api/src/main/templates/Kproxy/KrpcFilter.ftl b/api/kroxylicious-filter-api/src/main/templates/Kproxy/KrpcFilter.ftl
index 1b16d81a..c405194c 100644
--- a/api/kroxylicious-filter-api/src/main/templates/Kproxy/KrpcFilter.ftl
+++ b/api/kroxylicious-filter-api/src/main/templates/Kproxy/KrpcFilter.ftl
@@ -124,7 +124,7 @@ public /* sealed */ interface KrpcFilter /* TODO permits ... */ {
      * </ol>
      * @param apiKey The API key
      * @param apiVersion The API version
-     * @return
+     * @return true if request should be deserialized
      */
     default boolean shouldDeserializeRequest(ApiKeys apiKey, short apiVersion) {
         switch (apiKey) {
@@ -153,7 +153,7 @@ public /* sealed */ interface KrpcFilter /* TODO permits ... */ {
      * </ol>
      * @param apiKey The API key
      * @param apiVersion The API version
-     * @return
+     * @return true if response should be deserialized
      */
     default boolean shouldDeserializeResponse(ApiKeys apiKey, short apiVersion) {
         switch (apiKey) {
diff --git a/kroxylicious-multitenant/pom.xml b/kroxylicious-multitenant/pom.xml
index 837a08da..7796be7a 100644
--- a/kroxylicious-multitenant/pom.xml
+++ b/kroxylicious-multitenant/pom.xml
@@ -35,5 +35,46 @@
             <artifactId>kroxylicious-filter-api</artifactId>
             <version>${kroxyliciousApi.version}</version>
         </dependency>
+        <!-- Test dependencies -->
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-slf4j-impl</artifactId>
+            <scope>test</scope>
+        </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>
+        <dependency>
+            <groupId>com.flipkart.zjsonpatch</groupId>
+            <artifactId>zjsonpatch</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>
diff --git a/pom.xml b/pom.xml
index 0942fd9c..ca56c799 100644
--- a/pom.xml
+++ b/pom.xml
@@ -18,8 +18,8 @@
 
     <properties>
         <kroxyliciousApi.version>0.1.0-SNAPSHOT</kroxyliciousApi.version>
-        <java.version>11</java.version>
-        <java.test.version>19</java.test.version>
+        <java.version>17</java.version>
+        <java.test.version>17</java.test.version>
         <maven.compiler.parameters>true</maven.compiler.parameters>
         <maven.compiler.release>${java.version}</maven.compiler.release>
         <maven.compiler.testRelease>${java.test.version}</maven.compiler.testRelease>
@@ -32,10 +32,11 @@
 
         <debezium.version>1.9.0.Final</debezium.version>
         <freemarker.version>2.3.32</freemarker.version>
+        <guava.version>31.1-jre</guava.version>
         <jackson.version>2.14.2</jackson.version>
-        <jackson-databind.version>2.14.2</jackson-databind.version>
         <junit.version>5.9.2</junit.version>
         <kafka.version>3.4.0</kafka.version>
+        <kroxy.extension.version>0.1</kroxy.extension.version>
         <log4j.version>2.17.2</log4j.version>
         <netty.version>4.1.89.Final</netty.version>
         <netty.io_uring.version>0.0.14.Final</netty.io_uring.version>
@@ -44,10 +45,11 @@
         <netty.kqueue.classifier>osx-x86_64</netty.kqueue.classifier>
         <lombok.version>1.18.26</lombok.version>
         <assertj-core.version>3.24.2</assertj-core.version>
-        <mockito-core.version>4.6.1</mockito-core.version>
+        <mockito.version>5.2.0</mockito.version>
         <slf4j-api.version>1.7.36</slf4j-api.version>
         <micrometer.version>1.10.4</micrometer.version>
         <kroxy.extension.version>0.1</kroxy.extension.version>
+        <zjsonpatch.version>0.4.14</zjsonpatch.version>
     </properties>
 
     <name>Kroxylicious Parent</name>
@@ -107,34 +109,11 @@
     <dependencyManagement>
         <dependencies>
             <dependency>
-                <groupId>com.fasterxml.jackson.core</groupId>
-                <artifactId>jackson-annotations</artifactId>
-                <version>${jackson.version}</version>
-            </dependency>
-            <dependency>
-                <groupId>com.fasterxml.jackson.core</groupId>
-                <artifactId>jackson-core</artifactId>
-                <version>${jackson.version}</version>
-            </dependency>
-            <dependency>
-                <groupId>com.fasterxml.jackson.core</groupId>
-                <artifactId>jackson-databind</artifactId>
-                <version>${jackson-databind.version}</version>
-            </dependency>
-            <dependency>
-                <groupId>com.fasterxml.jackson.dataformat</groupId>
-                <artifactId>jackson-dataformat-yaml</artifactId>
-                <version>${jackson.version}</version>
-            </dependency>
-            <dependency>
-                <groupId>com.fasterxml.jackson.module</groupId>
-                <artifactId>jackson-module-parameter-names</artifactId>
-                <version>${jackson.version}</version>
-            </dependency>
-            <dependency>
-                <groupId>com.fasterxml.jackson.datatype</groupId>
-                <artifactId>jackson-datatype-jdk8</artifactId>
+                <groupId>com.fasterxml.jackson</groupId>
+                <artifactId>jackson-bom</artifactId>
                 <version>${jackson.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
             </dependency>
             <dependency>
                 <groupId>io.netty</groupId>
@@ -150,11 +129,6 @@
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
-            <dependency>
-                <groupId>org.apache.kafka</groupId>
-                <artifactId>kafka-clients</artifactId>
-                <version>${kafka.version}</version>
-            </dependency>
             <dependency>
                 <groupId>org.apache.logging.log4j</groupId>
                 <artifactId>log4j-bom</artifactId>
@@ -162,11 +136,6 @@
                 <scope>import</scope>
                 <type>pom</type>
             </dependency>
-            <dependency>
-                <groupId>org.freemarker</groupId>
-                <artifactId>freemarker</artifactId>
-                <version>${freemarker.version}</version>
-            </dependency>
             <dependency>
                 <groupId>org.junit</groupId>
                 <artifactId>junit-bom</artifactId>
@@ -174,6 +143,29 @@
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
+            <dependency>
+                <groupId>org.mockito</groupId>
+                <artifactId>mockito-bom</artifactId>
+                <version>${mockito.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.kafka</groupId>
+                <artifactId>kafka-clients</artifactId>
+                <version>${kafka.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.freemarker</groupId>
+                <artifactId>freemarker</artifactId>
+                <version>${freemarker.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>com.google.guava</groupId>
+                <artifactId>guava</artifactId>
+                <version>${guava.version}</version>
+                <scope>test</scope>
+            </dependency>
             <dependency>
                 <groupId>org.slf4j</groupId>
                 <artifactId>slf4j-api</artifactId>
@@ -191,12 +183,6 @@
                 <version>${assertj-core.version}</version>
                 <scope>test</scope>
             </dependency>
-            <dependency>
-                <groupId>org.mockito</groupId>
-                <artifactId>mockito-core</artifactId>
-                <version>${mockito-core.version}</version>
-                <scope>test</scope>
-            </dependency>
             <dependency>
                 <groupId>io.kroxylicious.testing</groupId>
                 <artifactId>testing-junit5-extension</artifactId>
@@ -209,6 +195,12 @@
                 <version>${kroxy.extension.version}</version>
                 <scope>test</scope>
             </dependency>
+            <dependency>
+                <groupId>com.flipkart.zjsonpatch</groupId>
+                <artifactId>zjsonpatch</artifactId>
+                <version>${zjsonpatch.version}</version>
+                <scope>test</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 
@@ -364,6 +356,9 @@
                     <configuration>
                         <skipTests>${skipUTs}</skipTests>
                         <runOrder>random</runOrder>
+                        <statelessTestsetReporter implementation="org.apache.maven.plugin.surefire.extensions.junit5.JUnit5Xml30StatelessReporter">
+                            <usePhrasedTestCaseMethodName>true</usePhrasedTestCaseMethodName>
+                        </statelessTestsetReporter>
                     </configuration>
                 </plugin>
                 <plugin>


Test patch:
diff --git a/integrationtests/pom.xml b/integrationtests/pom.xml
index 1bc6169b..92a9f9bd 100644
--- a/integrationtests/pom.xml
+++ b/integrationtests/pom.xml
@@ -21,6 +21,7 @@
 
     <properties>
         <java.version>19</java.version>
+        <java.test.version>19</java.test.version>
     </properties>
 
     <dependencies>
diff --git a/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java b/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
new file mode 100644
index 00000000..f8c8fd11
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
@@ -0,0 +1,258 @@
+/*
+ * 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.multitenant;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.BiFunction;
+import java.util.function.Predicate;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import org.apache.kafka.common.message.ApiMessageType;
+import org.apache.kafka.common.message.CreateTopicsRequestData;
+import org.apache.kafka.common.message.CreateTopicsRequestDataJsonConverter;
+import org.apache.kafka.common.message.CreateTopicsResponseData;
+import org.apache.kafka.common.message.CreateTopicsResponseDataJsonConverter;
+import org.apache.kafka.common.message.DeleteTopicsRequestData;
+import org.apache.kafka.common.message.DeleteTopicsRequestDataJsonConverter;
+import org.apache.kafka.common.message.DeleteTopicsResponseData;
+import org.apache.kafka.common.message.DeleteTopicsResponseDataJsonConverter;
+import org.apache.kafka.common.message.FetchRequestData;
+import org.apache.kafka.common.message.FetchRequestDataJsonConverter;
+import org.apache.kafka.common.message.FetchResponseData;
+import org.apache.kafka.common.message.FetchResponseDataJsonConverter;
+import org.apache.kafka.common.message.ListOffsetsRequestData;
+import org.apache.kafka.common.message.ListOffsetsRequestDataJsonConverter;
+import org.apache.kafka.common.message.ListOffsetsResponseData;
+import org.apache.kafka.common.message.ListOffsetsResponseDataJsonConverter;
+import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.MetadataRequestDataJsonConverter;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.message.MetadataResponseDataJsonConverter;
+import org.apache.kafka.common.message.OffsetCommitRequestData;
+import org.apache.kafka.common.message.OffsetCommitRequestDataJsonConverter;
+import org.apache.kafka.common.message.OffsetCommitResponseData;
+import org.apache.kafka.common.message.OffsetCommitResponseDataJsonConverter;
+import org.apache.kafka.common.message.OffsetFetchRequestData;
+import org.apache.kafka.common.message.OffsetFetchRequestDataJsonConverter;
+import org.apache.kafka.common.message.OffsetFetchResponseData;
+import org.apache.kafka.common.message.OffsetFetchResponseDataJsonConverter;
+import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData;
+import org.apache.kafka.common.message.OffsetForLeaderEpochRequestDataJsonConverter;
+import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData;
+import org.apache.kafka.common.message.OffsetForLeaderEpochResponseDataJsonConverter;
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.ProduceRequestDataJsonConverter;
+import org.apache.kafka.common.message.ProduceResponseData;
+import org.apache.kafka.common.message.ProduceResponseDataJsonConverter;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.ArgumentCaptor;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import com.flipkart.zjsonpatch.JsonDiff;
+import com.google.common.reflect.ClassPath;
+import com.google.common.reflect.ClassPath.ResourceInfo;
+
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class MultiTenantTransformationFilterTest {
+    private static final Pattern TEST_RESOURCE_FILTER = Pattern.compile(
+            String.format("%s/.*\\.yaml", MultiTenantTransformationFilterTest.class.getPackageName().replace(".", "/")));
+    private static final ObjectMapper MAPPER = new ObjectMapper(new YAMLFactory());
+
+    private record ApiMessageTestDef(ApiMessage message, JsonNode expectedPatch) {
+    }
+
+    private record RequestResponseTestDef(String testName, ApiMessageType apiKey, RequestHeaderData header, ApiMessageTestDef request, ApiMessageTestDef response) {
+    }
+
+    private record Converters(BiFunction<JsonNode, Short, ApiMessage> requestReader,
+                              BiFunction<JsonNode, Short, ApiMessage> responseReader,
+                              BiFunction<ApiMessage, Short, JsonNode> requestWriter,
+                              BiFunction<ApiMessage, Short, JsonNode> responseWriters) {
+    }
+
+    private static final Map<ApiMessageType, Converters> converters = Map.of(
+            ApiMessageType.CREATE_TOPICS, new Converters(
+                    CreateTopicsRequestDataJsonConverter::read,
+                    CreateTopicsResponseDataJsonConverter::read,
+                    (o, ver) -> CreateTopicsRequestDataJsonConverter.write(((CreateTopicsRequestData) o), ver),
+                    (o, ver) -> CreateTopicsResponseDataJsonConverter.write(((CreateTopicsResponseData) o), ver)),
+            ApiMessageType.DELETE_TOPICS, new Converters(
+                    DeleteTopicsRequestDataJsonConverter::read,
+                    DeleteTopicsResponseDataJsonConverter::read,
+                    (o, ver) -> DeleteTopicsRequestDataJsonConverter.write(((DeleteTopicsRequestData) o), ver),
+                    (o, ver) -> DeleteTopicsResponseDataJsonConverter.write(((DeleteTopicsResponseData) o), ver)),
+            ApiMessageType.METADATA, new Converters(
+                    MetadataRequestDataJsonConverter::read,
+                    MetadataResponseDataJsonConverter::read,
+                    (o, ver) -> MetadataRequestDataJsonConverter.write(((MetadataRequestData) o), ver),
+                    (o, ver) -> MetadataResponseDataJsonConverter.write(((MetadataResponseData) o), ver)),
+            ApiMessageType.PRODUCE, new Converters(
+                    ProduceRequestDataJsonConverter::read,
+                    ProduceResponseDataJsonConverter::read,
+                    (o, ver) -> ProduceRequestDataJsonConverter.write(((ProduceRequestData) o), ver),
+                    (o, ver) -> ProduceResponseDataJsonConverter.write(((ProduceResponseData) o), ver)),
+            ApiMessageType.LIST_OFFSETS, new Converters(
+                    ListOffsetsRequestDataJsonConverter::read,
+                    ListOffsetsResponseDataJsonConverter::read,
+                    (o, ver) -> ListOffsetsRequestDataJsonConverter.write(((ListOffsetsRequestData) o), ver),
+                    (o, ver) -> ListOffsetsResponseDataJsonConverter.write(((ListOffsetsResponseData) o), ver)),
+            ApiMessageType.OFFSET_FETCH, new Converters(
+                    OffsetFetchRequestDataJsonConverter::read,
+                    OffsetFetchResponseDataJsonConverter::read,
+                    (o, ver) -> OffsetFetchRequestDataJsonConverter.write(((OffsetFetchRequestData) o), ver),
+                    (o, ver) -> OffsetFetchResponseDataJsonConverter.write(((OffsetFetchResponseData) o), ver)),
+            ApiMessageType.OFFSET_FOR_LEADER_EPOCH, new Converters(
+                    OffsetForLeaderEpochRequestDataJsonConverter::read,
+                    OffsetForLeaderEpochResponseDataJsonConverter::read,
+                    (o, ver) -> OffsetForLeaderEpochRequestDataJsonConverter.write(((OffsetForLeaderEpochRequestData) o), ver),
+                    (o, ver) -> OffsetForLeaderEpochResponseDataJsonConverter.write(((OffsetForLeaderEpochResponseData) o), ver)),
+            ApiMessageType.OFFSET_COMMIT, new Converters(
+                    OffsetCommitRequestDataJsonConverter::read,
+                    OffsetCommitResponseDataJsonConverter::read,
+                    (o, ver) -> OffsetCommitRequestDataJsonConverter.write(((OffsetCommitRequestData) o), ver),
+                    (o, ver) -> OffsetCommitResponseDataJsonConverter.write(((OffsetCommitResponseData) o), ver)),
+            ApiMessageType.FETCH, new Converters(
+                    FetchRequestDataJsonConverter::read,
+                    FetchResponseDataJsonConverter::read,
+                    (o, ver) -> FetchRequestDataJsonConverter.write(((FetchRequestData) o), ver),
+                    (o, ver) -> FetchResponseDataJsonConverter.write(((FetchResponseData) o), ver)));
+
+    private static Stream<RequestResponseTestDef> requestResponseTestDefinitions() throws Exception {
+        var resourceInfoStream = ClassPath.from(MultiTenantTransformationFilterTest.class.getClassLoader()).getResources().stream()
+                .filter(ri -> TEST_RESOURCE_FILTER.matcher(ri.getResourceName()).matches()).toList();
+
+        // https://youtrack.jetbrains.com/issue/IDEA-315462: we've seen issues in IDEA in IntelliJ Workspace Model API mode where test resources
+        // don't get added to the Junit runner classpath. You can work around by not using that mode, or by adding src/test/resources to the
+        // runner's classpath using 'modify classpath' option in the dialogue.
+        checkState(!resourceInfoStream.isEmpty(), "no test resource files found on classpath matching %s", TEST_RESOURCE_FILTER);
+
+        return resourceInfoStream.stream()
+                .map(MultiTenantTransformationFilterTest::readTestResource)
+                .flatMap(Arrays::stream)
+                .filter(Predicate.not(td -> td.get("disabled").asBoolean(false)))
+                .map(MultiTenantTransformationFilterTest::buildRequestResponseTestDef);
+    }
+
+    private static JsonNode[] readTestResource(ResourceInfo resourceInfo) {
+        try {
+            var sourceFile = Path.of(resourceInfo.url().getPath()).getFileName().toString();
+            var jsonNodes = MAPPER.reader().readValue(resourceInfo.url(), JsonNode[].class);
+            for (int i = 0; i < jsonNodes.length; i++) {
+                if (jsonNodes[i] instanceof ObjectNode objectNode) {
+                    objectNode.put("source", String.format("%s[%d]", sourceFile, i));
+                }
+            }
+            return jsonNodes;
+        }
+        catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+    }
+
+    private static RequestResponseTestDef buildRequestResponseTestDef(JsonNode jsonNode) {
+        var sourceNode = jsonNode.get("source");
+        var descriptionNode = jsonNode.get("description");
+        var apiMessageTypeNode = jsonNode.get("apiMessageType");
+        var versionNode = jsonNode.get("version");
+        checkNotNull(sourceNode, "source missing from test definition");
+        var source = sourceNode.textValue();
+        checkNotNull(apiMessageTypeNode, "apiMessageType missing from test definition: %s", source);
+        checkNotNull(versionNode, "version missing from test definition: %s", source);
+        var messageType = ApiMessageType.valueOf(apiMessageTypeNode.asText());
+        var header = new RequestHeaderData().setRequestApiKey(messageType.apiKey()).setRequestApiVersion(versionNode.shortValue());
+
+        var testName = String.format("%s,%s,v%d[%s]", source, messageType, versionNode.intValue(),
+                Optional.ofNullable(descriptionNode).map(JsonNode::asText).orElse("-"));
+        var conv = converters.get(messageType);
+        var request = buildApiMessageTestDef(header.requestApiVersion(), jsonNode.get("request"), conv.requestReader());
+        var response = buildApiMessageTestDef(header.requestApiVersion(), jsonNode.get("response"), conv.responseReader());
+        return new RequestResponseTestDef(testName, messageType, header, request, response);
+    }
+
+    private static ApiMessageTestDef buildApiMessageTestDef(short apiVersion, JsonNode node, BiFunction<JsonNode, Short, ApiMessage> readerFunc) {
+        if (node != null) {
+            var request = readerFunc.apply(node.get("payload"), apiVersion);
+            var requestDiff = node.get("diff");
+            return new ApiMessageTestDef(request, requestDiff);
+        }
+        return null;
+    }
+
+    private final MultiTenantTransformationFilter filter = new MultiTenantTransformationFilter();
+
+    private final KrpcFilterContext context = mock(KrpcFilterContext.class);
+
+    private final ArgumentCaptor<ApiMessage> apiMessageCaptor = ArgumentCaptor.forClass(ApiMessage.class);
+
+    @BeforeEach
+    public void beforeEach() {
+        when(context.sniHostname()).thenReturn("tenant1.kafka.example.com");
+    }
+
+    public static Stream<Arguments> requests() throws Exception {
+        return requestResponseTestDefinitions().map(t -> Arguments.of(t.testName(), t.apiKey(), t.header(), t.request()));
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource(value = "requests")
+    void requestsTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef requestTestDef) {
+        var request = requestTestDef.message();
+        // marshalled the request object back to json, this is used for the comparison later.
+        var requestWriter = converters.get(apiMessageType).requestWriter();
+        var marshalled = requestWriter.apply(request, header.requestApiVersion());
+
+        filter.onRequest(ApiKeys.forId(apiMessageType.apiKey()), header, request, context);
+        verify(context).forwardRequest(apiMessageCaptor.capture());
+
+        var filtered = requestWriter.apply(apiMessageCaptor.getValue(), header.requestApiVersion());
+        assertEquals(requestTestDef.expectedPatch(), JsonDiff.asJson(marshalled, filtered));
+    }
+
+    public static Stream<Arguments> responses() throws Exception {
+        return requestResponseTestDefinitions().map(t -> Arguments.of(t.testName(), t.apiKey(), t.header(), t.response()));
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource(value = "responses")
+    void responseTransformed(@SuppressWarnings("unused") String testName, ApiMessageType apiMessageType, RequestHeaderData header, ApiMessageTestDef responseTestDef) {
+        var response = responseTestDef.message();
+        // marshalled the response object back to json, this is used for comparison later.
+        var responseWriter = converters.get(apiMessageType).responseWriters();
+
+        var marshalled = responseWriter.apply(response, header.requestApiVersion());
+
+        filter.onResponse(ApiKeys.forId(apiMessageType.apiKey()), new ResponseHeaderData(), response, context);
+        verify(context).forwardResponse(apiMessageCaptor.capture());
+
+        var filtered = responseWriter.apply(apiMessageCaptor.getValue(), header.requestApiVersion());
+        assertEquals(responseTestDef.expectedPatch(), JsonDiff.asJson(marshalled, filtered));
+    }
+}
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/CreateTopic.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/CreateTopic.test.yaml
new file mode 100644
index 00000000..dbaaf69f
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/CreateTopic.test.yaml
@@ -0,0 +1,39 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: CREATE_TOPICS
+  version: 7
+  request:
+    payload:
+      topics:
+        - name: foo
+          numPartitions: 0
+          replicationFactor: 1
+          assignments: [ ]
+          configs: [ ]
+      timeoutMs: 0
+      validateOnly: false
+    diff:
+      - op: replace
+        path: "/topics/0/name"
+        value: tenant1-foo
+  response:
+    payload:
+      topics:
+        - name: tenant1-foo
+          topicId: AAAAAAAAAAAAAAAAAAAAAA
+          numPartitions: 0
+          replicationFactor: 1
+          assignments: [ ]
+          configs: [ ]
+          errorCode: 0
+          errorMessage:
+      throttleTimeMs: 0
+    diff:
+      - op: replace
+        path: "/topics/0/name"
+        value: foo
+  disabled: false
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/DeleteTopic.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/DeleteTopic.test.yaml
new file mode 100644
index 00000000..3cc17298
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/DeleteTopic.test.yaml
@@ -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
+#
+
+- apiMessageType: DELETE_TOPICS
+  version: 7
+  request:
+    payload:
+      topics:
+        - name: foo
+          topicId: AAAAAAAAAAAAAAAAAAAAAA
+      timeoutMs: 0
+      validateOnly: false
+    diff:
+      - op: replace
+        path: "/topics/0/name"
+        value: tenant1-foo
+  response:
+    payload:
+      responses:
+        - name: tenant1-foo
+          topicId: AAAAAAAAAAAAAAAAAAAAAA
+          errorCode: 0
+          errorMessage:
+      throttleTimeMs: 0
+    diff:
+      - op: replace
+        path: "/responses/0/name"
+        value: foo
+  disabled: false
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/Fetch.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/Fetch.test.yaml
new file mode 100644
index 00000000..3ad530e2
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/Fetch.test.yaml
@@ -0,0 +1,94 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: FETCH
+  version: 12
+  request:
+    payload:
+      replicaId: 0
+      maxWaitMs: 0
+      minBytes: 0
+      maxBytes: 0
+      isolationLevel: 0
+      sessionId: 0
+      sessionEpoch: 0
+      topics:
+        - topic: foo
+          partitions:
+            - partition: 0
+              currentLeaderEpoch: 0
+              fetchOffset: 0
+              lastFetchedEpoch: 0
+              logStartOffset: 0
+              partitionMaxBytes: 0
+      forgottenTopicsData: []
+      rackId: us-east1
+    diff:
+      - op: replace
+        path: "/topics/0/topic"
+        value: tenant1-foo
+  response:
+    payload:
+      errorCode: 0
+      sessionId: 0
+      responses:
+        - topic: tenant1-foo
+          partitions:
+            - partitionIndex: 0
+              highWatermark: 0
+              lastStableOffset: 0
+              abortedTransactions: []
+              preferredReadReplica: 0
+              records: null
+              errorCode: 0
+              logStartOffset: 0
+      throttleTimeMs: 0
+    diff:
+      - op: replace
+        path: "/responses/0/topic"
+        value: foo
+  disabled: false
+- apiMessageType: FETCH
+  version: 13
+  request:
+    payload:
+      replicaId: 0
+      maxWaitMs: 0
+      minBytes: 0
+      maxBytes: 0
+      isolationLevel: 0
+      sessionId: 0
+      sessionEpoch: 0
+      topics:
+        - topicId: AAAAAAAAAAAAAAAAAAAAAA
+          partitions:
+            - partition: 0
+              currentLeaderEpoch: 0
+              fetchOffset: 0
+              lastFetchedEpoch: 0
+              logStartOffset: 0
+              partitionMaxBytes: 0
+      forgottenTopicsData: []
+      rackId: us-east1
+    diff: []
+  response:
+    payload:
+      errorCode: 0
+      sessionId: 0
+      responses:
+        - topicId: AAAAAAAAAAAAAAAAAAAAAA
+          partitions:
+            - partitionIndex: 0
+              highWatermark: 0
+              lastStableOffset: 0
+              abortedTransactions: []
+              preferredReadReplica: 0
+              records: null
+              errorCode: 0
+              logStartOffset: 0
+      throttleTimeMs: 0
+    diff: []
+  disabled: false
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/ListOffsets.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/ListOffsets.test.yaml
new file mode 100644
index 00000000..d835a557
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/ListOffsets.test.yaml
@@ -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
+#
+
+- apiMessageType: LIST_OFFSETS
+  version: 7
+  request:
+    payload:
+      replicaId: 1
+      isolationLevel: 0
+      topics:
+        - name: foo
+          partitions:
+            - partitionIndex: 0
+              currentLeaderEpoch: 0
+              timestamp: 0
+    diff:
+      - op: replace
+        path: "/topics/0/name"
+        value: tenant1-foo
+  response:
+    payload:
+      topics:
+        - name: tenant1-foo
+          partitions:
+            - partitionIndex: 0
+              errorCode: 0
+              offset: 0
+              leaderEpoch: 0
+              timestamp: 0
+      throttleTimeMs: 0
+    diff:
+      - op: replace
+        path: "/topics/0/name"
+        value: foo
+  disabled: false
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/Metadata.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/Metadata.test.yaml
new file mode 100644
index 00000000..b43364c2
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/Metadata.test.yaml
@@ -0,0 +1,49 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: METADATA
+  version: 12
+  request:
+    payload:
+      topics:
+        - name: foo
+          topicId: AAAAAAAAAAAAAAAAAAAAAA
+      allowAutoTopicCreation: true
+      includeTopicAuthorizedOperations: true
+    diff:
+      - op: replace
+        path: "/topics/0/name"
+        value: tenant1-foo
+  response:
+    payload:
+      topics:
+        - name: tenant1-foo
+          topicId: AAAAAAAAAAAAAAAAAAAAAA
+          isInternal: false
+          topicAuthorizedOperations: 0
+          errorCode: 0
+          errorMessage:
+          partitions:
+            - partitionIndex: 0
+              leaderId: 1
+              leaderEpoch: 1
+              replicaNodes: []
+              isrNodes: []
+              offlineReplicas: []
+              errorCode: 0
+      brokers:
+        - nodeId: 1
+          host: localhost
+          port: 9999
+          rack: us-east1
+      clusterId: AAAAAAAAAAAAAAAAAAAAAA
+      controllerId: 1
+      throttleTimeMs: 0
+    diff:
+      - op: replace
+        path: "/topics/0/name"
+        value: foo
+  disabled: false
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetCommit.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetCommit.test.yaml
new file mode 100644
index 00000000..0c271af1
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetCommit.test.yaml
@@ -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
+#
+
+- apiMessageType: OFFSET_COMMIT
+  version: 8
+  request:
+    payload:
+      groupId: mygroup
+      generationId: 0
+      memberId: mymemberid
+      groupInstanceId: mygroupinstanceid
+      topics:
+        - name: foo
+          partitions:
+            - partitionIndex: 0
+              committedOffset: 0
+              committedLeaderEpoch: 0
+              committedMetadata: null
+    diff:
+      - op: replace
+        path: "/topics/0/name"
+        value: tenant1-foo
+  response:
+    payload:
+      topics:
+        - name: tenant1-foo
+          partitions:
+            - partitionIndex: 0
+              errorCode: 0
+      throttleTimeMs: 0
+    diff:
+      - op: replace
+        path: "/topics/0/name"
+        value: foo
+  disabled: false
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetFetch.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetFetch.test.yaml
new file mode 100644
index 00000000..ebd1b78f
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetFetch.test.yaml
@@ -0,0 +1,39 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: OFFSET_FETCH
+  version: 8
+  request:
+    payload:
+      groups:
+        - groupId: mygroup
+          topics:
+            - name: foo
+              partitionIndexes: [0]
+      requireStable: false
+    diff:
+      - op: replace
+        path: "/groups/0/topics/0/name"
+        value: tenant1-foo
+  response:
+    payload:
+      groups:
+        - groupId: mygroup
+          topics:
+            - name: tenant1-foo
+              partitions:
+                - partitionIndex: 0
+                  committedOffset: 0
+                  committedLeaderEpoch: 0
+                  metadata: null
+                  errorCode: 0
+          errorCode: 0
+      throttleTimeMs: 0
+    diff:
+      - op: replace
+        path: "/groups/0/topics/0/name"
+        value: foo
+  disabled: false
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetForLeaderEpoch.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetForLeaderEpoch.test.yaml
new file mode 100644
index 00000000..3d21ee9b
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/OffsetForLeaderEpoch.test.yaml
@@ -0,0 +1,36 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: OFFSET_FOR_LEADER_EPOCH
+  version: 4
+  request:
+    payload:
+      replicaId: 1
+      topics:
+          - topic: foo
+            partitions:
+              - partition: 0
+                currentLeaderEpoch: 0
+                leaderEpoch: 0
+    diff:
+      - op: replace
+        path: "/topics/0/topic"
+        value: tenant1-foo
+  response:
+    payload:
+      topics:
+        - topic: tenant1-foo
+          partitions:
+            - partition: 0
+              errorCode: 0
+              leaderEpoch: 0
+              endOffset: 0
+      throttleTimeMs: 0
+    diff:
+      - op: replace
+        path: "/topics/0/topic"
+        value: foo
+  disabled: false
diff --git a/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/Produce.test.yaml b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/Produce.test.yaml
new file mode 100644
index 00000000..2379493e
--- /dev/null
+++ b/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/Produce.test.yaml
@@ -0,0 +1,43 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: PRODUCE
+  version: 9
+  request:
+    payload:
+      transactionalId: null
+      acks: -1
+      timeoutMs: 0
+      topicData:
+        - name: foo
+          partitionData:
+            - index: 0
+              records: !!binary |
+                Zm9v
+    diff:
+      - op: replace
+        path: "/topicData/0/name"
+        value: tenant1-foo
+  response:
+    payload:
+      responses:
+        - name: tenant1-foo
+          partitionResponses:
+            - index: 0
+              errorCode: 0
+              baseOffset: 0
+              logAppendTimeMs: 0
+              logStartOffset: 0
+              recordErrors:
+                - batchIndex: 0
+                  batchIndexErrorMessage: null
+              errorMessage: null
+      throttleTimeMs: 0
+    diff:
+      - op: replace
+        path: "/responses/0/name"
+        value: foo
+  disabled: false