Repository: kroxylicious/kroxylicious
Pull request #195 (2023-03-14 11:40:25)

Fix some javadoc warnings 
### Type of change

_Select the type of your PR_

- Documentation

### Description

Cleaning up some javadoc errors.  No functional change.

The sea of yellow when you run `mvn verify` gets tiring. 

### Additional Context

_Why are you making this pull request?_



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/README.md b/README.md
index 44237b16..5ad0b1f1 100644
--- a/README.md
+++ b/README.md
@@ -194,4 +194,7 @@ While Kroxylicious is a java application we've had reports of issues running the
       Expect a version string similar to `git version 2.37.1 (Apple Git-137.1)`
    2. Follow the [WSL-git tutorial](https://learn.microsoft.com/en-us/windows/wsl/tutorials/wsl-git) if needed.
 5. Checkout Kroxylicious to `${kroxylicious-checkout}`
-6. Build & develop following the [standard build](#Build) instructions
\ No newline at end of file
+6. Build & develop following the [standard build](#Build) instructions
+
+## Releasing the project
+See [The release guide](docs/Releasing.adoc)
\ No newline at end of file
diff --git a/api/kroxylicious-api/pom.xml b/api/kroxylicious-api/pom.xml
index 41324858..e4fa4612 100644
--- a/api/kroxylicious-api/pom.xml
+++ b/api/kroxylicious-api/pom.xml
@@ -14,12 +14,12 @@
     <parent>
         <groupId>io.kroxylicious</groupId>
         <artifactId>kroxylicious-parent</artifactId>
-        <version>1.0-SNAPSHOT</version>
+        <version>0.1.0-SNAPSHOT</version>
         <relativePath>../../pom.xml</relativePath>
     </parent>
 
     <artifactId>kroxylicious-api</artifactId>
-    <version>0.0.0-SNAPSHOT</version>
+    <version>0.1.0-SNAPSHOT</version>
     <packaging>jar</packaging>
 
     <name>Kroxylicious API</name>
diff --git a/api/kroxylicious-filter-api/pom.xml b/api/kroxylicious-filter-api/pom.xml
index 3863adc4..8636f8dc 100644
--- a/api/kroxylicious-filter-api/pom.xml
+++ b/api/kroxylicious-filter-api/pom.xml
@@ -14,12 +14,12 @@
     <parent>
         <groupId>io.kroxylicious</groupId>
         <artifactId>kroxylicious-parent</artifactId>
-        <version>1.0-SNAPSHOT</version>
+        <version>0.1.0-SNAPSHOT</version>
         <relativePath>../../pom.xml</relativePath>
     </parent>
-    
+
     <artifactId>kroxylicious-filter-api</artifactId>
-    <version>0.0.0-SNAPSHOT</version>
+    <version>0.1.0-SNAPSHOT</version>
     <packaging>jar</packaging>
 
     <properties>
@@ -33,7 +33,7 @@
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-api</artifactId>
-            <version>0.0.0-SNAPSHOT</version>
+            <version>${kroxyliciousApi.version}</version>
         </dependency>
         <dependency>
             <groupId>io.micrometer</groupId>
@@ -108,6 +108,7 @@
                             <messageSpecFilter>*{Request,Response}.json</messageSpecFilter>
                             <templateDirectory>${project.basedir}/src/main/templates</templateDirectory>
                             <templateNames>Kproxy/Filter.ftl</templateNames>
+                            <!--suppress UnresolvedMavenProperty -->
                             <outputFilePattern>${messageSpecName}Filter.java</outputFilePattern>
                             <outputPackage>io.kroxylicious.proxy.filter</outputPackage>
                             <outputDirectory>${project.build.directory}/generated-sources/krpc</outputDirectory>
@@ -124,6 +125,7 @@
                             <messageSpecFilter>*{Request,Response}.json</messageSpecFilter>
                             <templateDirectory>${project.basedir}/src/main/templates</templateDirectory>
                             <templateNames>Kproxy/KrpcFilter.ftl</templateNames>
+                            <!--suppress UnresolvedMavenProperty -->
                             <outputFilePattern>${templateName}.java</outputFilePattern>
                             <outputPackage>io.kroxylicious.proxy.filter</outputPackage>
                             <outputDirectory>${project.build.directory}/generated-sources/krpc</outputDirectory>
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/docs/Releasing.adoc b/docs/Releasing.adoc
new file mode 100644
index 00000000..59c851d2
--- /dev/null
+++ b/docs/Releasing.adoc
@@ -0,0 +1,64 @@
+= Releasing Kroxylicious
+3.0, July 29, 2022: AsciiDoc article template
+:toc:
+:icons: font
+:url-quickref: https://docs.asciidoctor.org/asciidoc/latest/syntax-quick-reference/
+
+A quick and dirty guide to actually releasing builds for consumption via maven central.
+
+== Requirements
+.Maven Central
+* A Maven Central/Sonatype account. Accounts are created via their https://issues.sonatype.org/secure/Signup!default.jspa[JIRA signup]
+* Your Sonatype account needs to be granted publication permissions for the Kroxylicous project.
+** One of the existing team members with publishing rights needs to raise a https://issues.sonatype.org/secure/CreateIssue.jspa?pid=10134&issuetype=11003[JIRA ticket] to request the new account is added (or removed...). The requirements for the ticket are listed in the https://central.sonatype.org/publish/manage-permissions/[Central Docs]
+
+Signing the builds with a GPG key is a requirement for publishing to Maven Central.
+
+.PGP
+* Ensure your PGP key is published to a public key server (for example keys.openpgp.org)
+* Get your PGP key signed by the Kroxylicious project key. Rather than share a single key between project members (which has issues with revocation and re-issuing) we sign the releases individually using a key signed by the project key as a mark of trust.
+* A copy of your keys short ID.
+
+TIP: To get the short ID for your key `gpg2 --show-keys --keyid-format short $\{YOUR_KEY_LONG_ID}`
+
+== Running a release
+
+.Releasing Locally
+. `mvn clean install`
+.. We are looking for a clean and green build.
+.. If it's not, shout scream and swear.
+.. Think *very very* carefully if its worth proceeding. Releases are forever!
+. Switch to the release version number.
+.. `export RELEASE_VERSION=0.1.0`
+.. `export RELEASE_API_VERSION=0.1.0`
+.. `mvn versions:set -DnewVersion=$\{RELEASE_VERSION} -pl '!:kroxylicious-api,!:kroxylicious-filter-api'`
+.. `mvn versions:set -DnewVersion=$\{RELEASE_API_VERSION} -pl ':kroxylicious-api,:kroxylicious-filter-api'`
+.. `mvn versions:set-property -Dproperty=kroxyliciousApi.version -DnewVersion=$\{RELEASE_API_VERSION}`
+.. `mvn clean install -Pquick`
+// .. *TODO*: Should we use the https://maven.apache.org/maven-release/maven-release-plugin[release plug-in] instead? .
+.. `git add **/pom.xml`
+.. `git commit -m "Release version v$\{RELEASE_VERSION}`
+.. `git tag v$\{RELEASE_VERSION}`
+.. `git push upstream --tags`
+. `mvn deploy -Prelease -DskipTests=true -DreleaseSigningKey=<YOUR_KEY_SHORT_ID>`
+.. The `release` profile enables GPG signing
+.. The release profile will not run without specifying the `releaseSigningKey` property.
+. Validate the release in the central Nexus instance.
+.. Locate the https://central.sonatype.org/publish/release/#locate-and-examine-your-staging-repository[Staging repository]
+.. Check that the right tree of artefacts are there, ensure there are source & docs jars.
+// .. *TODO*: Using the staging repository to validate the release.
+. `Close` the staging repository in the central Nexus instance, see the https://central.sonatype.org/publish/release/#close-and-drop-or-release-your-staging-repository[Central release docs]. This triggers the promotion from staging to being publicly available on central.
+. Switch to the next version number for development.
+.. `export DEV_VERSION=0.1.0-SNAPSHOT`
+.. `export DEV_API_VERSION=0.1.0-SNAPSHOT`
+.. `mvn versions:set -DnewVersion=$\{DEV_VERSION} -pl '!:kroxylicious-api,!:kroxylicious-filter-api'`
+.. `mvn versions:set -DnewVersion=$\{DEV_API_VERSION} -pl ':kroxylicious-api,:kroxylicious-filter-api'`
+.. `mvn versions:set-property -Dproperty=kroxyliciousApi.version -DnewVersion=$\{DEV_API_VERSIION}`
+.. `git add **/pom.xml`
+.. `git commit -m "Next development version v$\{DEV_VERSION}`
+.. `git push`
+
+.Releasing via the build server
+**TBD**
+
+Specifically we need to work out the commit rights and key signing for the build server.
\ No newline at end of file
diff --git a/kroxylicious-multitenant/pom.xml b/kroxylicious-multitenant/pom.xml
index 6af745eb..7796be7a 100644
--- a/kroxylicious-multitenant/pom.xml
+++ b/kroxylicious-multitenant/pom.xml
@@ -14,7 +14,7 @@
     <parent>
         <groupId>io.kroxylicious</groupId>
         <artifactId>kroxylicious-parent</artifactId>
-        <version>1.0-SNAPSHOT</version>
+        <version>0.1.0-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 
@@ -28,12 +28,53 @@
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-api</artifactId>
-            <version>0.0.0-SNAPSHOT</version>
+            <version>${kroxyliciousApi.version}</version>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-filter-api</artifactId>
-            <version>0.0.0-SNAPSHOT</version>
+            <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/kroxylicious/pom.xml b/kroxylicious/pom.xml
index c6097c36..0eff2f43 100644
--- a/kroxylicious/pom.xml
+++ b/kroxylicious/pom.xml
@@ -14,7 +14,7 @@
     <parent>
         <groupId>io.kroxylicious</groupId>
         <artifactId>kroxylicious-parent</artifactId>
-        <version>1.0-SNAPSHOT</version>
+        <version>0.1.0-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 
@@ -31,12 +31,12 @@
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-api</artifactId>
-            <version>0.0.0-SNAPSHOT</version>
+            <version>${kroxyliciousApi.version}</version>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-filter-api</artifactId>
-            <version>0.0.0-SNAPSHOT</version>
+            <version>${kroxyliciousApi.version}</version>
         </dependency>
         <!-- We depend on the specific Netty dependencies not netty-all to reduce the size of fatjars -->
         <dependency>
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
index 1ea3cbfd..ce8aeabe 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
@@ -22,6 +22,7 @@ import javax.net.ssl.KeyManagerFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import io.micrometer.core.instrument.Metrics;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
@@ -52,6 +53,10 @@ import io.kroxylicious.proxy.internal.MeterRegistries;
 import io.kroxylicious.proxy.internal.admin.AdminHttpInitializer;
 import io.kroxylicious.proxy.internal.filter.FixedNetFilter;
 
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES;
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES;
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_REQUEST_SIZE_BYTES;
+
 public final class KafkaProxy implements AutoCloseable {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProxy.class);
@@ -205,6 +210,10 @@ public final class KafkaProxy implements AutoCloseable {
             bindFuture = serverBootstrap.bind(proxyPort);
         }
         acceptorChannel = bindFuture.sync().channel();
+        // Pre-register counters/summaries to avoid creating them on first request and thus skewing the request latency
+        Metrics.counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES);
+        Metrics.counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES);
+        Metrics.summary(KROXYLICIOUS_REQUEST_SIZE_BYTES);
         return this;
     }
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java
index 13f1e520..eada7f32 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java
@@ -5,6 +5,8 @@
  */
 package io.kroxylicious.proxy.internal.codec;
 
