Repository: kroxylicious/kroxylicious
Pull request #171 (2023-03-05 18:19:29)

Unit tests for MultiTenantTransformationFilterTest
The POC didn't contribute any unit tests for the multitenant filter.
I'd like to fix that.



k-wall: Constructing the the kafka value objects in a fluent concise manner doesn't seem to be terribly easy.   The best way I've found so far is to lean on the JSON request/response transformation abilities of the Kafka itself.   Doing that gives a parameterized test, which I think, remains readable.

What do you think to the approach?

robobario: Maybe the JSON blobs could be externalised? Like have some path in `src/test/resources` that contains pairs of files like `request-create-topic-in.json` and `request-create-topic-expected-diff.json`. This might make it easier to manipulate them in bulk later since you could operate on them with `jq` or language-of-choice if you needed to update lots of inputs or assertions in future.

Also for unit level I'd rather have the request and response methods checked separately, despite them forming a logical pair in reality. So I know that one test is checking one request or one response.
robobario: Also I can see there is an argument for parameterisation as we are checking for topicname prefixing in each case, but practically I find that argument array hard to read and wonder if we are better off splitting it into individual methods.
k-wall: > Maybe the JSON blobs could be externalised? 

+1 to the json file idea, though I think having the request/response along side the expected diff might be best.  I'll try that refactoring.  

> Also for unit level I'd rather have the request and response methods checked separately, despite them forming a logical pair in reality. So I know that one test is checking one request or one response.

I think that request and response test do belong together.  Request and response formats have evolved together (e.g. the evolution towards topic ids), so I think having a single file expressing both halves will make for easier developer comprehension.

I am thinking of a JSON file with an object structured like this:

```yaml
[
   - testName: 
     version
     request:
       payload:
       diff:
     response:
       payload:
       diff:
]
```

and have a file per RPC.

> Also I can see there is an argument for parameterisation as we are checking for topicname prefixing in each case, but practically I find that argument array hard to read and wonder if we are better off splitting it into individual methods.
> edit: I imagine this can be improved, but currently if a test failed it won't be immediately clear which case has failed.

I think the tooling should make it clear which parameter is failure and the IDE let you re-run invoke a single instance of a parameterized test.



robobario: > I think the tooling should make it clear which parameter is failure and the IDE let you re-run invoke a single instance of a parameterized test.

Yes, but the parameters could be unreadable. In the usage so far it works since you see `apiMessageType=DELETE` in the failure message you'll know roughly file a failure came from. But if we make this technique available to other filter authors it may be they have lots of similar looking cases for the same message type with different input. If we are going for an array of scenarios per file I think it would help understanding if we can add a description to each scenario that shows up somewhere in the error.
k-wall: > > I think the tooling should make it clear which parameter is failure and the IDE let you re-run invoke a single instance of a parameterized test.
> 
> Yes, but the parameters could be unreadable. In the usage so far it works since you see something like: `apiMessageType=DELETE` in the failure message you'll know roughly which file a failure came from.

I did have thoughts about either having the yaml carry a user-defined test name, or exposing the test case filename/array index as parameterized test parameters.

> But if we make this technique available to other filter authors it may be they have lots of similar looking cases for the same message type with different input. If we are going for an array of scenarios per file I think it would help understanding if we can add an optional description to each scenario that shows up somewhere in the error.

I vote we wait until we discover other filter use-cases with similar test requirements before making the test mechanism part of a testing API.  I'm sure we'll want to rev the test mechanism as we build out multitenant functions.  Let's not abstract too early.



robobario: Note: @SamBarker found that tests weren't running in IDEA when using `Import using the new IntelliJ Workspace Model API (experimental)`. You see:
```
org.junit.platform.commons.PreconditionViolationException: Configuration error: You must configure at least one set of arguments for this @ParameterizedTest
```
when the tests are run.
k-wall: > Note: @SamBarker found that tests weren't running in IDEA when using `Import using the new IntelliJ Workspace Model API (experimental)`. You see:
> 
> ```
> org.junit.platform.commons.PreconditionViolationException: Configuration error: You must configure at least one set of arguments for this @ParameterizedTest
> ```
> 
> when the tests are run.

Yeah I've noticed that too.  It seems that Intellij isn't proper setting up the classpath to include resources when in that mode.
k-wall: > > Note: @SamBarker found that tests weren't running in IDEA when using `Import using the new IntelliJ Workspace Model API (experimental)`. You see:
> > ```
> > org.junit.platform.commons.PreconditionViolationException: Configuration error: You must configure at least one set of arguments for this @ParameterizedTest
> > ```
> > 
> > 
> >     
> >       
> >     
> > 
> >       
> >     
> > 
> >     
> >   
> > when the tests are run.
> 
> Yeah I've noticed that too. It seems that Intellij isn't properly setting up the classpath to include resources when in that mode.

I spent a bit more time looking into this without making much progress.   The issue is Intellij doesn't copy the test-resources to the test-classes, so the resources don't end up on the classpath.  You can fix by doing this.

```bash
 mvn process-test-resources -f kroxylicious-multitenant
```

* I tried setting up a new multi module maven project in the  IntelliJ Workspace Model  mode and couldn't reproduce the issue. 
* I also tried cleaning kroxylicious's Intellij files away and reimporting from scratch. No help.
* I reproduced the problem in other modules within kroxylicious, so I confident that issue isn't uniqie to the multi-tenant model.



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/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..b34de4d6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -32,10 +32,12 @@
 
         <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>
@@ -48,6 +50,7 @@
         <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>
@@ -167,6 +170,12 @@
                 <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.junit</groupId>
                 <artifactId>junit-bom</artifactId>
@@ -209,6 +218,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 +379,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/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