Repository: kroxylicious/kroxylicious
Pull request #99 (2022-12-23 14:33:22)

[poc] multi-tenant create-topic/publish
I'm experimenting with kroxy by implementing a really simply multi-tenant.
Not yet intended to be commit-standard code.

k-wall: I have a really simple multi-tenant filter written that rewrites the topic names allows simple publish/consume use-cases, with supporting integration tests.

Next I want to experiment with SNI as a way to distinguish the tenant, and test out if Java 18's `InetAddressResolverProvider` provides an in-unobtrusive way to write integration tests for the feature.

 
k-wall: The PR now uses SNI to derive a tenant prefix.  It also now filters the `Metadata` response so that a tenant sees only those topics that belong to them. 

The integration tests use a `InetAddressResolverProvider` to arrange for `*.multitenant.kafka` DNS names to localhost.  This allows SNI to be spoofed, meaning we can test the multi-tenant feature properly, while avoid have to mess with `/etc/hosts` files etc.  InetAddressResolverProvider requires Java 18, which is why the test code now requires Java 18.  The production code remains unchanged.

tombentley: > I don't see multi tenancy as being a one and done step, though I think this is close to making it one, by way of example we probably want to enforce partition limits on a per tenant basis thus any enforcement filter would need to be tenant aware.

Labouring the point: For me this is not a given. It only needs to support a convention about prefixing, it doesn't need to be explicitly tenant aware.

Can you point to a way in which prefixing would not be sufficient?

> The other thing I'm wondering about is filter ordering, do we need some mechanism for filters to declare that they need the tenant to be determined before they appear in the chain? Probably not even in the medium term as long as the filter definition order is deterministic from the configuration but worth thinking about. One alternative might be to add a TenancyAware marker interface which would allow us to ensure they are in the filter chain behind the filter which determines the tenant.

This violates the principal that filters should compose without knowing about each other. _Maybe_ that is inevitable, but I think it's too soon to make that call. 

Composability is a lot to give up, because it means people won't be able to combine filters to easily, which undermines the ability for 3rd parties to write and distribute their own filters. 



SamBarker: > This violates the principal that filters should compose without knowing about each other. _Maybe_ that is inevitable, but I think it's too soon to make that call.
> 
> Composability is a lot to give up, because it means people won't be able to combine filters to easily, which undermines the ability for 3rd parties to write and distribute their own filters.

I don't want to give up composability either. In both your encryption example and my suggestion of policy enforcement there is an implicit ordering (it doesn't matter if we surface tenancy into the API or leave it as a prefixing scheme) both filters are dependant on being after the multi tenancy filter to work *as intended* yes they both work in a stand alone fashion but not as desired. I think my underlying point was should filters be able to provide their pre-conditions. 
 

> > I don't see multi tenancy as being a one and done step, though I think this is close to making it one, by way of example we probably want to enforce partition limits on a per tenant basis thus any enforcement filter would need to be tenant aware.

> Labouring the point: For me this is not a given. It only needs to support a convention about prefixing, it doesn't need to be explicitly tenant aware.

The ergonomics of the API for filter authors (I'm not convinced this is a small win). Each filter having to determine the prefix from the identifier it cares about seems to me to offer a great mine of compatibility issues. It would also mean the Filter author can't determine if a missing prefix is because its deployed in a single tenant mode or if something upstream has gone awry. For the encryption use case I can see that being useful to ensure the topic is encrypted with the appropriate key.

I'm coming round to the idea that we can merge this PR without resolving the question, though I think its an important one to get an answer to. 
tombentley: > both filters are dependant on being after the multi tenancy filter to work as intended yes they both work in a stand alone fashion but not as desired

I don't think that's right. As a concrete counterexample: In addition to my example with a `(MT, encryption)` chain, I think `(encryption)` alone is a perfectly valid use case. More abstractly, what's _intended_ is defined by the person configuring the chain, not by a person writing a filter. Giving the filter-writer a way of specifying preconditions allows them a way to veto the person who's actually trying to use the filters. Personally I'd favour the flexibility rather than over-constraining use, partly because it's impossible to guarantee "intended behaviour" via policy, so the configurer will _always_ have to do some work to validate they got the behaviour they intended.

> Each filter having to determine the prefix from the identifier it cares about seems to me to offer a great mine of compatibility issues.

Yes, I accept this isn't _ideal_. However another point occurred to me: This problem will necessarily exist in the broker cluster _anyway_, where there is not, and cannot be, a first-class concept of tenant id. When looking at broker metrics all you'll have is topic names, group ids etc. That being the case improving the proxy story around first-class tenant id doesn't seem so beneficial.


k-wall: @tombentley @SamBarker I propose to merge this work as it stands.   We can iterate on it incrementally.

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..80e9b83d 100644
--- a/README.md
+++ b/README.md
@@ -106,7 +106,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/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/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/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/pom.xml b/pom.xml
index 0170b899..600885fc 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>


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;