Repository: kroxylicious/kroxylicious
Pull request #136 (2023-02-16 21:37:01)

Generate javadoc jars as part of build
Also including fixes to the javadoc to enable the JDK tooling to process our docs (still emits a lot of warnings)

Why:
When we release to maven central if we push up `*-javadoc.jar` files then we should have javadoc made available for us on https://javadoc.io This would save us having to check in 8MB of generated javadoc HTML somewhere and deal with updating it, we could link to javadoc.io from our asciidoc.


Patch:
diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml
index be44d66a..957c9d4a 100644
--- a/.github/workflows/maven.yml
+++ b/.github/workflows/maven.yml
@@ -35,7 +35,7 @@ jobs:
       - name: 'Set up Java'
         uses: actions/setup-java@v2
         with:
-          java-version: 17
+          java-version: 19
           distribution: 'zulu'
 
       - name: 'Cache Maven packages'
diff --git a/README.md b/README.md
index b887a3fd..ae5bed9b 100644
--- a/README.md
+++ b/README.md
@@ -70,6 +70,13 @@ Failed to load class org.slf4j.impl.StaticLoggerBinder
 
 Make sure to follow the [suggestions here](https://www.slf4j.org/codes.html#StaticLoggerBinder) to include one (and only one) of the suggested jars on the classpath.
 
+### Debugging
+Logging is turned off by default for better performance. In case you want to debug, logging should be turned on in the `example-proxy-config.yml` file:
+```yaml
+  logNetwork: true
+  logFrames: true
+```
+
 ## Rendering documentation
 
 The `docs` directory has some user documentation written in [AsciiDoc](https://docs.asciidoctor.org/asciidoc/latest/) format.
@@ -106,7 +113,7 @@ $ dot -Tpng kroxylicious/target/generated-sources/annotations/deptective.dot > k
 
 ## IntelliJ setup
 
-Currently the project uses JDK 11 for the actual code and JDK 17 for tests. 
+Currently the project uses JDK 11 for the actual code and JDK 19 for tests. 
 IntelliJ needs to be configured to use the "new Workspace Model" in order for it to understand this
 (and to be useful for things like debugging):
 
diff --git a/benchmarking.md b/benchmarking.md
index f2dc181c..f885cd7e 100644
--- a/benchmarking.md
+++ b/benchmarking.md
@@ -7,14 +7,14 @@ This document describes how to run some basic performance tests for the proxy.
 Download Apache Kafka:
 
 ```
-$ wget https://downloads.apache.org/kafka/3.1.0/kafka_2.13-3.1.0.tgz
-$ tar xvf kafka_2.13-3.1.0.tgz
+$ wget https://downloads.apache.org/kafka/3.1.0/kafka_2.13-3.4.0.tgz
+$ tar xvf kafka_2.13-3.4.0.tgz
 ```
 
 Start up ZooKeeper and Kafka:
 
 ```
-$ cd kafka_2.13-3.1.0
+$ cd kafka_2.13-3.4.0
 $ bin/zookeeper-server-start.sh config/zookeeper.properties
 $ bin/kafka-server-start.sh config/server.properties
 ```
@@ -25,7 +25,7 @@ Build and launch the proxy:
 
 ```
 $ mvn clean verify -Pdist -Dquick
-$ java -jar target/kproxy-1.0-SNAPSHOT.jar
+$ java -jar kroxylicious/target/kroxylicious-1.0-SNAPSHOT.jar --config kroxylicious/example-proxy-config.yml
 ```
 
 Run Kafka's _kafka-producer-perf-test.sh_ script:
@@ -44,33 +44,35 @@ This produces large volumes of Kafka logs in _/tmp/kafka-logs_ which should remo
 
 ## Results
 
-Results from a benchmark run on MacMini M1.
+Results from a benchmark run on Lenovo T14s.
 
 ### Direct Kafka
 
 These values are for connecting to Kafka directly.
 
 ```
-477187 records sent, 95437,4 records/sec (93,20 MB/sec), 0,4 ms avg latency, 149,0 ms max latency.
-507616 records sent, 101523,2 records/sec (99,14 MB/sec), 0,8 ms avg latency, 54,0 ms max latency.
-506905 records sent, 101381,0 records/sec (99,00 MB/sec), 0,3 ms avg latency, 18,0 ms max latency.
-504944 records sent, 100988,8 records/sec (98,62 MB/sec), 0,7 ms avg latency, 52,0 ms max latency.
-508329 records sent, 101665,8 records/sec (99,28 MB/sec), 0,3 ms avg latency, 18,0 ms max latency.
-507461 records sent, 101492,2 records/sec (99,11 MB/sec), 0,7 ms avg latency, 55,0 ms max latency.
-508605 records sent, 101721,0 records/sec (99,34 MB/sec), 0,3 ms avg latency, 18,0 ms max latency.
-509013 records sent, 101802,6 records/sec (99,42 MB/sec), 1,0 ms avg latency, 61,0 ms max latency.
-502944 records sent, 100588,8 records/sec (98,23 MB/sec), 0,4 ms avg latency, 26,0 ms max latency.
-483576 records sent, 96715,2 records/sec (94,45 MB/sec), 1,1 ms avg latency, 58,0 ms max latency.
-503447 records sent, 100689,4 records/sec (98,33 MB/sec), 0,6 ms avg latency, 42,0 ms max latency.
-508601 records sent, 101720,2 records/sec (99,34 MB/sec), 0,6 ms avg latency, 55,0 ms max latency.
-507329 records sent, 101465,8 records/sec (99,09 MB/sec), 0,3 ms avg latency, 18,0 ms max latency.
-506374 records sent, 101274,8 records/sec (98,90 MB/sec), 1,0 ms avg latency, 56,0 ms max latency.
-504911 records sent, 100982,2 records/sec (98,62 MB/sec), 0,4 ms avg latency, 27,0 ms max latency.
-504096 records sent, 100819,2 records/sec (98,46 MB/sec), 0,7 ms avg latency, 54,0 ms max latency.
-507512 records sent, 101502,4 records/sec (99,12 MB/sec), 0,3 ms avg latency, 18,0 ms max latency.
-505010 records sent, 101002,0 records/sec (98,63 MB/sec), 0,7 ms avg latency, 53,0 ms max latency.
-504117 records sent, 100823,4 records/sec (98,46 MB/sec), 0,3 ms avg latency, 19,0 ms max latency.
-10000000 records sent, 100731,309306 records/sec (98,37 MB/sec), 0,56 ms avg latency, 149,00 ms max latency, 0 ms 50th, 1 ms 95th, 13 ms 99th, 47 ms 99.9th.
+339458 records sent, 67891.6 records/sec (66.30 MB/sec), 0.7 ms avg latency, 41.0 ms max latency.
+340690 records sent, 68138.0 records/sec (66.54 MB/sec), 0.4 ms avg latency, 8.0 ms max latency.
+328872 records sent, 65774.4 records/sec (64.23 MB/sec), 3.0 ms avg latency, 166.0 ms max latency.
+350133 records sent, 70026.6 records/sec (68.39 MB/sec), 0.4 ms avg latency, 3.0 ms max latency.
+352684 records sent, 70536.8 records/sec (68.88 MB/sec), 0.5 ms avg latency, 33.0 ms max latency.
+345508 records sent, 69101.6 records/sec (67.48 MB/sec), 1.3 ms avg latency, 63.0 ms max latency.
+347900 records sent, 69580.0 records/sec (67.95 MB/sec), 1.5 ms avg latency, 79.0 ms max latency.
+313252 records sent, 62650.4 records/sec (61.18 MB/sec), 69.4 ms avg latency, 592.0 ms max latency.
+328890 records sent, 65778.0 records/sec (64.24 MB/sec), 3.2 ms avg latency, 177.0 ms max latency.
+330865 records sent, 66173.0 records/sec (64.62 MB/sec), 0.6 ms avg latency, 19.0 ms max latency.
+326143 records sent, 65228.6 records/sec (63.70 MB/sec), 0.6 ms avg latency, 14.0 ms max latency.
+331257 records sent, 66251.4 records/sec (64.70 MB/sec), 3.6 ms avg latency, 222.0 ms max latency.
+326291 records sent, 65258.2 records/sec (63.73 MB/sec), 0.8 ms avg latency, 40.0 ms max latency.
+316958 records sent, 63391.6 records/sec (61.91 MB/sec), 0.5 ms avg latency, 13.0 ms max latency.
+342797 records sent, 68559.4 records/sec (66.95 MB/sec), 1.9 ms avg latency, 122.0 ms max latency.
+334420 records sent, 66884.0 records/sec (65.32 MB/sec), 1.9 ms avg latency, 125.0 ms max latency.
+302166 records sent, 60433.2 records/sec (59.02 MB/sec), 1.3 ms avg latency, 36.0 ms max latency.
+300626 records sent, 59802.3 records/sec (58.40 MB/sec), 1.8 ms avg latency, 95.0 ms max latency.
+320412 records sent, 64082.4 records/sec (62.58 MB/sec), 2.6 ms avg latency, 186.0 ms max latency.
+345146 records sent, 69029.2 records/sec (67.41 MB/sec), 0.6 ms avg latency, 19.0 ms max latency.
+348797 records sent, 69759.4 records/sec (68.12 MB/sec), 0.6 ms avg latency, 36.0 ms max latency.
+10000000 records sent, 68342.422876 records/sec (66.74 MB/sec), 3.92 ms avg latency, 592.00 ms max latency, 0 ms 50th, 1 ms 95th, 116 ms 99th, 534 ms 99.9th.
 ```
 
 ### Proxy - pass-through
@@ -79,27 +81,33 @@ These values show the minimal overhead of the proxy; it doesn't apply any modifi
 other than adjusting the advertised host and negotiating API versions.
 
 ```
-443845 records sent, 88769,0 records/sec (86,69 MB/sec), 0,4 ms avg latency, 142,0 ms max latency.
-487991 records sent, 97598,2 records/sec (95,31 MB/sec), 0,8 ms avg latency, 61,0 ms max latency.
-486854 records sent, 97370,8 records/sec (95,09 MB/sec), 0,3 ms avg latency, 19,0 ms max latency.
-489226 records sent, 97845,2 records/sec (95,55 MB/sec), 0,7 ms avg latency, 52,0 ms max latency.
-490963 records sent, 98192,6 records/sec (95,89 MB/sec), 0,3 ms avg latency, 19,0 ms max latency.
-490906 records sent, 98181,2 records/sec (95,88 MB/sec), 1,3 ms avg latency, 77,0 ms max latency.
-480315 records sent, 96063,0 records/sec (93,81 MB/sec), 0,3 ms avg latency, 18,0 ms max latency.
-490301 records sent, 98060,2 records/sec (95,76 MB/sec), 1,0 ms avg latency, 61,0 ms max latency.
-490337 records sent, 98067,4 records/sec (95,77 MB/sec), 0,2 ms avg latency, 17,0 ms max latency.
-486247 records sent, 97249,4 records/sec (94,97 MB/sec), 0,8 ms avg latency, 62,0 ms max latency.
-485708 records sent, 97141,6 records/sec (94,86 MB/sec), 0,3 ms avg latency, 18,0 ms max latency.
-490806 records sent, 98161,2 records/sec (95,86 MB/sec), 1,7 ms avg latency, 77,0 ms max latency.
-478836 records sent, 95767,2 records/sec (93,52 MB/sec), 0,3 ms avg latency, 18,0 ms max latency.
-490790 records sent, 98158,0 records/sec (95,86 MB/sec), 0,3 ms avg latency, 18,0 ms max latency.
-483014 records sent, 96602,8 records/sec (94,34 MB/sec), 0,7 ms avg latency, 54,0 ms max latency.
-490089 records sent, 98017,8 records/sec (95,72 MB/sec), 0,3 ms avg latency, 19,0 ms max latency.
-488382 records sent, 97676,4 records/sec (95,39 MB/sec), 1,0 ms avg latency, 74,0 ms max latency.
-490937 records sent, 98187,4 records/sec (95,89 MB/sec), 0,3 ms avg latency, 22,0 ms max latency.
-482839 records sent, 96567,8 records/sec (94,30 MB/sec), 0,7 ms avg latency, 52,0 ms max latency.
-490512 records sent, 98102,4 records/sec (95,80 MB/sec), 0,3 ms avg latency, 18,0 ms max latency.
-10000000 records sent, 97093,034546 records/sec (94,82 MB/sec), 0,61 ms avg latency, 142,00 ms max latency, 0 ms 50th, 1 ms 95th, 14 ms 99th, 58 ms 99.9th.
+323930 records sent, 64786.0 records/sec (63.27 MB/sec), 0.4 ms avg latency, 6.0 ms max latency.
+300981 records sent, 60196.2 records/sec (58.79 MB/sec), 2.3 ms avg latency, 158.0 ms max latency.
+328703 records sent, 65740.6 records/sec (64.20 MB/sec), 0.4 ms avg latency, 5.0 ms max latency.
+329130 records sent, 65826.0 records/sec (64.28 MB/sec), 0.4 ms avg latency, 6.0 ms max latency.
+309969 records sent, 61993.8 records/sec (60.54 MB/sec), 2.1 ms avg latency, 192.0 ms max latency.
+328791 records sent, 65758.2 records/sec (64.22 MB/sec), 0.4 ms avg latency, 5.0 ms max latency.
+325668 records sent, 65133.6 records/sec (63.61 MB/sec), 0.4 ms avg latency, 4.0 ms max latency.
+319086 records sent, 63817.2 records/sec (62.32 MB/sec), 3.5 ms avg latency, 198.0 ms max latency.
+316409 records sent, 63281.8 records/sec (61.80 MB/sec), 0.4 ms avg latency, 7.0 ms max latency.
+327035 records sent, 65407.0 records/sec (63.87 MB/sec), 0.4 ms avg latency, 4.0 ms max latency.
+329362 records sent, 65872.4 records/sec (64.33 MB/sec), 0.4 ms avg latency, 4.0 ms max latency.
+312351 records sent, 62470.2 records/sec (61.01 MB/sec), 7.3 ms avg latency, 269.0 ms max latency.
+331314 records sent, 66262.8 records/sec (64.71 MB/sec), 0.4 ms avg latency, 4.0 ms max latency.
+328897 records sent, 65779.4 records/sec (64.24 MB/sec), 0.4 ms avg latency, 4.0 ms max latency.
+307341 records sent, 61468.2 records/sec (60.03 MB/sec), 4.0 ms avg latency, 228.0 ms max latency.
+326419 records sent, 65283.8 records/sec (63.75 MB/sec), 0.4 ms avg latency, 7.0 ms max latency.
+329697 records sent, 65939.4 records/sec (64.39 MB/sec), 0.4 ms avg latency, 3.0 ms max latency.
+320019 records sent, 64003.8 records/sec (62.50 MB/sec), 5.0 ms avg latency, 166.0 ms max latency.
+318647 records sent, 63729.4 records/sec (62.24 MB/sec), 1.7 ms avg latency, 117.0 ms max latency.
+319843 records sent, 63968.6 records/sec (62.47 MB/sec), 0.9 ms avg latency, 35.0 ms max latency.
+287973 records sent, 57594.6 records/sec (56.24 MB/sec), 6.6 ms avg latency, 281.0 ms max latency.
+310381 records sent, 62076.2 records/sec (60.62 MB/sec), 0.7 ms avg latency, 18.0 ms max latency.
+311969 records sent, 62393.8 records/sec (60.93 MB/sec), 0.6 ms avg latency, 9.0 ms max latency.
+311976 records sent, 62395.2 records/sec (60.93 MB/sec), 0.6 ms avg latency, 9.0 ms max latency.
+291281 records sent, 58256.2 records/sec (56.89 MB/sec), 11.2 ms avg latency, 329.0 ms max latency.
+311064 records sent, 62212.8 records/sec (60.75 MB/sec), 0.6 ms avg latency, 10.0 ms max latency.
+10000000 records sent, 64342.197543 records/sec (62.83 MB/sec), 9.33 ms avg latency, 635.00 ms max latency, 0 ms 50th, 1 ms 95th, 431 ms 99th, 561 ms 99.9th.
 ```
 
 ### Proxy - Lower-case transform
@@ -116,25 +124,25 @@ new ProduceRecordTransformationFilter(
 ```
 
 ```
-446260 records sent, 89252,0 records/sec (87,16 MB/sec), 9,5 ms avg latency, 156,0 ms max latency.
-489770 records sent, 97954,0 records/sec (95,66 MB/sec), 0,4 ms avg latency, 18,0 ms max latency.
-493483 records sent, 98696,6 records/sec (96,38 MB/sec), 1,8 ms avg latency, 77,0 ms max latency.
-494918 records sent, 98983,6 records/sec (96,66 MB/sec), 0,3 ms avg latency, 18,0 ms max latency.
-494935 records sent, 98987,0 records/sec (96,67 MB/sec), 1,5 ms avg latency, 60,0 ms max latency.
-495877 records sent, 99175,4 records/sec (96,85 MB/sec), 0,7 ms avg latency, 34,0 ms max latency.
-490801 records sent, 98160,2 records/sec (95,86 MB/sec), 2,7 ms avg latency, 55,0 ms max latency.
-484580 records sent, 96916,0 records/sec (94,64 MB/sec), 1,0 ms avg latency, 39,0 ms max latency.
-478055 records sent, 95611,0 records/sec (93,37 MB/sec), 1,6 ms avg latency, 57,0 ms max latency.
-493678 records sent, 98735,6 records/sec (96,42 MB/sec), 0,4 ms avg latency, 19,0 ms max latency.
-486569 records sent, 97313,8 records/sec (95,03 MB/sec), 1,3 ms avg latency, 54,0 ms max latency.
-497780 records sent, 99556,0 records/sec (97,22 MB/sec), 0,4 ms avg latency, 18,0 ms max latency.
-492807 records sent, 98561,4 records/sec (96,25 MB/sec), 0,5 ms avg latency, 20,0 ms max latency.
-491918 records sent, 98383,6 records/sec (96,08 MB/sec), 1,9 ms avg latency, 73,0 ms max latency.
-494801 records sent, 98960,2 records/sec (96,64 MB/sec), 0,4 ms avg latency, 19,0 ms max latency.
-496531 records sent, 99306,2 records/sec (96,98 MB/sec), 1,1 ms avg latency, 54,0 ms max latency.
-495687 records sent, 99137,4 records/sec (96,81 MB/sec), 0,4 ms avg latency, 18,0 ms max latency.
-495224 records sent, 99044,8 records/sec (96,72 MB/sec), 1,3 ms avg latency, 57,0 ms max latency.
-495995 records sent, 99199,0 records/sec (96,87 MB/sec), 0,6 ms avg latency, 21,0 ms max latency.
-491785 records sent, 98357,0 records/sec (96,05 MB/sec), 1,2 ms avg latency, 54,0 ms max latency.
-10000000 records sent, 98015,192355 records/sec (95,72 MB/sec), 1,39 ms avg latency, 156,00 ms max latency, 0 ms 50th, 3 ms 95th, 40 ms 99th, 89 ms 99.9th.
+287479 records sent, 57495.8 records/sec (56.15 MB/sec), 2.8 ms avg latency, 34.0 ms max latency.
+288400 records sent, 57680.0 records/sec (56.33 MB/sec), 1.6 ms avg latency, 17.0 ms max latency.
+298933 records sent, 59786.6 records/sec (58.39 MB/sec), 2.3 ms avg latency, 164.0 ms max latency.
+254990 records sent, 50998.0 records/sec (49.80 MB/sec), 173.1 ms avg latency, 414.0 ms max latency.
+273600 records sent, 54720.0 records/sec (53.44 MB/sec), 20.4 ms avg latency, 144.0 ms max latency.
+304748 records sent, 60949.6 records/sec (59.52 MB/sec), 2.4 ms avg latency, 32.0 ms max latency.
+280629 records sent, 56125.8 records/sec (54.81 MB/sec), 18.5 ms avg latency, 264.0 ms max latency.
+306855 records sent, 61358.7 records/sec (59.92 MB/sec), 125.8 ms avg latency, 427.0 ms max latency.
+323123 records sent, 64624.6 records/sec (63.11 MB/sec), 2.4 ms avg latency, 50.0 ms max latency.
+300819 records sent, 60163.8 records/sec (58.75 MB/sec), 19.2 ms avg latency, 287.0 ms max latency.
+292274 records sent, 58454.8 records/sec (57.08 MB/sec), 4.8 ms avg latency, 58.0 ms max latency.
+301514 records sent, 60302.8 records/sec (58.89 MB/sec), 1.4 ms avg latency, 34.0 ms max latency.
+289573 records sent, 57914.6 records/sec (56.56 MB/sec), 2.8 ms avg latency, 58.0 ms max latency.
+271608 records sent, 54321.6 records/sec (53.05 MB/sec), 25.4 ms avg latency, 294.0 ms max latency.
+291544 records sent, 58308.8 records/sec (56.94 MB/sec), 1.6 ms avg latency, 18.0 ms max latency.
+284910 records sent, 56982.0 records/sec (55.65 MB/sec), 7.3 ms avg latency, 81.0 ms max latency.
+268745 records sent, 53749.0 records/sec (52.49 MB/sec), 25.7 ms avg latency, 378.0 ms max latency.
+280286 records sent, 56057.2 records/sec (54.74 MB/sec), 2.7 ms avg latency, 59.0 ms max latency.
+287880 records sent, 57576.0 records/sec (56.23 MB/sec), 1.5 ms avg latency, 16.0 ms max latency.
+10000000 records sent, 60679.611650 records/sec (59.26 MB/sec), 25.00 ms avg latency, 1080.00 ms max latency, 1 ms 50th, 170 ms 95th, 601 ms 99th, 992 ms 99.9th.
+
 ```
diff --git a/kroxylicious/example-proxy-config.yml b/kroxylicious/example-proxy-config.yml
index 01df9f14..49cb035e 100644
--- a/kroxylicious/example-proxy-config.yml
+++ b/kroxylicious/example-proxy-config.yml
@@ -7,8 +7,8 @@
 ---
 proxy:
   address: localhost:9192
-  logNetwork: true
-  logFrames: true
+  logNetwork: false
+  logFrames: false
 adminHttp:
   endpoints:
     prometheus: {}
@@ -18,6 +18,6 @@ clusters:
 filters:
 - type: ApiVersions
 - type: BrokerAddress
-- type: ProduceRequestTransformation
-  config:
-    transformation: io.kroxylicious.proxy.internal.filter.ProduceRequestTransformationFilter$UpperCasing
+#- type: ProduceRequestTransformation
+#  config:
+#    transformation: io.kroxylicious.proxy.internal.filter.ProduceRequestTransformationFilter$UpperCasing
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
index 2ebefbac..599f2eeb 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
@@ -50,7 +50,7 @@ import io.kroxylicious.proxy.internal.MeterRegistries;
 import io.kroxylicious.proxy.internal.admin.AdminHttpInitializer;
 import io.kroxylicious.proxy.internal.filter.FixedNetFilter;
 
-public final class KafkaProxy {
+public final class KafkaProxy implements AutoCloseable {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProxy.class);
 
@@ -246,4 +246,10 @@ public final class KafkaProxy {
         metricsChannel = null;
     }
 
+    @Override
+    public void close() throws Exception {
+        if (acceptorChannel != null) {
+            shutdown();
+        }
+    }
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
index abe55c56..a1bcb9af 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
@@ -28,6 +28,12 @@ public interface KrpcFilterContext {
      */
     ByteBuf allocate(int initialCapacity);
 
+    /**
+     * @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.
+     */
+    String sniHostname();
+
     /**
      * Send a request towards the broker, invoking upstream filters.
      * @param request The request to forward to the broker.
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
new file mode 100644
index 00000000..96b81c9c
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilter.java
@@ -0,0 +1,233 @@
+/*
+ * 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.util.function.Consumer;
+import java.util.function.Supplier;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData;
+import org.apache.kafka.common.message.CreateTopicsResponseData;
+import org.apache.kafka.common.message.DeleteTopicsRequestData;
+import org.apache.kafka.common.message.DeleteTopicsResponseData;
+import org.apache.kafka.common.message.FetchRequestData;
+import org.apache.kafka.common.message.FetchResponseData;
+import org.apache.kafka.common.message.ListOffsetsRequestData;
+import org.apache.kafka.common.message.ListOffsetsResponseData;
+import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.message.OffsetCommitRequestData;
+import org.apache.kafka.common.message.OffsetCommitResponseData;
+import org.apache.kafka.common.message.OffsetFetchRequestData;
+import org.apache.kafka.common.message.OffsetFetchResponseData;
+import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData;
+import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData;
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.ProduceResponseData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.kroxylicious.proxy.filter.CreateTopicsRequestFilter;
+import io.kroxylicious.proxy.filter.CreateTopicsResponseFilter;
+import io.kroxylicious.proxy.filter.DeleteTopicsRequestFilter;
+import io.kroxylicious.proxy.filter.DeleteTopicsResponseFilter;
+import io.kroxylicious.proxy.filter.FetchRequestFilter;
+import io.kroxylicious.proxy.filter.FetchResponseFilter;
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+import io.kroxylicious.proxy.filter.ListOffsetsRequestFilter;
+import io.kroxylicious.proxy.filter.ListOffsetsResponseFilter;
+import io.kroxylicious.proxy.filter.MetadataRequestFilter;
+import io.kroxylicious.proxy.filter.MetadataResponseFilter;
+import io.kroxylicious.proxy.filter.OffsetCommitRequestFilter;
+import io.kroxylicious.proxy.filter.OffsetCommitResponseFilter;
+import io.kroxylicious.proxy.filter.OffsetFetchRequestFilter;
+import io.kroxylicious.proxy.filter.OffsetFetchResponseFilter;
+import io.kroxylicious.proxy.filter.OffsetForLeaderEpochRequestFilter;
+import io.kroxylicious.proxy.filter.OffsetForLeaderEpochResponseFilter;
+import io.kroxylicious.proxy.filter.ProduceRequestFilter;
+import io.kroxylicious.proxy.filter.ProduceResponseFilter;
+import io.kroxylicious.proxy.internal.filter.FilterConfig;
+
+/**
+ * Simple multi-tenant filter.
+ *
+ * Uses the first component of a fully-qualified host name as a tenant prefix.
+ * This tenant prefix is prepended to the kafka resources name in order to present an isolated
+ * environment for each tenant.
+ *
+ * TODO prefix other resources e.g. group names, transaction ids
+ * TODO disallow the use of topic uids belonging to one tenant by another.
+ */
+public class MultiTenantTransformationFilter
+        implements CreateTopicsRequestFilter, CreateTopicsResponseFilter,
+        DeleteTopicsRequestFilter, DeleteTopicsResponseFilter,
+        MetadataRequestFilter, MetadataResponseFilter,
+        ProduceRequestFilter, ProduceResponseFilter,
+        ListOffsetsRequestFilter, ListOffsetsResponseFilter,
+        FetchRequestFilter, FetchResponseFilter,
+        OffsetFetchRequestFilter, OffsetFetchResponseFilter,
+        OffsetCommitRequestFilter, OffsetCommitResponseFilter,
+        OffsetForLeaderEpochRequestFilter, OffsetForLeaderEpochResponseFilter {
+
+    public static class MultiTenantTransformationFilterConfig extends FilterConfig {
+    }
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantTransformationFilter.class);
+
+    @Override
+    public void onCreateTopicsRequest(CreateTopicsRequestData request, KrpcFilterContext context) {
+        request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
+        context.forwardRequest(request);
+    }
+
+    @Override
+    public void onCreateTopicsResponse(CreateTopicsResponseData response, KrpcFilterContext context) {
+        response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
+        context.forwardResponse(response);
+    }
+
+    @Override
+    public void onDeleteTopicsRequest(DeleteTopicsRequestData request, KrpcFilterContext context) {
+        request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, topic.topicId() != null));
+        context.forwardRequest(request);
+    }
+
+    @Override
+    public void onDeleteTopicsResponse(DeleteTopicsResponseData response, KrpcFilterContext context) {
+        response.responses().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
+        context.forwardResponse(response);
+    }
+
+    @Override
+    public void onMetadataRequest(MetadataRequestData request, KrpcFilterContext context) {
+        if (request.topics() != null) {
+            // n.b. request.topics() == null used to query all the topics.
+            request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
+        }
+        context.forwardRequest(request);
+
+    }
+
+    @Override
+    public void onMetadataResponse(MetadataResponseData response, KrpcFilterContext context) {
+        String tenantPrefix = getTenantPrefix(context);
+        response.topics().removeIf(topic -> !topic.name().startsWith(tenantPrefix)); // TODO: allow kafka internal topics to be returned?
+        response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
+        context.forwardResponse(response);
+    }
+
+    @Override
+    public void onProduceRequest(ProduceRequestData request, KrpcFilterContext context) {
+        request.topicData().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
+        context.forwardRequest(request);
+
+    }
+
+    @Override
+    public void onProduceResponse(ProduceResponseData response, KrpcFilterContext context) {
+        response.responses().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
+        context.forwardResponse(response);
+
+    }
+
+    @Override
+    public void onListOffsetsRequest(ListOffsetsRequestData request, KrpcFilterContext context) {
+        request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
+        context.forwardRequest(request);
+    }
+
+    @Override
+    public void onListOffsetsResponse(ListOffsetsResponseData response, KrpcFilterContext context) {
+        response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
+        context.forwardResponse(response);
+    }
+
+    @Override
+    public void onOffsetFetchRequest(OffsetFetchRequestData request, KrpcFilterContext context) {
+        request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
+        request.groups().forEach(requestGroup -> requestGroup.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false)));
+        context.forwardRequest(request);
+    }
+
+    @Override
+    public void onOffsetFetchResponse(OffsetFetchResponseData response, KrpcFilterContext context) {
+        response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
+        response.groups().forEach(responseGroup -> responseGroup.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false)));
+        context.forwardResponse(response);
+    }
+
+    @Override
+    public void onOffsetForLeaderEpochRequest(OffsetForLeaderEpochRequestData request, KrpcFilterContext context) {
+        request.topics().forEach(topic -> applyTenantPrefix(context, topic::topic, topic::setTopic, false));
+        context.forwardRequest(request);
+    }
+
+    @Override
+    public void onOffsetForLeaderEpochResponse(OffsetForLeaderEpochResponseData response, KrpcFilterContext context) {
+        response.topics().forEach(topic -> removeTenantPrefix(context, topic::topic, topic::setTopic, false));
+        context.forwardResponse(response);
+    }
+
+    @Override
+    public void onOffsetCommitRequest(OffsetCommitRequestData request, KrpcFilterContext context) {
+        request.topics().forEach(topic -> applyTenantPrefix(context, topic::name, topic::setName, false));
+        context.forwardRequest(request);
+    }
+
+    @Override
+    public void onOffsetCommitResponse(OffsetCommitResponseData response, KrpcFilterContext context) {
+        response.topics().forEach(topic -> removeTenantPrefix(context, topic::name, topic::setName, false));
+        context.forwardResponse(response);
+    }
+
+    @Override
+    public void onFetchRequest(FetchRequestData request, KrpcFilterContext context) {
+        request.topics().forEach(topic -> applyTenantPrefix(context, topic::topic, topic::setTopic, topic.topicId() != null));
+        context.forwardRequest(request);
+    }
+
+    @Override
+    public void onFetchResponse(FetchResponseData response, KrpcFilterContext context) {
+        response.responses().forEach(topic -> removeTenantPrefix(context, topic::topic, topic::setTopic, topic.topicId() != null));
+        context.forwardResponse(response);
+
+    }
+
+    private void applyTenantPrefix(KrpcFilterContext context, Supplier<String> getter, Consumer<String> setter, boolean ignoreEmpty) {
+        var tenantPrefix = getTenantPrefix(context);
+        String clientSideName = getter.get();
+        if (ignoreEmpty && (clientSideName == null || clientSideName.isEmpty())) {
+            return;
+        }
+        setter.accept(tenantPrefix + "-" + clientSideName);
+    }
+
+    private void removeTenantPrefix(KrpcFilterContext context, Supplier<String> getter, Consumer<String> setter, boolean ignoreEmpty) {
+        var tenantPrefix = getTenantPrefix(context);
+        var brokerSideName = getter.get();
+
+        if (ignoreEmpty && (brokerSideName == null || brokerSideName.isEmpty())) {
+            return;
+        }
+
+        setter.accept(brokerSideName.substring(tenantPrefix.length() + 1));
+    }
+
+    private static String getTenantPrefix(KrpcFilterContext context) {
+        // TODO naive - POC implementation uses the first component of a FQDN as the multi-tenant prefix.
+        var sniHostname = context.sniHostname();
+        if (sniHostname == null) {
+            throw new IllegalStateException("This filter requires that the client provides a TLS SNI hostname.");
+        }
+        int dot = sniHostname.indexOf(".");
+        if (dot < 1) {
+            throw new IllegalStateException("Unexpected SNI hostname formation. SNI hostname : " + sniHostname);
+        }
+        return sniHostname.substring(0, dot);
+    }
+
+    public MultiTenantTransformationFilter() {
+    }
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/package-info.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/package-info.java
index 16dee14f..447c5385 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/package-info.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/package-info.java
@@ -16,18 +16,18 @@
  * <p>filter chain is instantiated by a {@link io.kroxylicious.proxy.bootstrap.FilterChainFactory}
  * when a connection is made by a client.</p>
  *
- * <h3 id='assumptions'>Important facts about the Kafka protocol</h3>
+ * <h2 id='assumptions'>Important facts about the Kafka protocol</h2>
  *
- * <h4 id='pipelining'>Pipelining</h4>
+ * <h3 id='pipelining'>Pipelining</h3>
  * <p>The Kafka protocol supports pipelining (meaning a client can send multiple requests,
  * before getting a response for any of them). Therefore when writing a filter implementation
  * do not assume you won't see multiple requests before seeing any corresponding responses.</p>
  *
- * <h4 id='ordering'>Ordering</h4>
+ * <h3 id='ordering'>Ordering</h3>
  * <p>A broker does not, in general, send responses in the same order as it receives requests.
  * Therefore when writing a filter implementation do not assume ordering.</p>
  *
- * <h4 id='local_view'>Local view</h4>
+ * <h3 id='local_view'>Local view</h3>
  * <p>A client may obtain information from one broker in a cluster and use it to interact with other
  * brokers in the cluster (or the same broker, but on a different connection, and therefore a different
  * channel and filter chain). A classic example would
@@ -36,7 +36,7 @@
  *
  * <p>So although your filter
  * <em>implementation</em> might intercept both {@code Metadata} and {@code Produce} request/response
- * (for example), those requests will not pass through the same <emp>instance</emp> of your filter
+ * (for example), those requests will not pass through the same <em>instance</em> of your filter
  * implementation. Therefore it is incorrect, in general, to assume your filter has a global view of
  * the communication between the client and broker.</p>
  *
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/Frame.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/Frame.java
index 6b92ac2d..0fd0a894 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/Frame.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/Frame.java
@@ -5,8 +5,6 @@
  */
 package io.kroxylicious.proxy.frame;
 
-import io.netty.buffer.ByteBuf;
-
 /**
  * A frame in the Kafka protocol, which may or may not be fully decoded.
  */
@@ -14,7 +12,7 @@ public interface Frame {
 
     /**
      * Estimate the expected encoded size in bytes of this {@code Frame}.<br>
-     * In particular, written data by {@link #encode(ByteBuf)} should be the same as reported by this method.
+     * In particular, written data by {@link #encode(ByteBufAccessor)} should be the same as reported by this method.
      */
     int estimateEncodedSize();
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/future/CompositeFuture.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/future/CompositeFuture.java
index 318686b6..e231a2f6 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/future/CompositeFuture.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/future/CompositeFuture.java
@@ -26,9 +26,8 @@ import io.kroxylicious.proxy.internal.future.CompositeFutureImpl;
 public interface CompositeFuture extends Future<CompositeFuture> {
 
     /**
-     * Return a composite future, succeeded when all futures are succeeded, failed when any future is failed.
-     * <p/>
-     * The returned future fails as soon as one of {@code f1} or {@code f2} fails.
+     * <p>Return a composite future, succeeded when all futures are succeeded, failed when any future is failed.</p>
+     * <p>The returned future fails as soon as one of {@code f1} or {@code f2} fails.</p>
      *
      * @param f1 future
      * @param f2 future
@@ -76,9 +75,8 @@ public interface CompositeFuture extends Future<CompositeFuture> {
     }
 
     /**
-     * Return a composite future, succeeded when any futures is succeeded, failed when all futures are failed.
-     * <p/>
-     * The returned future succeeds as soon as one of {@code f1} or {@code f2} succeeds.
+     * <p>Return a composite future, succeeded when any futures is succeeded, failed when all futures are failed.</p>
+     * <p>The returned future succeeds as soon as one of {@code f1} or {@code f2} succeeds.</p>
      *
      * @param f1 future
      * @param f2 future
@@ -126,9 +124,8 @@ public interface CompositeFuture extends Future<CompositeFuture> {
     }
 
     /**
-     * Return a composite future, succeeded when all futures are succeeded, failed when any future is failed.
-     * <p/>
-     * It always wait until all its futures are completed and will not fail as soon as one of {@code f1} or {@code f2} fails.
+     * <p>Return a composite future, succeeded when all futures are succeeded, failed when any future is failed.</p>
+     * <p>It always wait until all its futures are completed and will not fail as soon as one of {@code f1} or {@code f2} fails.</p>
      *
      * @param f1 future
      * @param f2 future
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/future/Promise.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/future/Promise.java
index d0175a89..24610302 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/future/Promise.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/future/Promise.java
@@ -51,9 +51,8 @@ public interface Promise<T> extends Handler<AsyncResult<T>> {
     }
 
     /**
-     * Set the result. Any handler will be called, if there is one, and the promise will be marked as completed.
-     * <p/>
-     * Any handler set on the associated promise will be called.
+     * <p>Set the result. Any handler will be called, if there is one, and the promise will be marked as completed.</p>
+     * <p>Any handler set on the associated promise will be called.</p>
      *
      * @param result  the result
      * @throws IllegalStateException when the promise is already completed
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
index 97064509..6c8bb310 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
@@ -37,17 +37,20 @@ class DefaultFilterContext implements KrpcFilterContext {
     private final ChannelPromise promise;
     private final KrpcFilter filter;
     private final long timeoutMs;
+    private final String sniHostname;
 
     DefaultFilterContext(KrpcFilter filter,
                          ChannelHandlerContext channelContext,
                          DecodedFrame<?, ?> decodedFrame,
                          ChannelPromise promise,
-                         long timeoutMs) {
+                         long timeoutMs,
+                         String sniHostname) {
         this.filter = filter;
         this.channelContext = channelContext;
         this.decodedFrame = decodedFrame;
         this.promise = promise;
         this.timeoutMs = timeoutMs;
+        this.sniHostname = sniHostname;
     }
 
     /**
@@ -70,9 +73,15 @@ class DefaultFilterContext implements KrpcFilterContext {
     public ByteBuf allocate(int initialCapacity) {
         final ByteBuf buffer = channelContext.alloc().heapBuffer(initialCapacity);
         decodedFrame.add(buffer);
+
         return buffer;
     }
 
+    @Override
+    public String sniHostname() {
+        return sniHostname;
+    }
+
     /**
      * Forward a request to the next filter in the chain
      * (or to the upstream broker).
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
index c633fa0b..8a5cd9a1 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
@@ -34,10 +34,12 @@ public class FilterHandler
     private static final Logger LOGGER = LoggerFactory.getLogger(FilterHandler.class);
     private final KrpcFilter filter;
     private final long timeoutMs;
+    private final String sniHostname;
 
-    public FilterHandler(KrpcFilter filter, long timeoutMs) {
+    public FilterHandler(KrpcFilter filter, long timeoutMs, String sniHostname) {
         this.filter = Objects.requireNonNull(filter);
         this.timeoutMs = Assertions.requireStrictlyPositive(timeoutMs, "timeout");
+        this.sniHostname = sniHostname;
     }
 
     String filterDescriptor() {
@@ -50,7 +52,7 @@ public class FilterHandler
             DecodedRequestFrame<?> decodedFrame = (DecodedRequestFrame<?>) msg;
             // Guard against invoking the filter unexpectedly
             if (filter.shouldDeserializeRequest(decodedFrame.apiKey(), decodedFrame.apiVersion())) {
-                var filterContext = new DefaultFilterContext(filter, ctx, decodedFrame, promise, timeoutMs);
+                var filterContext = new DefaultFilterContext(filter, ctx, decodedFrame, promise, timeoutMs, sniHostname);
                 if (LOGGER.isDebugEnabled()) {
                     LOGGER.debug("{}: Dispatching downstream {} request to filter{}: {}",
                             ctx.channel(), decodedFrame.apiKey(), filterDescriptor(), msg);
@@ -95,7 +97,7 @@ public class FilterHandler
                 }
             }
             else if (filter.shouldDeserializeResponse(decodedFrame.apiKey(), decodedFrame.apiVersion())) {
-                var filterContext = new DefaultFilterContext(filter, ctx, decodedFrame, null, timeoutMs);
+                var filterContext = new DefaultFilterContext(filter, ctx, decodedFrame, null, timeoutMs, sniHostname);
                 if (LOGGER.isDebugEnabled()) {
                     LOGGER.debug("{}: Dispatching upstream {} response to filter {}: {}",
                             ctx.channel(), decodedFrame.apiKey(), filterDescriptor(), msg);
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaAuthnHandler.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaAuthnHandler.java
index cbd89371..5d0622a4 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaAuthnHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaAuthnHandler.java
@@ -195,19 +195,19 @@ import io.kroxylicious.proxy.tag.VisibleForTesting;
  * an {@link AuthenticationEvent} to upstream handlers, specifically {@link KafkaProxyFrontendHandler}, to use in
  * deciding how the connection to an upstream connection should be made.</p>
  *
- * @see "<a href="https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=51809888">KIP-12: Kafka Sasl/Kerberos and SSL implementation</a>
+ * @see <a href="https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=51809888">KIP-12: Kafka Sasl/Kerberos and SSL implementation</a>
  * added support for Kerberos authentication"
- * @see "<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-43%3A+Kafka+SASL+enhancements">KIP-43: Kafka SASL enhancements</a>
+ * @see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-43%3A+Kafka+SASL+enhancements">KIP-43: Kafka SASL enhancements</a>
  * added the SaslHandshake RPC in Kafka 0.10.0.0"
- * @see "<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-84%3A+Support+SASL+SCRAM+mechanisms">KIP-84: Support SASL SCRAM mechanisms</a>
+ * @see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-84%3A+Support+SASL+SCRAM+mechanisms">KIP-84: Support SASL SCRAM mechanisms</a>
  * added support for the SCRAM-SHA-256 and SCRAM-SHA-512 mechanisms"
- * @see "<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-152+-+Improve+diagnostics+for+SASL+authentication+failures">KIP-152: Improve diagnostics for SASL authentication failures</a>
+ * @see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-152+-+Improve+diagnostics+for+SASL+authentication+failures">KIP-152: Improve diagnostics for SASL authentication failures</a>
  * added support for the SaslAuthenticate RPC (previously the auth bytes were not encapsulated in a Kafka frame"
- * @see "<a href="https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=75968876">KIP-255: OAuth Authentication via SASL/OAUTHBEARER</a>
+ * @see <a href="https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=75968876">KIP-255: OAuth Authentication via SASL/OAUTHBEARER</a>
  * added support for OAUTH authentication"
- * @see "<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-368%3A+Allow+SASL+Connections+to+Periodically+Re-Authenticate">KIP-365: Allow SASL Connections to Periodically Re-Authenticate</a>
+ * @see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-368%3A+Allow+SASL+Connections+to+Periodically+Re-Authenticate">KIP-365: Allow SASL Connections to Periodically Re-Authenticate</a>
  * added time-based reauthentication requirements for clients"
- * @see "<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-684+-+Support+mutual+TLS+authentication+on+SASL_SSL+listeners">KIP-684: Support mTLS authentication on SASL_SSL listeners</a>
+ * @see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-684+-+Support+mutual+TLS+authentication+on+SASL_SSL+listeners">KIP-684: Support mTLS authentication on SASL_SSL listeners</a>
  * added support for mutual TLS authentication even on SASL_SSL listeners (which was previously ignored)"
  */
 public class KafkaAuthnHandler extends ChannelInboundHandlerAdapter {
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
index c8599380..294b0b32 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
@@ -282,7 +282,7 @@ public class KafkaProxyFrontendHandler
     private void addFiltersToPipeline(KrpcFilter[] filters, ChannelPipeline pipeline) {
         for (var filter : filters) {
             // TODO configurable timeout
-            pipeline.addFirst(filter.toString(), new FilterHandler(filter, 20000));
+            pipeline.addFirst(filter.toString(), new FilterHandler(filter, 20000, sniHostname));
         }
     }
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
index 6efed0c5..b7dc07b1 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
@@ -18,7 +18,9 @@ import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.haproxy.HAProxyMessageDecoder;
 import io.netty.handler.logging.LogLevel;
 import io.netty.handler.logging.LoggingHandler;
+import io.netty.handler.ssl.SniHandler;
 import io.netty.handler.ssl.SslContext;
+import io.netty.util.DomainWildcardMappingBuilder;
 
 import io.kroxylicious.proxy.filter.NetFilter;
 import io.kroxylicious.proxy.internal.codec.KafkaRequestDecoder;
@@ -57,8 +59,8 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
         ChannelPipeline pipeline = ch.pipeline();
 
         sslContext.ifPresent(s -> {
-            LOGGER.debug("Adding SSL handler");
-            pipeline.addLast(s.newHandler(ch.alloc()));
+            LOGGER.debug("Adding SSL/SNI handler");
+            pipeline.addLast(new SniHandler(new DomainWildcardMappingBuilder<>(s).build()));
         });
 
         if (logNetwork) {
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/AddressMapping.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/AddressMapping.java
new file mode 100644
index 00000000..bec7c0f6
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/AddressMapping.java
@@ -0,0 +1,15 @@
+/*
+ * 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.filter;
+
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+
+public interface AddressMapping {
+    String downstreamHost(KrpcFilterContext context, String upstreamHost, int upstreamPort);
+
+    int downstreamPort(KrpcFilterContext context, String upstreamHost, int upstreamPort);
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java
index d50eecbc..222930a9 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java
@@ -5,6 +5,7 @@
  */
 package io.kroxylicious.proxy.internal.filter;
 
+import java.lang.reflect.InvocationTargetException;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
 import java.util.function.ObjIntConsumer;
@@ -33,42 +34,35 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
     private static final Logger LOGGER = LoggerFactory.getLogger(BrokerAddressFilter.class);
 
     public static class BrokerAddressFilterConfig extends FilterConfig {
-    }
-
-    public interface AddressMapping {
-        String downstreamHost(String upstreamHost, int upstreamPort);
-
-        int downstreamPort(String upstreamHost, int upstreamPort);
-    }
-
-    private static class FixedAddressMapping implements AddressMapping {
 
-        private final String targetHost;
-        private final int targetPort;
+        private final Class<? extends AddressMapping> addressMapperClazz;
 
-        public FixedAddressMapping(String targetHost, int targetPort) {
-            this.targetHost = targetHost;
-            this.targetPort = targetPort;
+        public BrokerAddressFilterConfig(String addressMapper) {
+            try {
+                this.addressMapperClazz = addressMapper == null ? null : (Class<? extends AddressMapping>) Class.forName(addressMapper);
+                ;
+            }
+            catch (ClassNotFoundException e) {
+                throw new RuntimeException(e);
+            }
         }
 
-        @Override
-        public String downstreamHost(String host, int port) {
-            return targetHost;
-        }
-
-        @Override
-        public int downstreamPort(String host, int port) {
-            return targetPort;
+        public Class<? extends AddressMapping> addressMapper() {
+            return addressMapperClazz;
         }
     }
 
     private final AddressMapping mapping;
 
-    public BrokerAddressFilter(ProxyConfig config) {
-        String proxyAddress = config.address();
-        String[] proxyAddressParts = proxyAddress.split(":");
+    public BrokerAddressFilter(ProxyConfig all, BrokerAddressFilterConfig config) {
 
-        this.mapping = new FixedAddressMapping(proxyAddressParts[0], Integer.valueOf(proxyAddressParts[1]));
+        try {
+            this.mapping = config == null || config.addressMapperClazz == null ? new FixedAddressMapping(all)
+                    : config.addressMapper().getDeclaredConstructor(ProxyConfig.class).newInstance(all);
+        }
+        catch (InstantiationException | IllegalAccessException | InvocationTargetException | NoSuchMethodException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     @Override
@@ -100,8 +94,8 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
         String incomingHost = hostGetter.apply(broker);
         int incomingPort = portGetter.applyAsInt(broker);
 
-        String host = mapping.downstreamHost(incomingHost, incomingPort);
-        int port = mapping.downstreamPort(incomingHost, incomingPort);
+        String host = mapping.downstreamHost(context, incomingHost, incomingPort);
+        int port = mapping.downstreamPort(context, incomingHost, incomingPort);
 
         LOGGER.trace("{}: Rewriting broker address in response {}:{} -> {}:{}", context, incomingHost, incomingPort, host, port);
         hostSetter.accept(broker, host);
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BuiltinFilterContributor.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BuiltinFilterContributor.java
index f5cbf3e0..9fa25740 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BuiltinFilterContributor.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/BuiltinFilterContributor.java
@@ -8,6 +8,8 @@ package io.kroxylicious.proxy.internal.filter;
 import io.kroxylicious.proxy.config.ProxyConfig;
 import io.kroxylicious.proxy.filter.FilterContributor;
 import io.kroxylicious.proxy.filter.KrpcFilter;
+import io.kroxylicious.proxy.filter.multitenant.MultiTenantTransformationFilter;
+import io.kroxylicious.proxy.filter.multitenant.MultiTenantTransformationFilter.MultiTenantTransformationFilterConfig;
 import io.kroxylicious.proxy.internal.filter.ApiVersionsFilter.ApiVersionsFilterConfig;
 import io.kroxylicious.proxy.internal.filter.BrokerAddressFilter.BrokerAddressFilterConfig;
 import io.kroxylicious.proxy.internal.filter.FetchResponseTransformationFilter.FetchResponseTransformationFilterConfig;
@@ -26,6 +28,8 @@ public class BuiltinFilterContributor implements FilterContributor {
                 return ProduceRequestTransformationFilterConfig.class;
             case "FetchResponseTransformation":
                 return FetchResponseTransformationFilterConfig.class;
+            case "MultiTenant":
+                return MultiTenantTransformationFilterConfig.class;
             default:
                 return null;
         }
@@ -37,11 +41,13 @@ public class BuiltinFilterContributor implements FilterContributor {
             case "ApiVersions":
                 return new ApiVersionsFilter();
             case "BrokerAddress":
-                return new BrokerAddressFilter(proxyConfig);
+                return new BrokerAddressFilter(proxyConfig, ((BrokerAddressFilterConfig) filterConfig));
             case "ProduceRequestTransformation":
                 return new ProduceRequestTransformationFilter((ProduceRequestTransformationFilterConfig) filterConfig);
             case "FetchResponseTransformation":
                 return new FetchResponseTransformationFilter((FetchResponseTransformationFilterConfig) filterConfig);
+            case "MultiTenant":
+                return new MultiTenantTransformationFilter();
             default:
                 return null;
         }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FixedAddressMapping.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FixedAddressMapping.java
new file mode 100644
index 00000000..87857b0e
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FixedAddressMapping.java
@@ -0,0 +1,34 @@
+/*
+ * 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.filter;
+
+import io.kroxylicious.proxy.config.ProxyConfig;
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+
+class FixedAddressMapping implements AddressMapping {
+
+    private final String targetHost;
+    private final int targetPort;
+
+    public FixedAddressMapping(ProxyConfig config) {
+        String proxyAddress = config.address();
+        String[] proxyAddressParts = proxyAddress.split(":");
+
+        this.targetHost = proxyAddressParts[0];
+        this.targetPort = Integer.parseInt(proxyAddressParts[1]);
+    }
+
+    @Override
+    public String downstreamHost(KrpcFilterContext context, String host, int port) {
+        return targetHost;
+    }
+
+    @Override
+    public int downstreamPort(KrpcFilterContext context, String host, int port) {
+        return targetPort;
+    }
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/SniAddressMapping.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/SniAddressMapping.java
new file mode 100644
index 00000000..fd0a39a9
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/SniAddressMapping.java
@@ -0,0 +1,34 @@
+/*
+ * 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.filter;
+
+import io.kroxylicious.proxy.config.ProxyConfig;
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+
+class SniAddressMapping implements AddressMapping {
+
+    private final String targetHost;
+    private final int targetPort;
+
+    public SniAddressMapping(ProxyConfig config) {
+        String proxyAddress = config.address();
+        String[] proxyAddressParts = proxyAddress.split(":");
+
+        this.targetHost = proxyAddressParts[0];
+        this.targetPort = Integer.parseInt(proxyAddressParts[1]);
+    }
+
+    @Override
+    public String downstreamHost(KrpcFilterContext context, String host, int port) {
+        return context.sniHostname() != null ? context.sniHostname() : targetHost;
+    }
+
+    @Override
+    public int downstreamPort(KrpcFilterContext context, String host, int port) {
+        return targetPort;
+    }
+}
diff --git a/kroxylicious/src/main/templates/Kproxy/Filter.ftl b/kroxylicious/src/main/templates/Kproxy/Filter.ftl
index 27d6f869..9f867916 100644
--- a/kroxylicious/src/main/templates/Kproxy/Filter.ftl
+++ b/kroxylicious/src/main/templates/Kproxy/Filter.ftl
@@ -44,8 +44,6 @@ public interface ${filterClass} extends KrpcFilter {
      *
      * @param ${msgType} The KRPC message to handle.
      * @param context The context.
-     * @return the {@code ${msgType}} to be passed to the next filter.
-     * If null is returned then the given {@code ${msgType}} will be used.
      */
     public void on${messageSpec.name}(${dataClass} ${msgType}, KrpcFilterContext context);
 
diff --git a/kroxylicious/src/main/templates/Kproxy/KrpcFilter.ftl b/kroxylicious/src/main/templates/Kproxy/KrpcFilter.ftl
index 318b5e89..8b1b5e49 100644
--- a/kroxylicious/src/main/templates/Kproxy/KrpcFilter.ftl
+++ b/kroxylicious/src/main/templates/Kproxy/KrpcFilter.ftl
@@ -41,12 +41,12 @@ import io.kroxylicious.proxy.frame.DecodedResponseFrame;
  * the {@code on*Request} method(s), unless your filter can avoid deserialization in which case
  * you can override {@link #shouldDeserializeRequest(ApiKeys, short)} as well.</p>
  *
- * <h3>Guarantees</h3>
+ * <h2>Guarantees</h2>
  * <p>Implementors of this API may assume the following:</p>
  * <ol>
  *     <li>That each instance of the filter is associated with a single channel</li>
  *     <li>That {@link #shouldDeserializeRequest(ApiKeys, short)} and
- *     {@link #apply(DecodedRequestFrame, KrpcFilterContext)} (or {@code on*Request} as appropriate)
+ *     {@link #onRequest(DecodedRequestFrame, KrpcFilterContext)} (or {@code on*Request} as appropriate)
  *     will always be invoked on the same thread.</li>
  *     <li>That filters are applied in the order they were configured.</li>
  * </ol>
@@ -85,7 +85,6 @@ public /* sealed */ interface KrpcFilter /* TODO permits ... */ {
      * Apply the filter to the given {@code decodedFrame} using the given {@code filterContext}.
      * @param decodedFrame The response frame.
      * @param filterContext The filter context.
-     * @return The state of the filter.
      */
     public default void onResponse(DecodedResponseFrame<?> decodedFrame,
                                    KrpcFilterContext filterContext) {
@@ -112,8 +111,8 @@ public /* sealed */ interface KrpcFilter /* TODO permits ... */ {
      *     <li>{@code shouldDeserializeRequest} on request A</li>
      *     <li>{@code shouldDeserializeRequest} on request B</li>
      *     <li>{@code shouldDeserializeRequest} on request A</li>
-     *     <li>{@code apply} on request A</li>
-     *     <li>{@code apply} on request B</li>
+     *     <li>{@code onRequest} on request A</li>
+     *     <li>{@code onRequest} on request B</li>
      * </ol>
      * @param apiKey The API key
      * @param apiVersion The API version
diff --git a/pom.xml b/pom.xml
index 0170b899..bc4e0adc 100644
--- a/pom.xml
+++ b/pom.xml
@@ -18,7 +18,7 @@
 
     <properties>
         <java.version>11</java.version>
-        <java.test.version>17</java.test.version>
+        <java.test.version>19</java.test.version>
         <maven.compiler.parameters>true</maven.compiler.parameters>
         <maven.compiler.release>${java.version}</maven.compiler.release>
         <maven.compiler.testRelease>${java.test.version}</maven.compiler.testRelease>
@@ -41,7 +41,7 @@
         <netty.epoll.classifier>linux-x86_64</netty.epoll.classifier>
         <netty.io_uring.classifier>linux-x86_64</netty.io_uring.classifier>
         <netty.kqueue.classifier>osx-x86_64</netty.kqueue.classifier>
-        <lombok.version>1.18.24</lombok.version>
+        <lombok.version>1.18.26</lombok.version>
         <assertj-core.version>3.22.0</assertj-core.version>
         <mockito-core.version>4.6.1</mockito-core.version>
         <slf4j-api.version>1.7.36</slf4j-api.version>
@@ -210,6 +210,7 @@
         <module>integrationtests</module>
     </modules>
 
+
     <build>
         <!-- Keep alphabetic order -->
         <pluginManagement>
@@ -248,12 +249,13 @@
                 <plugin>
                     <groupId>net.revelc.code</groupId>
                     <artifactId>impsort-maven-plugin</artifactId>
-                    <version>1.7.0</version>
+                    <version>1.8.0</version>
                     <configuration>
                         <groups>java.,javax.,org.,com.,io.,io.kroxylicious</groups>
                         <removeUnused>true</removeUnused>
                         <staticAfter>true</staticAfter>
-                        <compliance>${java.test.version}</compliance>
+                        <!-- use langauge level 17, see https://github.com/javaparser/javaparser/issues/3833 -->
+                        <compliance>17</compliance>
                     </configuration>
                 </plugin>
                 <plugin>
@@ -305,7 +307,7 @@
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-plugin-plugin</artifactId>
-                    <version>3.6.1</version>
+                    <version>3.7.0</version>
                 </plugin>
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
@@ -377,6 +379,19 @@
             </plugins>
         </pluginManagement>
         <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-javadoc-plugin</artifactId>
+                <version>3.5.0</version>
+                <executions>
+                    <execution>
+                        <id>attach-javadocs</id>
+                        <goals>
+                            <goal>jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
             <plugin>
                 <groupId>org.asciidoctor</groupId>
                 <artifactId>asciidoctor-maven-plugin</artifactId>


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/net/MultiTenantSniInetAddressResolverProvider.java b/integrationtests/src/test/java/io/kroxylicious/net/MultiTenantSniInetAddressResolverProvider.java
new file mode 100644
index 00000000..97dd5c29
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/net/MultiTenantSniInetAddressResolverProvider.java
@@ -0,0 +1,59 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.net;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.net.spi.InetAddressResolver;
+import java.net.spi.InetAddressResolverProvider;
+import java.util.stream.Stream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MultiTenantSniInetAddressResolverProvider extends InetAddressResolverProvider {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantSniInetAddressResolverProvider.class);
+
+    public MultiTenantSniInetAddressResolverProvider() {
+        LOGGER.info("Creating MultiTenantSniInetAddressResolverProvider");
+    }
+
+    @Override
+    public InetAddressResolver get(Configuration configuration) {
+        return new InetAddressResolver() {
+            @Override
+            public Stream<InetAddress> lookupByName(String host, LookupPolicy lookupPolicy) throws UnknownHostException {
+                if (host != null && host.endsWith(".multitenant.kafka")) {
+                    Stream<InetAddress> inetAddressStream = configuration.builtinResolver().lookupByName("localhost", lookupPolicy);
+                    return inetAddressStream.map(in -> {
+                        try {
+                            return InetAddress.getByAddress(host, in.getAddress());
+                        }
+                        catch (UnknownHostException e) {
+                            // Should never happen
+                            throw new RuntimeException(e);
+                        }
+                    }).toList().stream();
+                }
+                else {
+                    return configuration.builtinResolver().lookupByName(host, lookupPolicy);
+                }
+            }
+
+            @Override
+            public String lookupByAddress(byte[] addr) throws UnknownHostException {
+                return configuration.builtinResolver().lookupByAddress(addr);
+            }
+        };
+    }
+
+    @Override
+    public String name() {
+        return MultiTenantSniInetAddressResolverProvider.class.getSimpleName();
+    }
+}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java
index 06f4dd9e..8f7f79ed 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java
@@ -44,12 +44,14 @@ public class MetricsIT {
         this.testInfo = testInfo;
         Metrics.globalRegistry.clear();
     }
+
     @AfterEach
     public void teardown() {
-        if(proxy != null){
+        if (proxy != null) {
             try {
                 proxy.shutdown();
-            } catch (InterruptedException e) {
+            }
+            catch (InterruptedException e) {
                 throw new RuntimeException(e);
             }
         }
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
new file mode 100644
index 00000000..5962bf76
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
@@ -0,0 +1,368 @@
+/*
+ * 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.multitenant;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.DeleteTopicsResult;
+import org.apache.kafka.clients.admin.ListTopicsResult;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.clients.admin.TopicListing;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetResetStrategy;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.TopicCollection;
+import org.apache.kafka.common.TopicCollection.TopicNameCollection;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.assertj.core.api.Condition;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.kroxylicious.proxy.KafkaProxy;
+import io.kroxylicious.proxy.config.ConfigParser;
+import io.kroxylicious.proxy.config.Configuration;
+import io.kroxylicious.proxy.testkafkacluster.KafkaCluster;
+import io.kroxylicious.proxy.testkafkacluster.KafkaClusterConfig;
+import io.kroxylicious.proxy.testkafkacluster.KafkaClusterFactory;
+import io.kroxylicious.proxy.testkafkacluster.KeytoolCertificateGenerator;
+
+import static org.assertj.core.api.Assertions.allOf;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+public class MultiTenantIT {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantIT.class);
+
+    private static final String TOPIC_1 = "my-test-topic";
+    private static final NewTopic NEW_TOPIC_1 = new NewTopic(TOPIC_1, 1, (short) 1);
+    private static final String TOPIC_2 = "other-test-topic";
+    private static final NewTopic NEW_TOPIC_2 = new NewTopic(TOPIC_2, 1, (short) 1);
+
+    private static final String TOPIC_3 = "and-another-test-topic";
+    private static final NewTopic NEW_TOPIC_3 = new NewTopic(TOPIC_3, 1, (short) 1);
+
+    private static final String PROXY_ADDRESS = "localhost:9192";
+    private static final String TENANT1_PROXY_ADDRESS = "foo.multitenant.kafka:9192";
+    private static final String TENANT2_PROXY_ADDRESS = "bar.multitenant.kafka:9192";
+    private static final String MY_KEY = "my-key";
+    private static final String MY_VALUE = "my-value";
+    private TestInfo testInfo;
+    private KeytoolCertificateGenerator certificateGenerator;
+
+    @BeforeEach
+    public void beforeEach(TestInfo testInfo) throws Exception {
+        this.testInfo = testInfo;
+        this.certificateGenerator = new KeytoolCertificateGenerator();
+        this.certificateGenerator.generateSelfSignedCertificateEntry("test@redhat.com", "*.multitenant.kafka", "KI", "RedHat", null, null, "US");
+    }
+
+    @Test
+    public void createAndDeleteTopic() throws Exception {
+        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
+            cluster.start();
+
+            String config = getConfig(PROXY_ADDRESS, cluster);
+
+            try (var proxy = startProxy(config)) {
+                try (var admin = Admin.create(commonConfig(TENANT1_PROXY_ADDRESS, Map.of()))) {
+                    var created = createTopics(admin, List.of(NEW_TOPIC_1, NEW_TOPIC_2));
+
+                    ListTopicsResult listTopicsResult = admin.listTopics();
+                    var topicMap = listTopicsResult.namesToListings().get();
+                    assertEquals(2, topicMap.size());
+                    assertThat(topicMap).hasSize(2);
+                    assertThat(topicMap).hasEntrySatisfying(TOPIC_1,
+                            allOf(matches(TopicListing::name, TOPIC_1),
+                                    matches(TopicListing::topicId, created.topicId(TOPIC_1).get())));
+                    assertThat(topicMap).hasEntrySatisfying(TOPIC_1,
+                            allOf(matches(TopicListing::name, TOPIC_1),
+                                    matches(TopicListing::topicId, created.topicId(TOPIC_1).get())));
+                    assertThat(topicMap).hasEntrySatisfying(TOPIC_2, matches(TopicListing::name, TOPIC_2));
+
+                    // Delete by name
+                    var topics1 = TopicNameCollection.ofTopicNames(List.of(TOPIC_1));
+                    var deleted = deleteTopics(admin, topics1);
+                    assertThat(deleted.topicNameValues().keySet()).containsAll(topics1.topicNames());
+
+                    // Delete by id
+                    var topics2 = TopicCollection.ofTopicIds(List.of(created.topicId(TOPIC_2).get()));
+                    deleted = deleteTopics(admin, topics2);
+                    assertThat(deleted.topicIdValues().keySet()).containsAll(topics2.topicIds());
+                }
+            }
+        }
+    }
+
+    @Test
+    public void describeTopic() throws Exception {
+        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
+            cluster.start();
+
+            String config = getConfig(PROXY_ADDRESS, cluster);
+
+            try (var proxy = startProxy(config)) {
+                try (var admin = Admin.create(commonConfig(TENANT1_PROXY_ADDRESS, Map.of()))) {
+                    var created = createTopics(admin, List.of(NEW_TOPIC_1));
+
+                    var describeTopicsResult = admin.describeTopics(TopicNameCollection.ofTopicNames(List.of(TOPIC_1)));
+                    var topicMap = describeTopicsResult.allTopicNames().get();
+                    assertThat(topicMap).hasEntrySatisfying(TOPIC_1,
+                            allOf(matches(TopicDescription::name, TOPIC_1),
+                                    matches(TopicDescription::topicId, created.topicId(TOPIC_1).get())));
+                }
+            }
+        }
+    }
+
+    @Test
+    public void publishOne() throws Exception {
+        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
+            cluster.start();
+
+            String config = getConfig(PROXY_ADDRESS, cluster);
+
+            try (var proxy = startProxy(config)) {
+                createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+                produceAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE);
+            }
+        }
+    }
+
+    @Test
+    public void consumeOne() throws Exception {
+        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
+            cluster.start();
+
+            String config = getConfig(PROXY_ADDRESS, cluster);
+
+            try (var proxy = startProxy(config)) {
+                createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+                produceAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE);
+                consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE, false);
+            }
+        }
+    }
+
+    @Test
+    public void consumeOneAndOffsetCommit() throws Exception {
+        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
+            cluster.start();
+            String config = getConfig(PROXY_ADDRESS, cluster);
+
+            try (var proxy = startProxy(config)) {
+                createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+                produceAndVerify(TENANT1_PROXY_ADDRESS,
+                        Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, MY_VALUE + "1"), new ProducerRecord<>(TOPIC_1, MY_KEY, MY_VALUE + "2")));
+                consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE + "1", true);
+                consumeAndVerify(TENANT1_PROXY_ADDRESS, TOPIC_1, MY_KEY, MY_VALUE + "2", true);
+            }
+        }
+    }
+
+    @Test
+    public void tenantIsolation() throws Exception {
+        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder().testInfo(testInfo).build())) {
+            cluster.start();
+            String config = getConfig(PROXY_ADDRESS, cluster);
+
+            try (var proxy = startProxy(config)) {
+                createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
+                createTopics(TENANT2_PROXY_ADDRESS, List.of(NEW_TOPIC_2, NEW_TOPIC_3));
+
+                verifyTenant(TENANT1_PROXY_ADDRESS, TOPIC_1);
+                verifyTenant(TENANT2_PROXY_ADDRESS, TOPIC_2, TOPIC_3);
+            }
+        }
+    }
+
+    private void verifyTenant(String address, String... expectedTopics) throws Exception {
+        try (var admin = Admin.create(commonConfig(address, Map.of()))) {
+
+            var listTopicsResult = admin.listTopics();
+
+            var topicListMap = listTopicsResult.namesToListings().get();
+            assertEquals(expectedTopics.length, topicListMap.size());
+            Arrays.stream(expectedTopics).forEach(
+                    expectedTopic -> assertThat(topicListMap).hasEntrySatisfying(expectedTopic,
+                            allOf(matches(TopicListing::name, expectedTopic))));
+
+            var describeTopicsResult = admin.describeTopics(TopicNameCollection.ofTopicNames(Arrays.stream(expectedTopics).toList()));
+            var topicDescribeMap = describeTopicsResult.allTopicNames().get();
+            assertEquals(expectedTopics.length, topicDescribeMap.size());
+            Arrays.stream(expectedTopics).forEach(expectedTopic -> assertThat(topicDescribeMap).hasEntrySatisfying(expectedTopic,
+                    allOf(matches(TopicDescription::name, expectedTopic))));
+        }
+    }
+
+    private void consumeAndVerify(String address, String topicName, String expectedKey, String expectedValue, boolean offsetCommit) {
+        consumeAndVerify(address, topicName, new LinkedList<>(List.of(matchesRecord(topicName, expectedKey, expectedValue))), offsetCommit);
+    }
+
+    private void consumeAndVerify(String address, String topicName, Deque<Predicate<ConsumerRecord<String, String>>> expected, boolean offsetCommit) {
+        try (var consumer = new KafkaConsumer<String, String>(commonConfig(address, Map.of(
+                ConsumerConfig.GROUP_ID_CONFIG, testInfo.getDisplayName(),
+                ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, Boolean.FALSE.toString(),
+                ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, Boolean.FALSE.toString(),
+                ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, OffsetResetStrategy.EARLIEST.toString(),
+                ConsumerConfig.MAX_POLL_RECORDS_CONFIG, String.format("%d", expected.size()),
+                ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+                ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class)))) {
+
+            var topicPartitions = List.of(new TopicPartition(topicName, 0));
+            consumer.assign(topicPartitions);
+
+            while (!expected.isEmpty()) {
+                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(30));
+                assertThat(records.partitions()).hasSizeGreaterThanOrEqualTo(1);
+                records.forEach(r -> {
+                    assertFalse(expected.isEmpty(), String.format("received unexpected record %s", r));
+                    assertThat(r).matches(expected.pop());
+                });
+            }
+
+            if (offsetCommit) {
+                consumer.commitSync(Duration.ofSeconds(5));
+            }
+
+        }
+    }
+
+    private void produceAndVerify(String address, String topic, String key, String value) throws Exception {
+        produceAndVerify(address, Stream.of(new ProducerRecord<>(topic, key, value)));
+    }
+
+    private void produceAndVerify(String address, Stream<ProducerRecord<String, String>> records) throws Exception {
+
+        try (var producer = new KafkaProducer<String, String>(commonConfig(address, Map.of(
+                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000)))) {
+
+            records.forEach(rec -> {
+                RecordMetadata recordMetadata;
+                try {
+                    recordMetadata = producer.send(rec).get();
+                }
+                catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+                assertNotNull(recordMetadata);
+                assertNotNull(rec.topic(), recordMetadata.topic());
+            });
+
+        }
+    }
+
+    private CreateTopicsResult createTopics(String address, List<NewTopic> topics) throws Exception {
+        try (var admin = Admin.create(commonConfig(address, Map.of()))) {
+            return createTopics(admin, topics);
+        }
+    }
+
+    private static CreateTopicsResult createTopics(Admin admin, List<NewTopic> topics) throws Exception {
+        var created = admin.createTopics(topics);
+        assertEquals(topics.size(), created.values().size());
+        created.all().get();
+        return created;
+    }
+
+    private DeleteTopicsResult deleteTopics(String address, TopicCollection topics) throws Exception {
+        try (var admin = Admin.create(commonConfig(
+                address, Map.of()))) {
+            return deleteTopics(admin, topics);
+        }
+    }
+
+    private DeleteTopicsResult deleteTopics(Admin admin, TopicCollection topics) throws Exception {
+        var deleted = admin.deleteTopics(topics);
+        deleted.all().get();
+        return deleted;
+    }
+
+    @NotNull
+    private Map<String, Object> commonConfig(String address, Map<String, Object> m) {
+        var config = new HashMap<String, Object>();
+        config.putAll(m);
+        config.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, address);
+        config.put(CommonClientConfigs.CLIENT_ID_CONFIG, testInfo.getDisplayName());
+        config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name);
+        config.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, certificateGenerator.getCertLocation());
+        config.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, certificateGenerator.getPassword());
+        return config;
+    }
+
+    private String getConfig(String proxyAddress, KafkaCluster cluster) {
+        String config = """
+                proxy:
+                  address: %s
+                  keyStoreFile: %s
+                  keyPassword: %s
+                clusters:
+                  demo:
+                    bootstrap_servers: %s
+                filters:
+                - type: ApiVersions
+                - type: BrokerAddress
+                  config:
+                    addressMapperClazz: io.kroxylicious.proxy.internal.filter.SniAddressMapping
+                - type: MultiTenant
+                """.formatted(proxyAddress, certificateGenerator.getCertLocation(), certificateGenerator.getPassword(), cluster.getBootstrapServers());
+        return config;
+    }
+
+    private KafkaProxy startProxy(String config) throws InterruptedException {
+        Configuration proxyConfig = new ConfigParser().parseConfiguration(config);
+
+        KafkaProxy kafkaProxy = new KafkaProxy(proxyConfig);
+        kafkaProxy.startup();
+
+        return kafkaProxy;
+    }
+
+    @NotNull
+    private <T, V> Condition<T> matches(Function<T, V> extractor, V expectedValue) {
+        return new Condition<>(item -> Objects.equals(extractor.apply(item), expectedValue), "unexpected entry");
+    }
+
+    @NotNull
+    private static <K, V> Predicate<ConsumerRecord<K, V>> matchesRecord(final String expectedTopic, final K expectedKey, final V expectedValue) {
+        return item -> {
+            var rec = ((ConsumerRecord<K, V>) item);
+            return Objects.equals(rec.topic(), expectedTopic) && Objects.equals(rec.key(), expectedKey) && Objects.equals(rec.value(), expectedValue);
+        };
+    }
+}
diff --git a/integrationtests/src/test/resources/META-INF/services/java.net.spi.InetAddressResolverProvider b/integrationtests/src/test/resources/META-INF/services/java.net.spi.InetAddressResolverProvider
new file mode 100644
index 00000000..b80851e8
--- /dev/null
+++ b/integrationtests/src/test/resources/META-INF/services/java.net.spi.InetAddressResolverProvider
@@ -0,0 +1,1 @@
+io.kroxylicious.net.MultiTenantSniInetAddressResolverProvider
\ No newline at end of file
diff --git a/integrationtests/src/test/resources/log4j2-test.properties b/integrationtests/src/test/resources/log4j2-test.properties
index b8017dd9..a95d06a5 100644
--- a/integrationtests/src/test/resources/log4j2-test.properties
+++ b/integrationtests/src/test/resources/log4j2-test.properties
@@ -31,10 +31,15 @@ rootLogger.additivity = false
 #logger.netty.additivity = false
 #logger.netty.appenderRef.console.ref = STDOUT
 #
-#logger.broker.name = kafka
-#logger.broker.level = DEBUG
-#logger.broker.additivity = false
-#logger.broker.appenderRef.console.ref = STDOUT
+logger.broker.name = kafka
+logger.broker.level = DEBUG
+logger.broker.additivity = false
+logger.broker.appenderRef.console.ref = STDOUT
+
+logger.net.name = org.apache.kafka.clients.NetworkClient
+logger.net.level = debug
+
+
 #
 #logger.scl.name = state.change.logger
 #logger.scl.level = WARN
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
index 47db74f0..7309abe8 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
@@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.fail;
  * A test harness for {@link KrpcFilter} implementations.
  */
 public abstract class FilterHarness {
+    public static final String TEST_CLIENT = "test-client";
     protected EmbeddedChannel channel;
     private FilterHandler filterHandler;
     private KrpcFilter filter;
@@ -45,7 +46,7 @@ public abstract class FilterHarness {
      */
     protected void buildChannel(KrpcFilter filter, long timeoutMs) {
         this.filter = filter;
-        filterHandler = new FilterHandler(filter, timeoutMs);
+        filterHandler = new FilterHandler(filter, timeoutMs, null);
         channel = new EmbeddedChannel(filterHandler);
     }
 
@@ -62,7 +63,7 @@ public abstract class FilterHarness {
         header.setCorrelationId(correlationId);
         header.setRequestApiKey(apiKey.id);
         header.setRequestApiVersion(apiKey.latestVersion());
-        header.setClientId("test-client");
+        header.setClientId(TEST_CLIENT);
         var frame = new DecodedRequestFrame<>(apiKey.latestVersion(), correlationId, false, header, data);
         channel.writeOutbound(frame);
         return frame;
diff --git a/krpc-code-gen/src/test/resources/Kproxy/KrpcRequestFilter.ftl b/krpc-code-gen/src/test/resources/Kproxy/KrpcRequestFilter.ftl
index a1de5fdc..3ef48d9c 100644
--- a/krpc-code-gen/src/test/resources/Kproxy/KrpcRequestFilter.ftl
+++ b/krpc-code-gen/src/test/resources/Kproxy/KrpcRequestFilter.ftl
@@ -45,7 +45,7 @@ import io.kroxylicious.proxy.codec.DecodedRequestFrame;
  * <p>When extending {@link KrpcGenericRequestFilter} you need to override {@link #apply(DecodedRequestFrame, KrpcFilterContext)},
  * and may override {@link #shouldDeserializeRequest(ApiKeys, short)} as well.</p>
  *
- * <h3>Guarantees</h3>
+ * <h2>Guarantees</h2>
  * <p>Implementors of this API may assume the following:</p>
  * <ol>
  *     <li>That each instance of the filter is associated with a single channel</li>