+import java.util.List;
+
 import org.apache.kafka.common.message.AddOffsetsToTxnRequestData;
 import org.apache.kafka.common.message.AddPartitionsToTxnRequestData;
 import org.apache.kafka.common.message.ApiVersionsRequestData;
@@ -42,6 +44,8 @@ import org.apache.kafka.common.protocol.Readable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Tag;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 
@@ -50,6 +54,11 @@ import io.kroxylicious.proxy.frame.Frame;
 import io.kroxylicious.proxy.frame.OpaqueRequestFrame;
 import io.kroxylicious.proxy.frame.RequestFrame;
 
+import static io.kroxylicious.proxy.internal.util.Metrics.FLOWING_UPSTREAM;
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES;
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES;
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_REQUEST_SIZE_BYTES;
+
 public class KafkaRequestDecoder extends KafkaMessageDecoder {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaRequestDecoder.class);
@@ -85,12 +94,19 @@ public class KafkaRequestDecoder extends KafkaMessageDecoder {
 
         RequestHeaderData header = null;
         final ByteBufAccessorImpl accessor;
+        Metrics.counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES, List.of(Tag.of("flowing", "downstream"))).increment();
         var decodeRequest = decodePredicate.shouldDecodeRequest(apiKey, apiVersion);
         LOGGER.debug("Decode {}/v{} request? {}, Predicate {} ", apiKey, apiVersion, decodeRequest, decodePredicate);
         boolean decodeResponse = decodePredicate.shouldDecodeResponse(apiKey, apiVersion);
         LOGGER.debug("Decode {}/v{} response? {}, Predicate {}", apiKey, apiVersion, decodeResponse, decodePredicate);
         short headerVersion = apiKey.requestHeaderVersion(apiVersion);
         if (decodeRequest) {
+            Metrics.counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES, List.of(Tag.of("flowing", "downstream"))).increment();
+            final List<Tag> LIST_OF_TAGS = List.of(
+                    Tag.of("ApiKey", apiKey.name()),
+                    Tag.of("ApiVersion", String.valueOf(apiVersion)),
+                    FLOWING_UPSTREAM);
+            Metrics.summary(KROXYLICIOUS_REQUEST_SIZE_BYTES, LIST_OF_TAGS).record(length);
             if (log().isTraceEnabled()) { // avoid boxing
                 log().trace("{}: headerVersion {}", ctx, headerVersion);
             }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaResponseDecoder.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaResponseDecoder.java
index a9b3a817..3bb7d287 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaResponseDecoder.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaResponseDecoder.java
@@ -5,6 +5,8 @@
  */
 package io.kroxylicious.proxy.internal.codec;
 
+import java.util.List;
+
 import org.apache.kafka.common.message.AddOffsetsToTxnResponseData;
 import org.apache.kafka.common.message.AddPartitionsToTxnResponseData;
 import org.apache.kafka.common.message.ApiVersionsResponseData;
@@ -42,6 +44,8 @@ import org.apache.kafka.common.protocol.Readable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Tag;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 
@@ -52,6 +56,9 @@ import io.kroxylicious.proxy.frame.OpaqueFrame;
 import io.kroxylicious.proxy.frame.OpaqueResponseFrame;
 import io.kroxylicious.proxy.internal.InternalResponseFrame;
 
+import static io.kroxylicious.proxy.internal.util.Metrics.FLOWING_DOWNSTREAM;
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_REQUEST_SIZE_BYTES;
+
 public class KafkaResponseDecoder extends KafkaMessageDecoder {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaResponseDecoder.class);
@@ -99,6 +106,11 @@ public class KafkaResponseDecoder extends KafkaMessageDecoder {
             ApiMessage body = readBody(apiKey, apiVersion, accessor);
             log().trace("{}: Body: {}", ctx, body);
             KrpcFilter recipient = correlation.recipient();
+            final List<Tag> LIST_OF_TAGS = List.of(
+                    Tag.of("ApiKey", apiKey.name()),
+                    Tag.of("ApiVersion", String.valueOf(apiVersion)),
+                    FLOWING_DOWNSTREAM);
+            Metrics.summary(KROXYLICIOUS_REQUEST_SIZE_BYTES, LIST_OF_TAGS).record(length);
             if (recipient == null) {
                 frame = new DecodedResponseFrame<>(apiVersion, correlationId, header, body);
             }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/util/Metrics.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/util/Metrics.java
new file mode 100644
index 00000000..1c8e40e5
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/util/Metrics.java
@@ -0,0 +1,27 @@
+/*
+ * 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.internal.util;
+
+import io.micrometer.core.instrument.Tag;
+
+public class Metrics {
+
+    // creating a constant for all Metrics in the one place so we can easily see what metrics there are
+
+    public static final String KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES = "kroxylicious_inbound_downstream_messages";
+
+    public static final String KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES = "kroxylicious_inbound_downstream_decoded_messages";
+
+    public static final String KROXYLICIOUS_REQUEST_SIZE_BYTES = "kroxylicious_request_size_bytes";
+
+    public static final String FLOWING_TAG = "flowing";
+
+    public static final Tag FLOWING_UPSTREAM = Tag.of(FLOWING_TAG, "upstream");
+
+    public static final Tag FLOWING_DOWNSTREAM = Tag.of(FLOWING_TAG, "downstream");
+
+}
diff --git a/krpc-code-gen/pom.xml b/krpc-code-gen/pom.xml
index 67e2a729..67045180 100644
--- a/krpc-code-gen/pom.xml
+++ b/krpc-code-gen/pom.xml
@@ -14,7 +14,7 @@
     <parent>
         <groupId>io.kroxylicious</groupId>
         <artifactId>kroxylicious-parent</artifactId>
-        <version>1.0-SNAPSHOT</version>
+        <version>0.1.0-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 
diff --git a/pom.xml b/pom.xml
index 93e92cd8..b34de4d6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -13,10 +13,11 @@
 
     <groupId>io.kroxylicious</groupId>
     <artifactId>kroxylicious-parent</artifactId>
-    <version>1.0-SNAPSHOT</version>
+    <version>0.1.0-SNAPSHOT</version>
     <packaging>pom</packaging>
 
     <properties>
+        <kroxyliciousApi.version>0.1.0-SNAPSHOT</kroxyliciousApi.version>
         <java.version>11</java.version>
         <java.test.version>19</java.test.version>
         <maven.compiler.parameters>true</maven.compiler.parameters>
@@ -31,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>
@@ -47,12 +50,13 @@
         <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>
     <description>A proof-of-concept for a Kafka wire protocol proxy</description>
     <inceptionYear>2021</inceptionYear>
-    <url>https://github.com/kroxylicious</url>
+    <url>https://kroxylicious.io/kroxylicious/</url>
 
     <licenses>
         <license>
@@ -62,6 +66,46 @@
         </license>
     </licenses>
 
+    <developers>
+        <developer>
+            <name>Tom Bentley</name>
+            <email>tbentley@redhat.com</email>
+            <organization>Red Hat</organization>
+            <organizationUrl>https://www.redhat.com</organizationUrl>
+        </developer>
+        <developer>
+            <name>Keith Wall</name>
+            <email>kwall@redhat.com</email>
+            <organization>Red Hat</organization>
+            <organizationUrl>https://www.redhat.com</organizationUrl>
+        </developer>
+        <developer>
+            <name>Sam Barker</name>
+            <email>sbarker@redhat.com</email>
+            <organization>Red Hat</organization>
+            <organizationUrl>https://www.redhat.com</organizationUrl>
+            <timezone>Pacific/Auckland</timezone>
+        </developer>
+        <developer>
+            <name>Robert Young</name>
+            <email>robeyoun@redhat.com</email>
+            <organization>Red Hat</organization>
+            <organizationUrl>https://www.redhat.com</organizationUrl>
+            <timezone>Pacific/Auckland</timezone>
+        </developer>
+        <developer>
+            <name>Francisco Vila</name>
+            <email>fvila@redhat.com</email>
+            <organization>Red Hat</organization>
+            <organizationUrl>https://www.redhat.com</organizationUrl>
+        </developer>
+    </developers>
+
+    <scm>
+        <connection>scm:git:git@github.com:kroxylicious/kroxylicious.git</connection>
+        <developerConnection>scm:git:git@github.com:kroxylicious/kroxylicious.git</developerConnection>
+        <url>https://github.com/kroxylicious/kroxylicious</url>
+    </scm>
 
     <dependencyManagement>
         <dependencies>
@@ -126,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>
@@ -168,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>
 
@@ -185,15 +241,35 @@
         <!-- Keep alphabetic order -->
         <pluginManagement>
             <plugins>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-javadoc-plugin</artifactId>
+                    <version>3.5.0</version>
+                </plugin>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-source-plugin</artifactId>
+                    <version>3.2.1</version>
+                </plugin>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-gpg-plugin</artifactId>
+                    <version>3.0.1</version>
+                </plugin>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-deploy-plugin</artifactId>
+                    <version>3.1.0</version>
+                </plugin>
                 <plugin>
                     <groupId>com.mycila</groupId>
                     <artifactId>license-maven-plugin</artifactId>
                     <version>4.1</version>
                     <configuration>
-                        <!--suppress MavenModelInspection -->
                         <strictCheck>true</strictCheck>
                         <licenseSets>
                             <licenseSet>
+                                <!--suppress MavenModelInspection -->
                                 <header>${rootdir}/etc/license.txt</header>
                                 <excludes>
                                     <exclude>LICENSE.txt</exclude>
@@ -212,7 +288,7 @@
                 <plugin>
                     <groupId>io.kroxylicious</groupId>
                     <artifactId>kroxylicious-krpc-plugin</artifactId>
-                    <version>1.0-SNAPSHOT</version>
+                    <version>0.1.0-SNAPSHOT</version>
                 </plugin>
                 <plugin>
                     <groupId>net.revelc.code.formatter</groupId>
@@ -258,11 +334,6 @@
                     <artifactId>maven-dependency-plugin</artifactId>
                     <version>3.5.0</version>
                 </plugin>
-                <plugin>
-                    <groupId>org.apache.maven.plugins</groupId>
-                    <artifactId>maven-deploy-plugin</artifactId>
-                    <version>3.1.0</version>
-                </plugin>
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-enforcer-plugin</artifactId>
@@ -308,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>
@@ -364,7 +438,6 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-javadoc-plugin</artifactId>
-                <version>3.5.0</version>
                 <executions>
                     <execution>
                         <id>attach-javadocs</id>
@@ -374,6 +447,18 @@
                     </execution>
                 </executions>
             </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-source-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>attach-sources</id>
+                        <goals>
+                            <goal>jar-no-fork</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
             <plugin>
                 <groupId>org.asciidoctor</groupId>
                 <artifactId>asciidoctor-maven-plugin</artifactId>
@@ -417,6 +502,17 @@
         </plugins>
     </build>
 
+    <distributionManagement>
+        <snapshotRepository>
+            <id>ossrh</id>
+            <url>https://s01.oss.sonatype.org/content/repositories/snapshots</url>
+        </snapshotRepository>
+        <repository>
+            <id>ossrh</id>
+            <url>https://s01.oss.sonatype.org/service/local/staging/deploy/maven2/</url>
+        </repository>
+    </distributionManagement>
+
     <profiles>
         <profile>
             <id>qa</id>
@@ -565,5 +661,55 @@
                 <skipTests>true</skipTests>
             </properties>
         </profile>
+        <profile>
+            <id>release</id>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-gpg-plugin</artifactId>
+                        <executions>
+                            <execution>
+                                <id>sign-artifacts</id>
+                                <phase>verify</phase>
+                                <goals>
+                                    <goal>sign</goal>
+                                </goals>
+                                <configuration>
+                                    <!--suppress UnresolvedMavenProperty -->
+                                    <keyname>${releaseSigningKey}</keyname>
+                                </configuration>
+                            </execution>
+                        </executions>
+                    </plugin>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-enforcer-plugin</artifactId>
+                        <executions>
+                            <execution>
+                                <id>enforce-property</id>
+                                <goals>
+                                    <goal>enforce</goal>
+                                </goals>
+                                <configuration>
+                                    <rules>
+                                        <requireProperty>
+                                            <property>releaseSigningKey</property>
+                                            <message>You must specify your GPG key when running a release</message>
+                                        </requireProperty>
+                                        <requireReleaseDeps>
+                                            <message>No Snapshots Allowed!</message>
+                                        </requireReleaseDeps>
+                                        <requireReleaseVersion>
+                                            <message>Ensure all modules have a release version</message>
+                                        </requireReleaseVersion>
+                                    </rules>
+                                </configuration>
+                            </execution>
+                        </executions>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
     </profiles>
 </project>


Test patch:
diff --git a/integrationtests/pom.xml b/integrationtests/pom.xml
index 0b660351..1bc6169b 100644
--- a/integrationtests/pom.xml
+++ b/integrationtests/pom.xml
@@ -12,7 +12,7 @@
     <parent>
         <artifactId>kroxylicious-parent</artifactId>
         <groupId>io.kroxylicious</groupId>
-        <version>1.0-SNAPSHOT</version>
+        <version>0.1.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
@@ -27,13 +27,13 @@
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious</artifactId>
-            <version>1.0-SNAPSHOT</version>
+            <version>0.1.0-SNAPSHOT</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-multitenant</artifactId>
-            <version>1.0-SNAPSHOT</version>
+            <version>0.1.0-SNAPSHOT</version>
             <scope>test</scope>
         </dependency>
         <dependency>
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