Repository: kroxylicious/kroxylicious
Pull request #254 (2023-04-04 16:54:48)

Expose multiple clusters / multiple brokers (via static configuration)
### Type of change

_Select the type of your PR_

- Enhancement / new feature

### Description

With this change, kroxylicious is capable of proxying multiple kafka clusters and has static support for clusters formed of more than one broker.

The model is changed to introduce the concept of _virtual cluster_ (downstream side) and _target cluster_ (upstream side) to disambiguate the two notions of cluster.  Kafka clients connect the the _virtual cluster_ and kroxylicious forwards those connections through to the cluster identified by the target cluster definition.

Two topologies are supported:
* *one to one* -  one virtual cluster maps to one target cluster, and
* *many to one*  - many virtual clusters map to one target cluster.

The `ClusterEndpointProvider` is class responsible for providing the virtual cluster's bootstrap.  It also acts as a provider for broker addresses - the endpoints used by the client to talk to the individual brokers.  There currently is a single implementation of the CEP - `StaticClusterEndpointProvider` which allows a cluster that defines of more than one broker to be defined statically.   

It is envisaged that there will be other CEP implementations that will react dynamically to the the target cluster's broker topology and respond by binding/unbinding ports on the downstream stream side accordingly.  It is also envisaged that there will be a CEP implementation that will permit, in TLS/SNI use-cases, the sharing a single port by all virtual clusters and all brokers.   I plan to raise these as separate issues and can be delivered by separate PRs.

I also plan to raise a separate issue for the refactoring of the metrics.  I think metrics should be gathered on a per-virtual cluster basis, rather than kroxy wide.  I think this change can follow as a separate PR.


### Additional Context

_Why are you making this pull request?_


sonatype-lift[bot]: ## 🛠 Lift Auto-fix
Some of the Lift findings in this PR can be automatically fixed. You can download and apply these changes in your local project directory of your branch to review the suggestions before committing.[^1]
```bash
# Download the patch
curl https://lift.sonatype.com/api/patch/github.com/kroxylicious/kroxylicious/254.diff -o lift-autofixes.diff

# Apply the patch with git
git apply lift-autofixes.diff

# Review the changes
git diff
```
Want it all in a single command? Open a terminal in your project's directory and copy and paste the following command:
```bash
curl https://lift.sonatype.com/api/patch/github.com/kroxylicious/kroxylicious/254.diff | git apply
```
Once you're satisfied, commit and push your changes in your project.
[^1]: You can preview the patch by opening the [patch URL](https://lift.sonatype.com/api/patch/github.com/kroxylicious/kroxylicious/254.diff) in the browser.
k-wall: @SamBarker @devguyio @robobario My hope is that this PR is viewed as a reason step towards having kroxy dynamically configure itself based on the target cluster broker topology.  It also has the bonus that we can start experimenting with proxying clusters with > 1 broker now.   

robobario: @k-wall currently, in this PR the physical-cluster is not yet an entity in configuration. Are we still going to add that in future, or is it not needed?
k-wall: > @k-wall currently, in this PR the physical-cluster is not yet an entity in configuration. Are we still going to add that in future, or is it not needed?

I gave the `virtualCluster` a `targetCluster`.  That's the model representation of the cluster that Kroxylicious will connect to.     That's what I'd previously referred to as a physical cluster.  The term physical cluster got awkward in the writing, especially when  I wanted to refer to the actual physical kafka cluster.
k-wall: @robobario @SamBarker thanks for all the comments so far.  I hope the resolutions are okay.  Please re-review.
k-wall: @robobario @SamBarker I made one further change to fix a defect I found today.  The change is functional, but I want to refactor it more.   I expect a future PR will remove NetFilter, so I expect that BrokerAddressCachingNetFilter won't be around for long.
k-wall: > @robobario @SamBarker I made one further change to fix a defect I found today. The change is functional, but I want to refactor it more. I expect a future PR will remove NetFilter, so I expect that BrokerAddressCachingNetFilter won't be around for long.

The refactoring I talked about above is done.  Like I say in the code `UpstreamBrokerAddressCachingNetFilter` is expected to a temporary solution.  I see this being replace once we give kroxy the responsibility to monitor broker topology and start downstream side listeners.


Patch:
diff --git a/.github/workflows/performance.yml b/.github/workflows/performance.yml
index e720a8af..fb0e7190 100644
--- a/.github/workflows/performance.yml
+++ b/.github/workflows/performance.yml
@@ -113,6 +113,7 @@ jobs:
           auto-push: true
           alert-threshold: '150%'
           comment-on-alert: true
+          summary-always: true
           fail-on-alert: true
           alert-comment-cc-users: '@kroxylicious/developers'
 
@@ -202,5 +203,6 @@ jobs:
           auto-push: true
           alert-threshold: '150%'
           comment-on-alert: false
+          summary-always: true
           fail-on-alert: true
           alert-comment-cc-users: '@kroxylicious/developers'
diff --git a/api/kroxylicious-api/pom.xml b/api/kroxylicious-api/pom.xml
index 77a6e2d5..621709ca 100644
--- a/api/kroxylicious-api/pom.xml
+++ b/api/kroxylicious-api/pom.xml
@@ -41,5 +41,10 @@
             <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusterendpointprovider/ClusterEndpointProviderContributor.java b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusterendpointprovider/ClusterEndpointProviderContributor.java
new file mode 100644
index 00000000..a325c71b
--- /dev/null
+++ b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusterendpointprovider/ClusterEndpointProviderContributor.java
@@ -0,0 +1,16 @@
+/*
+ * 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.clusterendpointprovider;
+
+import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
+import io.kroxylicious.proxy.service.Contributor;
+
+/**
+ * ClusterEndpointConfigProvider is a pluggable source of Kroxylicious endpoint provider implementations.
+ * @see Contributor
+ */
+public interface ClusterEndpointProviderContributor extends Contributor<ClusterEndpointConfigProvider> {
+}
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/config/ProxyConfig.java b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/config/ProxyConfig.java
deleted file mode 100644
index 3035aead..00000000
--- a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/config/ProxyConfig.java
+++ /dev/null
@@ -1,10 +0,0 @@
-/*
- * 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.config;
-
-public interface ProxyConfig {
-    String address();
-}
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java
index 1697cfc7..983af51a 100644
--- a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java
+++ b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java
@@ -12,7 +12,6 @@ import java.util.function.Function;
 import java.util.function.Supplier;
 
 import io.kroxylicious.proxy.config.BaseConfig;
-import io.kroxylicious.proxy.config.ProxyConfig;
 
 /**
  * A convenience base class for creating concrete contributor subclasses using a typesafe builder
@@ -38,28 +37,28 @@ public abstract class BaseContributor<T> implements Contributor<T> {
     }
 
     @Override
-    public T getInstance(String shortName, ProxyConfig proxyConfig, BaseConfig config) {
+    public T getInstance(String shortName, ClusterEndpointConfigProvider endpointConfigProvider, BaseConfig config) {
         InstanceBuilder<? extends BaseConfig, T> instanceBuilder = shortNameToInstanceBuilder.get(shortName);
-        return instanceBuilder == null ? null : instanceBuilder.construct(proxyConfig, config);
+        return instanceBuilder == null ? null : instanceBuilder.construct(endpointConfigProvider, config);
     }
 
     private static class InstanceBuilder<T extends BaseConfig, L> {
 
         private final Class<T> configClass;
-        private final BiFunction<ProxyConfig, T, L> instanceFunction;
+        private final BiFunction<ClusterEndpointConfigProvider, T, L> instanceFunction;
 
-        InstanceBuilder(Class<T> configClass, BiFunction<ProxyConfig, T, L> instanceFunction) {
+        InstanceBuilder(Class<T> configClass, BiFunction<ClusterEndpointConfigProvider, T, L> instanceFunction) {
             this.configClass = configClass;
             this.instanceFunction = instanceFunction;
         }
 
-        L construct(ProxyConfig proxyConfig, BaseConfig config) {
+        L construct(ClusterEndpointConfigProvider endpointConfigProvider, BaseConfig config) {
             if (config == null) {
                 // tests pass in a null config, which some instance functions can tolerate
-                return instanceFunction.apply(proxyConfig, null);
+                return instanceFunction.apply(endpointConfigProvider, null);
             }
             else if (configClass.isAssignableFrom(config.getClass())) {
-                return instanceFunction.apply(proxyConfig, configClass.cast(config));
+                return instanceFunction.apply(endpointConfigProvider, configClass.cast(config));
             }
             else {
                 throw new IllegalArgumentException("config has the wrong type, expected "
@@ -89,7 +88,8 @@ public abstract class BaseContributor<T> implements Contributor<T> {
          * @return this
          * @param <T> the configuration concrete type
          */
-        public <T extends BaseConfig> BaseContributorBuilder<L> add(String shortName, Class<T> configClass, BiFunction<ProxyConfig, T, L> instanceFunction) {
+        public <T extends BaseConfig> BaseContributorBuilder<L> add(String shortName, Class<T> configClass,
+                                                                    BiFunction<ClusterEndpointConfigProvider, T, L> instanceFunction) {
             if (shortNameToInstanceBuilder.containsKey(shortName)) {
                 throw new IllegalArgumentException(shortName + " already registered");
             }
@@ -104,7 +104,7 @@ public abstract class BaseContributor<T> implements Contributor<T> {
          * @param instanceFunction function that constructs the service instance
          * @return this
          */
-        public BaseContributorBuilder<L> add(String shortName, Function<ProxyConfig, L> instanceFunction) {
+        public BaseContributorBuilder<L> add(String shortName, Function<ClusterEndpointConfigProvider, L> instanceFunction) {
             add(shortName, BaseConfig.class, (proxyConfig, config) -> instanceFunction.apply(proxyConfig));
             return this;
         }
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterEndpointConfigProvider.java b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterEndpointConfigProvider.java
new file mode 100644
index 00000000..772bbdfc
--- /dev/null
+++ b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterEndpointConfigProvider.java
@@ -0,0 +1,73 @@
+/*
+ * 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.service;
+
+import java.util.Optional;
+
+/**
+ * Provides the addresses of the network endpoints required by a virtual cluster.
+ */
+public interface ClusterEndpointConfigProvider {
+
+    /**
+     * Address of the cluster's bootstrap address.
+     *
+     * @return cluster's bootstrap address.
+     */
+    HostPort getClusterBootstrapAddress();
+
+    /**
+     * Address of broker with the given node id, includes the port. Note that
+     * {@code nodeId} are generally expected to be consecutively numbered and starting from zero. However, gaps in the sequence can potentially emerge as
+     * the target cluster topology evolves.
+     *
+     * @param nodeId node identifier
+     * @return broker address
+     * @throws IllegalArgumentException if this provider cannot produce a broker address for the given nodeId.
+     */
+    HostPort getBrokerAddress(int nodeId) throws IllegalArgumentException;
+
+    /**
+     * Tests whether this endpoint providers has a matching endpoint for the given parameters.  This considers both bootstrap and broker endpoints.
+     * If the endpoint is known and corresponds to a broker endpoint, the result encapsulates the {@code nodeId} of the endpoint.
+     *
+     * @param sniHostname sniHostname provided by TLS negotiation or null if TLS is not in use or SNI is not available.
+     * @param port local port number of the connection
+     * @return an {@link EndpointMatchResult}
+     */
+    EndpointMatchResult hasMatchingEndpoint(String sniHostname, int port);
+
+    /**
+     * Provides the number of broker endpoints to pre-bind.
+     * Kroxylicious will pre-bind all the broker end-points between
+     * 0..{@code numberOfWarmStartBrokerEndpoints}-1 on startup, rather
+     * than waiting for the first bootstrap connection to allow their
+     * discovery.
+     *
+     * @return number of endpoints to pre-bind on startup.
+     */
+    default int getNumberOfBrokerEndpointsToPrebind() {
+        return 1;
+    }
+
+    /**
+     * Gets the bind address used when binding socket.  Used to restrict
+     * listening to particular network interfaces.
+     * @return bind address such as "127.0.0.1" or {@code }Optional.empty()} if all address should be bound.
+     */
+    default Optional<String> getBindAddress() {
+        return Optional.empty();
+    }
+
+    /**
+     * Encapsulates the result of a endpoint match.
+     *
+     * @param matched true if matched, false otherwise.
+     * @param nodeId if a broker endpoint match, carries the {@code nodeId}.
+     */
+    record EndpointMatchResult(boolean matched, Integer nodeId) {
+    }
+}
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/Contributor.java b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/Contributor.java
index 09dbfa10..912cb783 100644
--- a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/Contributor.java
+++ b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/Contributor.java
@@ -6,7 +6,6 @@
 package io.kroxylicious.proxy.service;
 
 import io.kroxylicious.proxy.config.BaseConfig;
-import io.kroxylicious.proxy.config.ProxyConfig;
 
 /**
  * Support loading an Instance of a service, optionally providing it with configuration obtained
@@ -28,9 +27,9 @@ public interface Contributor<T> {
      * Creates an instance of the service.
      *
      * @param shortName service short name
-     * @param proxyConfig proxy configuration
+     * @param endpointConfigProvider endpoint provider
      * @param config service configuration which may be null if the service instance does not accept configuration.
      * @return the service instance
      */
-    T getInstance(String shortName, ProxyConfig proxyConfig, BaseConfig config);
+    T getInstance(String shortName, ClusterEndpointConfigProvider endpointConfigProvider, BaseConfig config);
 }
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/HostPort.java b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/HostPort.java
new file mode 100644
index 00000000..34eb3401
--- /dev/null
+++ b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/HostPort.java
@@ -0,0 +1,83 @@
+/*
+ * 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.service;
+
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+
+/**
+ * Represents a host port pair.
+ *
+ * @param host host name
+ * @param port port number.
+ */
+public record HostPort(String host, int port) {
+    private static final Pattern IPV6_WITH_PORT = Pattern.compile("^(\\[.+]):(.+)$");
+    private static final Pattern PORT_SEPARATOR = Pattern.compile(":");
+
+    /**
+     * Creates a host port.
+     */
+     public HostPort {
+        Objects.requireNonNull(host, "host cannot be null");
+    }
+
+    @Override
+    public String toString() {
+        return host + ":" + port;
+    }
+
+    /**
+     * Parses a host port pair from a string representation.
+     *
+     * For the host part, symbolic hostname, FQDN, IPv4, and IPv6 forms are supported.  In the case of IPv6,
+     * the notation specified by <a href="https://www.rfc-editor.org/rfc/rfc4038#section-5.1">rfc4038</a> must be used.
+     *
+     * For the port part, it must be numeric.
+     *
+     * @param address stringified form of the host port, separated by colon (:).
+     * @return a {@link HostPort}
+     */
+    @JsonCreator
+    public static HostPort parse(String address) {
+        var exceptionText = "unexpected address formation '%s'. Valid formations are 'host:9092', 'host.example.com:9092', or '[::ffff:c0a8:1]:9092', ".formatted(address);
+
+        if (address == null) {
+            throw new IllegalArgumentException(exceptionText);
+        }
+
+        var ipv6Match = IPV6_WITH_PORT.matcher(address);
+        if (ipv6Match.matches()) {
+            var host = ipv6Match.group(1);
+            var port = parsePort(exceptionText, ipv6Match.group(2));
+            return new HostPort(host, port);
+        }
+        else {
+            var split = PORT_SEPARATOR.split(address);
+            if (split.length != 2) {
+                throw new IllegalArgumentException(exceptionText);
+            }
+            var host = split[0];
+            if (host.isEmpty() || host.isBlank()) {
+                throw new IllegalArgumentException(exceptionText);
+            }
+            var port = parsePort(exceptionText, split[1]);
+            return new HostPort(host, port);
+        }
+    }
+
+    private static int parsePort(String exceptionText, String group) {
+        try {
+            return Integer.parseInt(group);
+        }
+        catch (NumberFormatException nfe) {
+            throw new IllegalArgumentException(exceptionText, nfe);
+        }
+    }
+}
diff --git a/kroxylicious/example-proxy-config.yml b/kroxylicious/example-proxy-config.yml
index 49cb035e..8491dba7 100644
--- a/kroxylicious/example-proxy-config.yml
+++ b/kroxylicious/example-proxy-config.yml
@@ -5,16 +5,19 @@
 #
 
 ---
-proxy:
-  address: localhost:9192
-  logNetwork: false
-  logFrames: false
 adminHttp:
   endpoints:
     prometheus: {}
-clusters:
+virtualClusters:
   demo:
-    bootstrap_servers: localhost:9092
+    targetCluster:
+      bootstrap_servers: localhost:9092
+    clusterEndpointConfigProvider:
+      type: StaticCluster
+      config:
+        bootstrapAddress: localhost:9192
+    logNetwork: false
+    logFrames: false
 filters:
 - type: ApiVersions
 - type: BrokerAddress
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
index ce8aeabe..c31ef4bf 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
@@ -13,9 +13,14 @@ import java.security.KeyStoreException;
 import java.security.NoSuchAlgorithmException;
 import java.security.UnrecoverableKeyException;
 import java.security.cert.CertificateException;
+import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.net.ssl.KeyManagerFactory;
 
@@ -37,21 +42,23 @@ import io.netty.channel.kqueue.KQueueEventLoopGroup;
 import io.netty.channel.kqueue.KQueueServerSocketChannel;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.incubator.channel.uring.IOUring;
 import io.netty.incubator.channel.uring.IOUringEventLoopGroup;
 import io.netty.incubator.channel.uring.IOUringServerSocketChannel;
+import io.netty.util.concurrent.Future;
 
+import io.kroxylicious.proxy.bootstrap.ClusterEndpointProviderFactory;
 import io.kroxylicious.proxy.bootstrap.FilterChainFactory;
 import io.kroxylicious.proxy.config.Configuration;
 import io.kroxylicious.proxy.config.MicrometerDefinition;
-import io.kroxylicious.proxy.config.ProxyConfig;
+import io.kroxylicious.proxy.config.VirtualCluster;
 import io.kroxylicious.proxy.config.admin.AdminHttpConfiguration;
 import io.kroxylicious.proxy.internal.KafkaProxyInitializer;
 import io.kroxylicious.proxy.internal.MeterRegistries;
 import io.kroxylicious.proxy.internal.admin.AdminHttpInitializer;
-import io.kroxylicious.proxy.internal.filter.FixedNetFilter;
+import io.kroxylicious.proxy.internal.filter.UpstreamBrokerAddressCachingNetFilter;
+import io.kroxylicious.proxy.service.HostPort;
 
 import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES;
 import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES;
@@ -59,76 +66,31 @@ import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_REQUEST_S
 
 public final class KafkaProxy implements AutoCloseable {
 
+    private record EventGroupConfig(EventLoopGroup bossGroup, EventLoopGroup workerGroup, Class<? extends ServerChannel> clazz) {
+        public List<Future<?>> shutdownGracefully() {
+            return List.of(bossGroup.shutdownGracefully(), workerGroup.shutdownGracefully());
+        }
+    };
+
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProxy.class);
 
-    private final String proxyHost;
-    private final int proxyPort;
-    private final String brokerHost;
-    private final int brokerPort;
-    private final boolean logNetwork;
-    private final boolean logFrames;
-    private final boolean useIoUring;
-    private final FilterChainFactory filterChainFactory;
+    private final Configuration config;
     private final AdminHttpConfiguration adminHttpConfig;
     private final List<MicrometerDefinition> micrometerConfig;
-    private final ProxyConfig proxyConfig;
-    private EventLoopGroup bossGroup;
-    private EventLoopGroup workerGroup;
-    private Channel acceptorChannel;
+    private final Map<String, VirtualCluster> virtualClusterMap;
+    private EventGroupConfig adminEventGroup;
+    private final List<EventGroupConfig> virtualHostEventGroups = new CopyOnWriteArrayList<>();
+
+    private final Queue<Channel> acceptorChannels = new ConcurrentLinkedQueue<>();
+    private final AtomicBoolean running = new AtomicBoolean();
     private Channel metricsChannel;
-    private Optional<File> keyStoreFile;
-    private Optional<String> keyStorePassword;
 
     public KafkaProxy(Configuration config) {
-        String proxyAddress = config.proxy().address();
-        String[] proxyAddressParts = proxyAddress.split(":");
-
-        // TODO: deal with list
-        String brokerAddress = config.clusters().entrySet().iterator().next().getValue().bootstrapServers();
-        String[] brokerAddressParts = brokerAddress.split(":");
-
-        this.proxyHost = proxyAddressParts[0];
-        this.proxyPort = Integer.valueOf(proxyAddressParts[1]);
-        this.brokerHost = brokerAddressParts[0];
-        this.brokerPort = Integer.valueOf(brokerAddressParts[1]);
-        this.logNetwork = config.proxy().logNetwork();
-        this.logFrames = config.proxy().logFrames();
-        this.useIoUring = config.proxy().useIoUring();
+        this.config = config;
+        this.virtualClusterMap = config.virtualClusters();
         this.adminHttpConfig = config.adminHttpConfig();
         this.micrometerConfig = config.getMicrometer();
-        this.proxyConfig = config.proxy();
-        this.filterChainFactory = new FilterChainFactory(config);
-
-        this.keyStoreFile = config.proxy().keyStoreFile().map(File::new);
-        this.keyStorePassword = config.proxy().keyPassword();
-    }
 
-    public String proxyHost() {
-        return proxyHost;
-    }
-
-    public int proxyPort() {
-        return proxyPort;
-    }
-
-    public String proxyAddress() {
-        return proxyHost() + ":" + proxyPort();
-    }
-
-    public String brokerHost() {
-        return brokerHost;
-    }
-
-    public int brokerPort() {
-        return brokerPort;
-    }
-
-    public String brokerAddress() {
-        return brokerHost() + ":" + brokerPort();
-    }
-
-    public boolean useIoUring() {
-        return useIoUring;
     }
 
     /**
@@ -136,40 +98,111 @@ public final class KafkaProxy implements AutoCloseable {
      * @return This proxy.
      */
     public KafkaProxy startup() throws InterruptedException {
-        if (acceptorChannel != null) {
+        if (running.getAndSet(true)) {
             throw new IllegalStateException("This proxy is already running");
         }
 
-        LOGGER.info("Proxying local {} to remote {}",
-                proxyAddress(), brokerAddress());
-
-        Optional<SslContext> sslContext = keyStoreFile.map(ksf -> {
-            try (var is = new FileInputStream(ksf)) {
-                var password = keyStorePassword.map(String::toCharArray).orElse(null);
-                var keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
-                keyStore.load(is, password);
-                var keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
-                keyManagerFactory.init(keyStore, password);
-                return SslContextBuilder.forServer(keyManagerFactory).build();
+        var availableCores = Runtime.getRuntime().availableProcessors();
+        var meterRegistries = new MeterRegistries(micrometerConfig);
+        this.adminEventGroup = buildNettyEventGroups(availableCores, false);
+        maybeStartMetricsListener(adminEventGroup, meterRegistries);
+
+        virtualClusterMap.forEach((name, virtualCluster) -> {
+
+            var endpointProvider = new ClusterEndpointProviderFactory(config, virtualCluster).createClusterEndpointProvider();
+
+            // TODO: target bootstrap servers could be a list - we should be prepared to try them all.
+            var targetBootstrapServers = virtualCluster.targetCluster().bootstrapServers();
+            var targetBootstrapServersParts = targetBootstrapServers.split(",");
+            var brokerAddressParts = targetBootstrapServersParts[0].split(":");
+            var brokerHost = brokerAddressParts[0];
+            var brokerPort = Integer.valueOf(brokerAddressParts[1]);
+
+            var keyStoreFile = virtualCluster.keyStoreFile().map(File::new);
+            var keyStorePassword = virtualCluster.keyPassword();
+
+            var filterChainFactory = new FilterChainFactory(config, endpointProvider);
+
+            LOGGER.info("{}: Proxying local {} to remote {}",
+                    name, endpointProvider.getClusterBootstrapAddress(), targetBootstrapServers);
+
+            var sslContext = keyStoreFile.map(ksf -> {
+                try (var is = new FileInputStream(ksf)) {
+                    var password = keyStorePassword.map(String::toCharArray).orElse(null);
+                    var keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
+                    keyStore.load(is, password);
+                    var keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+                    keyManagerFactory.init(keyStore, password);
+                    return SslContextBuilder.forServer(keyManagerFactory).build();
+                }
+                catch (KeyStoreException | IOException | NoSuchAlgorithmException | CertificateException | UnrecoverableKeyException e) {
+                    throw new RuntimeException(e);
+                }
+            });
+
+            KafkaProxyInitializer initializer = new KafkaProxyInitializer(false,
+                    Map.of(),
+                    new UpstreamBrokerAddressCachingNetFilter(
+                            new HostPort(brokerHost, brokerPort),
+                            filterChainFactory,
+                            endpointProvider),
+                    virtualCluster.isLogNetwork(),
+                    virtualCluster.isLogFrames(),
+                    sslContext);
+
+            // Configure the bootstrap.
+            var virtualHostEventGroup = buildNettyEventGroups(availableCores, virtualCluster.isUseIoUring());
+            this.virtualHostEventGroups.add(virtualHostEventGroup);
+
+            ServerBootstrap serverBootstrap = new ServerBootstrap().group(virtualHostEventGroup.bossGroup(), virtualHostEventGroup.workerGroup())
+                    .channel(virtualHostEventGroup.clazz())
+                    .option(ChannelOption.SO_REUSEADDR, true)
+                    .childHandler(initializer)
+                    .childOption(ChannelOption.AUTO_READ, false)
+                    .childOption(ChannelOption.TCP_NODELAY, true);
+            ChannelFuture bindFuture;
+
+            var toBind = new HashSet<HostPort>();
+            toBind.add(endpointProvider.getClusterBootstrapAddress());
+            for (int i = 0; i < endpointProvider.getNumberOfBrokerEndpointsToPrebind(); i++) {
+                toBind.add(endpointProvider.getBrokerAddress(i));
             }
-            catch (KeyStoreException | IOException | NoSuchAlgorithmException | CertificateException | UnrecoverableKeyException e) {
-                throw new RuntimeException(e);
+
+            List<ChannelFuture> bindFutures = new ArrayList<>();
+            for (var address : toBind) {
+                var proxyPort = address.port();
+
+                if (endpointProvider.getBindAddress().isPresent()) {
+                    bindFuture = serverBootstrap.bind(endpointProvider.getBindAddress().get(), proxyPort);
+                }
+                else {
+                    bindFuture = serverBootstrap.bind(proxyPort);
+                }
+                bindFutures.add(bindFuture);
+
+                bindFuture.addListener(f -> {
+                    var channel = ((ChannelFuture) f).channel();
+                    acceptorChannels.add(channel);
+                    LOGGER.info("{}: Listener started {}", name, channel.localAddress());
+                });
             }
-        });
 
-        KafkaProxyInitializer initializer = new KafkaProxyInitializer(false,
-                Map.of(),
-                new FixedNetFilter(brokerHost,
-                        brokerPort,
-                        filterChainFactory),
-                logNetwork,
-                logFrames,
-                sslContext);
+            bindFutures.forEach(ChannelFuture::syncUninterruptibly);
+        });
 
-        final int availableCores = Runtime.getRuntime().availableProcessors();
+        // Pre-register counters/summaries to avoid creating them on first request and thus skewing the request latency
+        // TODO add a virtual host tag to metrics
+        Metrics.counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES);
+        Metrics.counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES);
+        Metrics.summary(KROXYLICIOUS_REQUEST_SIZE_BYTES);
+        return this;
+    }
 
-        // Configure the bootstrap.
+    private EventGroupConfig buildNettyEventGroups(int availableCores, boolean useIoUring) {
         final Class<? extends ServerChannel> channelClass;
+        final EventLoopGroup bossGroup;
+        final EventLoopGroup workerGroup;
+
         if (useIoUring) {
             if (!IOUring.isAvailable()) {
                 throw new IllegalStateException("io_uring not available due to: " + IOUring.unavailabilityCause());
@@ -193,39 +226,17 @@ public final class KafkaProxy implements AutoCloseable {
             workerGroup = new NioEventLoopGroup(availableCores);
             channelClass = NioServerSocketChannel.class;
         }
-
-        MeterRegistries meterRegistries = new MeterRegistries(micrometerConfig, proxyConfig);
-        maybeStartMetricsListener(bossGroup, workerGroup, channelClass, meterRegistries);
-
-        ServerBootstrap serverBootstrap = new ServerBootstrap().group(bossGroup, workerGroup)
-                .channel(channelClass)
-                .childHandler(initializer)
-                .childOption(ChannelOption.AUTO_READ, false)
-                .childOption(ChannelOption.TCP_NODELAY, true);
-        ChannelFuture bindFuture;
-        if (proxyHost != null) {
-            bindFuture = serverBootstrap.bind(proxyHost, proxyPort);
-        }
-        else {
-            bindFuture = serverBootstrap.bind(proxyPort);
-        }
-        acceptorChannel = bindFuture.sync().channel();
-        // Pre-register counters/summaries to avoid creating them on first request and thus skewing the request latency
-        Metrics.counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES);
-        Metrics.counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES);
-        Metrics.summary(KROXYLICIOUS_REQUEST_SIZE_BYTES);
-        return this;
+        return new EventGroupConfig(bossGroup, workerGroup, channelClass);
     }
 
-    private void maybeStartMetricsListener(EventLoopGroup bossGroup,
-                                           EventLoopGroup workerGroup,
-                                           Class<? extends ServerChannel> channelClass,
+    private void maybeStartMetricsListener(EventGroupConfig eventGroupConfig,
                                            MeterRegistries meterRegistries)
             throws InterruptedException {
         if (adminHttpConfig != null
                 && adminHttpConfig.getEndpoints().maybePrometheus().isPresent()) {
-            ServerBootstrap metricsBootstrap = new ServerBootstrap().group(bossGroup, workerGroup)
-                    .channel(channelClass)
+            ServerBootstrap metricsBootstrap = new ServerBootstrap().group(eventGroupConfig.bossGroup(), eventGroupConfig.workerGroup())
+                    .option(ChannelOption.SO_REUSEADDR, true)
+                    .channel(eventGroupConfig.clazz())
                     .childHandler(new AdminHttpInitializer(meterRegistries, adminHttpConfig));
             metricsChannel = metricsBootstrap.bind(adminHttpConfig.getHost(), adminHttpConfig.getPort()).sync().channel();
         }
@@ -237,10 +248,15 @@ public final class KafkaProxy implements AutoCloseable {
      * @throws InterruptedException
      */
     public void block() throws InterruptedException {
-        if (acceptorChannel == null) {
+        if (!running.get()) {
             throw new IllegalStateException("This proxy is not running");
         }
-        acceptorChannel.closeFuture().sync();
+        while (!acceptorChannels.isEmpty()) {
+            Channel channel = acceptorChannels.peek();
+            if (channel != null) {
+                channel.closeFuture().sync();
+            }
+        }
     }
 
     /**
@@ -248,20 +264,21 @@ public final class KafkaProxy implements AutoCloseable {
      * @throws InterruptedException
      */
     public void shutdown() throws InterruptedException {
-        if (acceptorChannel == null) {
+        if (!running.getAndSet(false)) {
             throw new IllegalStateException("This proxy is not running");
         }
-        bossGroup.shutdownGracefully().sync();
-        workerGroup.shutdownGracefully().sync();
-        bossGroup = null;
-        workerGroup = null;
-        acceptorChannel = null;
+        var closeFutures = new ArrayList<Future<?>>();
+        virtualHostEventGroups.forEach(g -> closeFutures.addAll(g.shutdownGracefully()));
+        closeFutures.addAll(adminEventGroup.shutdownGracefully());
+        closeFutures.forEach(Future::syncUninterruptibly);
+        virtualHostEventGroups.clear();
+        acceptorChannels.clear();
         metricsChannel = null;
     }
 
     @Override
     public void close() throws Exception {
-        if (acceptorChannel != null) {
+        if (running.get()) {
             shutdown();
         }
     }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/ClusterEndpointProviderFactory.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/ClusterEndpointProviderFactory.java
new file mode 100644
index 00000000..3eb17485
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/ClusterEndpointProviderFactory.java
@@ -0,0 +1,28 @@
+/*
+ * 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.bootstrap;
+
+import io.kroxylicious.proxy.config.Configuration;
+import io.kroxylicious.proxy.config.VirtualCluster;
+import io.kroxylicious.proxy.internal.clusterendpointprovider.ClusterEndpointConfigProviderContributorManager;
+import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
+
+public class ClusterEndpointProviderFactory {
+
+    private final Configuration config;
+    private final VirtualCluster virtualCluster;
+
+    public ClusterEndpointProviderFactory(Configuration config, VirtualCluster virtualCluster) {
+        this.config = config;
+        this.virtualCluster = virtualCluster;
+    }
+
+    public ClusterEndpointConfigProvider createClusterEndpointProvider() {
+        return ClusterEndpointConfigProviderContributorManager.getInstance()
+                .getClusterEndpointConfigProvider(virtualCluster.clusterEndpointProvider().type(), virtualCluster.clusterEndpointProvider()
+                        .config());
+    }
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java
index f6873019..bcff3bb9 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java
@@ -8,6 +8,7 @@ package io.kroxylicious.proxy.bootstrap;
 import io.kroxylicious.proxy.config.Configuration;
 import io.kroxylicious.proxy.filter.KrpcFilter;
 import io.kroxylicious.proxy.internal.filter.FilterContributorManager;
+import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
 
 /**
  * Abstracts the creation of a chain of filter instances, hiding the configuration
@@ -17,9 +18,11 @@ import io.kroxylicious.proxy.internal.filter.FilterContributorManager;
 public class FilterChainFactory {
 
     private final Configuration config;
+    private final ClusterEndpointConfigProvider clusterEndpointConfigProvider;
 
-    public FilterChainFactory(Configuration config) {
+    public FilterChainFactory(Configuration config, ClusterEndpointConfigProvider clusterEndpointConfigProvider) {
         this.config = config;
+        this.clusterEndpointConfigProvider = clusterEndpointConfigProvider;
     }
 
     /**
@@ -31,7 +34,7 @@ public class FilterChainFactory {
 
         return config.filters()
                 .stream()
-                .map(f -> filterContributorManager.getFilter(f.type(), config.proxy(), f.config()))
+                .map(f -> filterContributorManager.getFilter(f.type(), clusterEndpointConfigProvider, f.config()))
                 .toArray(KrpcFilter[]::new);
     }
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/Cluster.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/Cluster.java
deleted file mode 100644
index e0dcb49f..00000000
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/Cluster.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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.config;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-public class Cluster {
-
-    private final String bootstrapServers;
-
-    public Cluster(@JsonProperty(value = "bootstrap_servers") String bootstrapServers) {
-        this.bootstrapServers = bootstrapServers;
-    }
-
-    public String bootstrapServers() {
-        return bootstrapServers;
-    }
-
-    @Override
-    public String toString() {
-        return "Cluster [bootstrapServers=" + bootstrapServers + "]";
-    }
-}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderDefinition.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderDefinition.java
new file mode 100644
index 00000000..ee0d2679
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderDefinition.java
@@ -0,0 +1,31 @@
+/*
+ * 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.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.fasterxml.jackson.databind.annotation.JsonTypeIdResolver;
+
+public class ClusterEndpointConfigProviderDefinition {
+
+    private final String type;
+    private final BaseConfig config;
+
+    @JsonCreator
+    public ClusterEndpointConfigProviderDefinition(String type,
+                                                   @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.EXTERNAL_PROPERTY, property = "type") @JsonTypeIdResolver(ClusterEndpointConfigProviderTypeIdResolver.class) BaseConfig config) {
+        this.type = type;
+        this.config = config;
+    }
+
+    public String type() {
+        return type;
+    }
+
+    public BaseConfig config() {
+        return config;
+    }
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderTypeIdResolver.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderTypeIdResolver.java
new file mode 100644
index 00000000..1e012c4f
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderTypeIdResolver.java
@@ -0,0 +1,45 @@
+/*
+ * 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.config;
+
+import java.io.IOException;
+
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.fasterxml.jackson.databind.DatabindContext;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.jsontype.impl.TypeIdResolverBase;
+
+import io.kroxylicious.proxy.internal.clusterendpointprovider.ClusterEndpointConfigProviderContributorManager;
+
+public class ClusterEndpointConfigProviderTypeIdResolver extends TypeIdResolverBase {
+    private JavaType superType;
+
+    @Override
+    public void init(JavaType baseType) {
+        superType = baseType;
+    }
+
+    @Override
+    public String idFromValue(Object value) {
+        throw new UnsupportedOperationException("Not implemented");
+    }
+
+    @Override
+    public String idFromValueAndType(Object value, Class<?> suggestedType) {
+        throw new UnsupportedOperationException("Not implemented");
+    }
+
+    @Override
+    public JsonTypeInfo.Id getMechanism() {
+        throw new UnsupportedOperationException("Not implemented");
+    }
+
+    @Override
+    public JavaType typeFromId(DatabindContext context, String id) throws IOException {
+        Class<?> subType = ClusterEndpointConfigProviderContributorManager.getInstance().getConfigType(id);
+        return context.constructSpecializedType(superType, subType);
+    }
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/Configuration.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/Configuration.java
index b4a31338..26332189 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/Configuration.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/Configuration.java
@@ -12,35 +12,27 @@ import io.kroxylicious.proxy.config.admin.AdminHttpConfiguration;
 
 public class Configuration {
 
-    private final DefaultProxyConfig proxy;
-
     private final AdminHttpConfiguration adminHttp;
-    private final Map<String, Cluster> clusters;
+    private final Map<String, VirtualCluster> virtualClusters;
     private final List<FilterDefinition> filters;
     private final List<MicrometerDefinition> micrometer;
 
-    public Configuration(DefaultProxyConfig proxy,
-                         AdminHttpConfiguration adminHttp,
-                         Map<String, Cluster> clusters,
+    public Configuration(AdminHttpConfiguration adminHttp,
+                         Map<String, VirtualCluster> virtualClusters,
                          List<FilterDefinition> filters,
                          List<MicrometerDefinition> micrometer) {
-        this.proxy = proxy;
         this.adminHttp = adminHttp;
-        this.clusters = clusters;
+        this.virtualClusters = virtualClusters;
         this.filters = filters;
         this.micrometer = micrometer;
     }
 
-    public DefaultProxyConfig proxy() {
-        return proxy;
-    }
-
     public AdminHttpConfiguration adminHttpConfig() {
         return adminHttp;
     }
 
-    public Map<String, Cluster> clusters() {
-        return clusters;
+    public Map<String, VirtualCluster> virtualClusters() {
+        return virtualClusters;
     }
 
     public List<FilterDefinition> filters() {
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/DefaultProxyConfig.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/DefaultProxyConfig.java
deleted file mode 100644
index 6cfdd4fa..00000000
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/DefaultProxyConfig.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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.config;
-
-import java.util.Optional;
-
-public class DefaultProxyConfig implements ProxyConfig {
-
-    private final String address;
-    private final boolean logNetwork;
-    private final boolean logFrames;
-    private final boolean useIoUring;
-    private final Optional<String> keyStoreFile;
-    private final Optional<String> keyPassword;
-
-    public DefaultProxyConfig(String address, boolean logNetwork, boolean logFrames, boolean useIoUring, Optional<String> keyStoreFile, Optional<String> keyPassword) {
-        this.address = address;
-        this.logNetwork = logNetwork;
-        this.logFrames = logFrames;
-        this.useIoUring = useIoUring;
-        this.keyStoreFile = keyStoreFile;
-        this.keyPassword = keyPassword;
-    }
-
-    @Override
-    public String address() {
-        return address;
-    }
-
-    public boolean logNetwork() {
-        return logNetwork;
-    }
-
-    public boolean logFrames() {
-        return logFrames;
-    }
-
-    public boolean useIoUring() {
-        return useIoUring;
-    }
-
-    public Optional<String> keyStoreFile() {
-        return keyStoreFile;
-    }
-
-    public Optional<String> keyPassword() {
-        return keyPassword;
-    }
-}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/TargetCluster.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/TargetCluster.java
new file mode 100644
index 00000000..e4261524
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/TargetCluster.java
@@ -0,0 +1,35 @@
+/*
+ * 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.config;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Represents the target (upstream) kafka cluster.
+ */
+public class TargetCluster {
+
+    private final String bootstrapServers;
+
+    public TargetCluster(@JsonProperty(value = "bootstrap_servers") String bootstrapServers) {
+        this.bootstrapServers = bootstrapServers;
+    }
+
+    /**
+     * A list of host/port pairs to use for establishing the initial connection to the target (upstream) Kafka cluster.
+     * This list should be in the form host1:port1,host2:port2,...
+     *
+     * @return comma separated list of bootstrap servers.
+     */
+    public String bootstrapServers() {
+        return bootstrapServers;
+    }
+
+    @Override
+    public String toString() {
+        return "TargetCluster [bootstrapServers=" + bootstrapServers + "]";
+    }
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/VirtualCluster.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/VirtualCluster.java
new file mode 100644
index 00000000..dd85bf40
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/VirtualCluster.java
@@ -0,0 +1,77 @@
+/*
+ * 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.config;
+
+import java.util.Optional;
+
+public class VirtualCluster {
+
+    private final TargetCluster targetCluster;
+    private final ClusterEndpointConfigProviderDefinition clusterEndpointConfigProvider;
+
+    private final Optional<String> keyStoreFile;
+    private final Optional<String> keyPassword;
+
+    private final boolean logNetwork;
+
+    private final boolean logFrames;
+    private final boolean useIoUring;
+
+    public VirtualCluster(TargetCluster targetCluster, ClusterEndpointConfigProviderDefinition clusterEndpointConfigProvider, Optional<String> keyStoreFile,
+                          Optional<String> keyPassword,
+                          boolean logNetwork, boolean logFrames, boolean useIoUring) {
+        this.targetCluster = targetCluster;
+        this.clusterEndpointConfigProvider = clusterEndpointConfigProvider;
+        this.logNetwork = logNetwork;
+        this.logFrames = logFrames;
+        this.useIoUring = useIoUring;
+        this.keyStoreFile = keyStoreFile;
+        this.keyPassword = keyPassword;
+    }
+
+    public TargetCluster targetCluster() {
+        return targetCluster;
+    }
+
+    public ClusterEndpointConfigProviderDefinition clusterEndpointProvider() {
+        return clusterEndpointConfigProvider;
+    }
+
+    public Optional<String> keyStoreFile() {
+        return keyStoreFile;
+    }
+
+    public Optional<String> keyPassword() {
+        return keyPassword;
+    }
+
+    public boolean isLogNetwork() {
+        return logNetwork;
+    }
+
+    public boolean isLogFrames() {
+        return logFrames;
+    }
+
+    public boolean isUseIoUring() {
+        return useIoUring;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder("VirtualCluster [");
+        sb.append("targetCluster=").append(targetCluster);
+        sb.append(", clusterEndpointProvider=").append(clusterEndpointConfigProvider);
+        sb.append(", keyStoreFile=").append(keyStoreFile);
+        sb.append(", keyPassword=").append(keyPassword);
+        sb.append(", logNetwork=").append(logNetwork);
+        sb.append(", logFrames=").append(logFrames);
+        sb.append(", useIoUring=").append(useIoUring);
+        sb.append(']');
+        return sb.toString();
+    }
+
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/NetFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/NetFilter.java
index 749aa1b6..ace5d1f8 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/NetFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/NetFilter.java
@@ -43,6 +43,11 @@ public interface NetFilter {
          */
         SocketAddress srcAddress();
 
+        /**
+         * @return Returns the address of the local connection.
+         */
+        SocketAddress localAddress();
+
         /**
          * The authorized id, or null if there is no authentication configured for this listener.
          * @return
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 294b0b32..aee3fd18 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
@@ -445,6 +445,11 @@ public class KafkaProxyFrontendHandler
         return inboundCtx.channel().remoteAddress();
     }
 
+    @Override
+    public SocketAddress localAddress() {
+        return inboundCtx.channel().localAddress();
+    }
+
     @Override
     public String authorizedId() {
         return authentication != null ? authentication.authorizationId() : null;
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/MeterRegistries.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/MeterRegistries.java
index 6bf32c16..cb95c8ee 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/MeterRegistries.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/MeterRegistries.java
@@ -14,24 +14,23 @@ import io.micrometer.prometheus.PrometheusConfig;
 import io.micrometer.prometheus.PrometheusMeterRegistry;
 
 import io.kroxylicious.proxy.config.MicrometerDefinition;
-import io.kroxylicious.proxy.config.ProxyConfig;
 import io.kroxylicious.proxy.micrometer.MicrometerConfigurationHookContributorManager;
 
 public class MeterRegistries {
     private final PrometheusMeterRegistry prometheusMeterRegistry;
 
-    public MeterRegistries(List<MicrometerDefinition> micrometerConfig, ProxyConfig proxyConfig) {
-        configureMicrometer(micrometerConfig, proxyConfig);
+    public MeterRegistries(List<MicrometerDefinition> micrometerConfig) {
+        configureMicrometer(micrometerConfig);
         this.prometheusMeterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
         Metrics.addRegistry(prometheusMeterRegistry);
     }
 
-    private void configureMicrometer(List<MicrometerDefinition> micrometerConfig, ProxyConfig proxyConfig) {
+    private void configureMicrometer(List<MicrometerDefinition> micrometerConfig) {
         CompositeMeterRegistry globalRegistry = Metrics.globalRegistry;
         MicrometerConfigurationHookContributorManager manager = MicrometerConfigurationHookContributorManager.getInstance();
         micrometerConfig
                 .stream()
-                .map(f -> manager.getHook(f.type(), proxyConfig, f.config()))
+                .map(f -> manager.getHook(f.type(), f.config()))
                 .forEach(micrometerConfigurationHook -> micrometerConfigurationHook.configure(globalRegistry));
     }
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/BuiltinClusterEndpointProviderContributor.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/BuiltinClusterEndpointProviderContributor.java
new file mode 100644
index 00000000..9182e0b8
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/BuiltinClusterEndpointProviderContributor.java
@@ -0,0 +1,21 @@
+/*
+ * 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.clusterendpointprovider;
+
+import io.kroxylicious.proxy.clusterendpointprovider.ClusterEndpointProviderContributor;
+import io.kroxylicious.proxy.internal.clusterendpointprovider.StaticClusterEndpointConfigProvider.StaticClusterEndpointProviderConfig;
+import io.kroxylicious.proxy.service.BaseContributor;
+import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
+
+public class BuiltinClusterEndpointProviderContributor extends BaseContributor<ClusterEndpointConfigProvider> implements ClusterEndpointProviderContributor {
+
+    public static final BaseContributorBuilder<ClusterEndpointConfigProvider> FILTERS = BaseContributor.<ClusterEndpointConfigProvider> builder()
+            .add("StaticCluster", StaticClusterEndpointProviderConfig.class, StaticClusterEndpointConfigProvider::new);
+
+    public BuiltinClusterEndpointProviderContributor() {
+        super(FILTERS);
+    }
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/ClusterEndpointConfigProviderContributorManager.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/ClusterEndpointConfigProviderContributorManager.java
new file mode 100644
index 00000000..e8adb887
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/ClusterEndpointConfigProviderContributorManager.java
@@ -0,0 +1,52 @@
+/*
+ * 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.clusterendpointprovider;
+
+import java.util.Iterator;
+import java.util.ServiceLoader;
+
+import io.kroxylicious.proxy.clusterendpointprovider.ClusterEndpointProviderContributor;
+import io.kroxylicious.proxy.config.BaseConfig;
+import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
+
+public class ClusterEndpointConfigProviderContributorManager {
+
+    private static final ClusterEndpointConfigProviderContributorManager INSTANCE = new ClusterEndpointConfigProviderContributorManager();
+
+    private final ServiceLoader<ClusterEndpointProviderContributor> contributors;
+
+    private ClusterEndpointConfigProviderContributorManager() {
+        this.contributors = ServiceLoader.load(ClusterEndpointProviderContributor.class);
+    }
+
+    public static ClusterEndpointConfigProviderContributorManager getInstance() {
+        return INSTANCE;
+    }
+
+    public Class<? extends BaseConfig> getConfigType(String shortName) {
+        Iterator<ClusterEndpointProviderContributor> it = contributors.iterator();
+        while (it.hasNext()) {
+            ClusterEndpointProviderContributor contributor = it.next();
+            Class<? extends BaseConfig> configType = contributor.getConfigType(shortName);
+            if (configType != null) {
+                return configType;
+            }
+        }
+
+        throw new IllegalArgumentException("No endpoint provider found for name '" + shortName + "'");
+    }
+
+    public ClusterEndpointConfigProvider getClusterEndpointConfigProvider(String shortName, BaseConfig baseConfig) {
+        for (ClusterEndpointProviderContributor contributor : contributors) {
+            var assigner = contributor.getInstance(shortName, null, baseConfig);
+            if (assigner != null) {
+                return assigner;
+            }
+        }
+
+        throw new IllegalArgumentException("No endpoint provider found for name '" + shortName + "'");
+    }
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/StaticClusterEndpointConfigProvider.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/StaticClusterEndpointConfigProvider.java
new file mode 100644
index 00000000..8e14cd15
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/StaticClusterEndpointConfigProvider.java
@@ -0,0 +1,89 @@
+/*
+ * 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.clusterendpointprovider;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import io.kroxylicious.proxy.config.BaseConfig;
+import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
+import io.kroxylicious.proxy.service.HostPort;
+
+public class StaticClusterEndpointConfigProvider implements ClusterEndpointConfigProvider {
+
+    private static final EndpointMatchResult BOOTSTRAP_MATCHED = new EndpointMatchResult(true, null);
+    private static final EndpointMatchResult NO_MATCH = new EndpointMatchResult(false, null);
+    private final HostPort bootstrapAddress;
+    private final Map<Integer, HostPort> brokers;
+    private final Map<Integer, EndpointMatchResult> portToNodeIdMatchedMap;
+
+    public StaticClusterEndpointConfigProvider(StaticClusterEndpointProviderConfig config) {
+        this.bootstrapAddress = config.bootstrapAddress;
+        this.brokers = config.brokers;
+        this.portToNodeIdMatchedMap = this.brokers.entrySet().stream()
+                .collect(Collectors.toMap(e -> e.getValue().port(), e -> new EndpointMatchResult(true, e.getKey())));
+    }
+
+    @Override
+    public HostPort getClusterBootstrapAddress() {
+        return this.bootstrapAddress;
+    }
+
+    @Override
+    public HostPort getBrokerAddress(int nodeId) throws IllegalArgumentException {
+        var addr = this.brokers.get(nodeId);
+        if (addr == null) {
+            throw new IllegalArgumentException("No broker address known for nodeId %d".formatted(nodeId));
+        }
+        return addr;
+    }
+
+    @Override
+    public int getNumberOfBrokerEndpointsToPrebind() {
+        return this.brokers.size();
+    }
+
+    @Override
+    public EndpointMatchResult hasMatchingEndpoint(String sniHostname, int port) {
+        if (bootstrapAddress.port() == port) {
+            return BOOTSTRAP_MATCHED;
+        }
+        return portToNodeIdMatchedMap.getOrDefault(port, NO_MATCH);
+    }
+
+    public static class StaticClusterEndpointProviderConfig extends BaseConfig {
+        private final HostPort bootstrapAddress;
+        private final Map<Integer, HostPort> brokers;
+
+        public StaticClusterEndpointProviderConfig(HostPort bootstrapAddress, Map<Integer, HostPort> brokers) {
+            this.bootstrapAddress = bootstrapAddress;
+            if (brokers == null) {
+                this.brokers = Map.of(0, this.bootstrapAddress);
+            }
+            else {
+                if (brokers.isEmpty()) {
+                    throw new IllegalArgumentException("requires non-empty map of nodeid to broker address mappings");
+                }
+
+                var duplicateBrokerAddresses = brokers.values().stream().filter(e -> Collections.frequency(brokers.values(), e) > 1).distinct().toList();
+                if (!duplicateBrokerAddresses.isEmpty()) {
+                    throw new IllegalArgumentException("all broker addresses must be unique (found duplicates: %s)".formatted(
+                            duplicateBrokerAddresses.stream().map(HostPort::toString).collect(Collectors.joining(","))));
+                }
+                var allPorts = brokers.values().stream().map(HostPort::port).toList();
+                var duplicatePorts = allPorts.stream().filter(e -> Collections.frequency(allPorts, e) > 1).distinct().toList();
+                if (!duplicatePorts.isEmpty()) {
+                    throw new IllegalArgumentException("all broker addresses must have unique ports (found duplicates: %s)".formatted(
+                            brokers.values().stream().filter(hp -> duplicatePorts.contains(hp.port())).map(HostPort::toString).collect(Collectors.joining(","))));
+                }
+                this.brokers = brokers;
+            }
+        }
+    }
+
+}
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
deleted file mode 100644
index bec7c0f6..00000000
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/AddressMapping.java
+++ /dev/null
@@ -1,15 +0,0 @@
-/*
- * 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 432510c5..1c67af5b 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,7 +5,6 @@
  */
 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;
@@ -21,12 +20,11 @@ import org.apache.kafka.common.message.ResponseHeaderData;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.kroxylicious.proxy.config.BaseConfig;
-import io.kroxylicious.proxy.config.ProxyConfig;
 import io.kroxylicious.proxy.filter.DescribeClusterResponseFilter;
 import io.kroxylicious.proxy.filter.FindCoordinatorResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.filter.MetadataResponseFilter;
+import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
 
 /**
  * A filter that rewrites broker addresses in all relevant responses to the corresponding proxy address.
@@ -35,41 +33,17 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
 
     private static final Logger LOGGER = LoggerFactory.getLogger(BrokerAddressFilter.class);
 
-    public static class BrokerAddressConfig extends BaseConfig {
+    private final ClusterEndpointConfigProvider clusterEndpointConfigProvider;
 
-        private final Class<? extends AddressMapping> addressMapperClazz;
-
-        public BrokerAddressConfig(String addressMapper) {
-            try {
-                this.addressMapperClazz = addressMapper == null ? null : (Class<? extends AddressMapping>) Class.forName(addressMapper);
-            }
-            catch (ClassNotFoundException e) {
-                throw new RuntimeException(e);
-            }
-        }
-
-        public Class<? extends AddressMapping> addressMapper() {
-            return addressMapperClazz;
-        }
-    }
-
-    private final AddressMapping mapping;
-
-    public BrokerAddressFilter(ProxyConfig all, BrokerAddressConfig config) {
-
-        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);
-        }
+    public BrokerAddressFilter(ClusterEndpointConfigProvider clusterEndpointConfigProvider) {
+        this.clusterEndpointConfigProvider = clusterEndpointConfigProvider;
     }
 
     @Override
     public void onMetadataResponse(ResponseHeaderData header, MetadataResponseData data, KrpcFilterContext context) {
         for (MetadataResponseBroker broker : data.brokers()) {
-            apply(context, broker, MetadataResponseBroker::host, MetadataResponseBroker::port, MetadataResponseBroker::setHost, MetadataResponseBroker::setPort);
+            apply(context, broker, MetadataResponseBroker::nodeId, MetadataResponseBroker::host, MetadataResponseBroker::port, MetadataResponseBroker::setHost,
+                    MetadataResponseBroker::setPort);
         }
         context.forwardResponse(data);
     }
@@ -77,7 +51,8 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
     @Override
     public void onDescribeClusterResponse(ResponseHeaderData header, DescribeClusterResponseData data, KrpcFilterContext context) {
         for (DescribeClusterBroker broker : data.brokers()) {
-            apply(context, broker, DescribeClusterBroker::host, DescribeClusterBroker::port, DescribeClusterBroker::setHost, DescribeClusterBroker::setPort);
+            apply(context, broker, DescribeClusterBroker::brokerId, DescribeClusterBroker::host, DescribeClusterBroker::port, DescribeClusterBroker::setHost,
+                    DescribeClusterBroker::setPort);
         }
         context.forwardResponse(data);
     }
@@ -85,21 +60,21 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
     @Override
     public void onFindCoordinatorResponse(ResponseHeaderData header, FindCoordinatorResponseData data, KrpcFilterContext context) {
         for (Coordinator coordinator : data.coordinators()) {
-            apply(context, coordinator, Coordinator::host, Coordinator::port, Coordinator::setHost, Coordinator::setPort);
+            apply(context, coordinator, Coordinator::nodeId, Coordinator::host, Coordinator::port, Coordinator::setHost, Coordinator::setPort);
         }
         context.forwardResponse(data);
     }
 
-    private <T> void apply(KrpcFilterContext context, T broker, Function<T, String> hostGetter, ToIntFunction<T> portGetter, BiConsumer<T, String> hostSetter,
+    private <T> void apply(KrpcFilterContext context, T broker, Function<T, Integer> nodeIdGetter, Function<T, String> hostGetter, ToIntFunction<T> portGetter,
+                           BiConsumer<T, String> hostSetter,
                            ObjIntConsumer<T> portSetter) {
         String incomingHost = hostGetter.apply(broker);
         int incomingPort = portGetter.applyAsInt(broker);
 
-        String host = mapping.downstreamHost(context, incomingHost, incomingPort);
-        int port = mapping.downstreamPort(context, incomingHost, incomingPort);
+        var downstreamAddress = clusterEndpointConfigProvider.getBrokerAddress(nodeIdGetter.apply(broker));
 
-        LOGGER.trace("{}: Rewriting broker address in response {}:{} -> {}:{}", context, incomingHost, incomingPort, host, port);
-        hostSetter.accept(broker, host);
-        portSetter.accept(broker, port);
+        LOGGER.trace("{}: Rewriting broker address in response {}:{} -> {}", context, incomingHost, incomingPort, downstreamAddress);
+        hostSetter.accept(broker, downstreamAddress.host());
+        portSetter.accept(broker, downstreamAddress.port());
     }
 }
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 c11ca4a2..3dbdb21c 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
@@ -7,7 +7,6 @@ package io.kroxylicious.proxy.internal.filter;
 
 import io.kroxylicious.proxy.filter.FilterContributor;
 import io.kroxylicious.proxy.filter.KrpcFilter;
-import io.kroxylicious.proxy.internal.filter.BrokerAddressFilter.BrokerAddressConfig;
 import io.kroxylicious.proxy.internal.filter.FetchResponseTransformationFilter.FetchResponseTransformationConfig;
 import io.kroxylicious.proxy.internal.filter.ProduceRequestTransformationFilter.ProduceRequestTransformationConfig;
 import io.kroxylicious.proxy.service.BaseContributor;
@@ -16,7 +15,7 @@ public class BuiltinFilterContributor extends BaseContributor<KrpcFilter> implem
 
     public static final BaseContributorBuilder<KrpcFilter> FILTERS = BaseContributor.<KrpcFilter> builder()
             .add("ApiVersions", ApiVersionsFilter::new)
-            .add("BrokerAddress", BrokerAddressConfig.class, BrokerAddressFilter::new)
+            .add("BrokerAddress", BrokerAddressFilter::new)
             .add("ProduceRequestTransformation", ProduceRequestTransformationConfig.class, ProduceRequestTransformationFilter::new)
             .add("FetchResponseTransformation", FetchResponseTransformationConfig.class, FetchResponseTransformationFilter::new);
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FilterContributorManager.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FilterContributorManager.java
index aa50bd00..be57ec58 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FilterContributorManager.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FilterContributorManager.java
@@ -9,9 +9,9 @@ import java.util.Iterator;
 import java.util.ServiceLoader;
 
 import io.kroxylicious.proxy.config.BaseConfig;
-import io.kroxylicious.proxy.config.ProxyConfig;
 import io.kroxylicious.proxy.filter.FilterContributor;
 import io.kroxylicious.proxy.filter.KrpcFilter;
+import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
 
 public class FilterContributorManager {
 
@@ -40,7 +40,7 @@ public class FilterContributorManager {
         throw new IllegalArgumentException("No filter found for name '" + shortName + "'");
     }
 
-    public KrpcFilter getFilter(String shortName, ProxyConfig proxyConfig, BaseConfig filterConfig) {
+    public KrpcFilter getFilter(String shortName, ClusterEndpointConfigProvider proxyConfig, BaseConfig filterConfig) {
         Iterator<FilterContributor> it = contributors.iterator();
         while (it.hasNext()) {
             FilterContributor contributor = it.next();
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
deleted file mode 100644
index ad5c77b6..00000000
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FixedAddressMapping.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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;
-
-    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
deleted file mode 100644
index 4d29dbf0..00000000
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/SniAddressMapping.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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;
-
-    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/java/io/kroxylicious/proxy/internal/filter/UpstreamBrokerAddressCachingNetFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/UpstreamBrokerAddressCachingNetFilter.java
new file mode 100644
index 00000000..ac352495
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/UpstreamBrokerAddressCachingNetFilter.java
@@ -0,0 +1,90 @@
+/*
+ * 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 java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.kroxylicious.proxy.bootstrap.FilterChainFactory;
+import io.kroxylicious.proxy.filter.KrpcFilter;
+import io.kroxylicious.proxy.filter.MetadataResponseFilter;
+import io.kroxylicious.proxy.filter.NetFilter;
+import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
+import io.kroxylicious.proxy.service.HostPort;
+
+/**
+ * Temporary implementation of NetFilter that uses the port number from the local end of the
+ * downstream connection to identify the virtual clusters broker.  It then uses this to identify
+ * the upstream broker address from a cache built from a previous connection's metadata response.
+ */
+public class UpstreamBrokerAddressCachingNetFilter implements NetFilter {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(UpstreamBrokerAddressCachingNetFilter.class);
+
+    private final HostPort targetClusterBootstrap;
+    private final FilterChainFactory filterChainFactory;
+    private final ClusterEndpointConfigProvider endpointConfigProvider;
+
+    private final Map<Integer, HostPort> upstreamBrokers = new ConcurrentHashMap<>();
+
+    public UpstreamBrokerAddressCachingNetFilter(HostPort targetClusterBootstrap, FilterChainFactory filterChainFactory,
+                                                 ClusterEndpointConfigProvider endpointConfigProvider) {
+        this.targetClusterBootstrap = targetClusterBootstrap;
+        this.filterChainFactory = filterChainFactory;
+        this.endpointConfigProvider = endpointConfigProvider;
+    }
+
+    @Override
+    public void selectServer(NetFilterContext context) {
+        var filters = new ArrayList<>(Arrays.stream(filterChainFactory.createFilters()).toList());
+
+        // Add a filter to the *end of the chain* that gathers the true nodeId/upstream broker mapping.
+        filters.add((MetadataResponseFilter) (header, response, filterContext) -> {
+            response.brokers().forEach(b -> {
+                var replacement = new HostPort(b.host(), b.port());
+                var existing = upstreamBrokers.put(b.nodeId(), replacement);
+                if (!replacement.equals(existing)) {
+                    LOGGER.info("Got upstream for broker {} : {}", b.nodeId(), replacement);
+                }
+            });
+            filterContext.forwardResponse(response);
+        });
+
+        var targetPort = ((InetSocketAddress) context.localAddress()).getPort();
+        var endpointMatchResult = endpointConfigProvider.hasMatchingEndpoint(context.sniHostname(), targetPort);
+
+        HostPort target;
+        if (endpointMatchResult.matched() && endpointMatchResult.nodeId() != null) {
+            var upstreamBroker = upstreamBrokers.get(endpointMatchResult.nodeId());
+            if (upstreamBroker != null) {
+                target = upstreamBroker;
+            }
+            else {
+                // TODO: this behaviour is sub-optimal as it means a client will proceed with a connection to the wrong broker.
+                // This will lead to difficult to diagnose failure cases later (produces going to the wrong broker, metadata refresh cycles, etc).
+                LOGGER.warn("An upstream address for broker {} is not yet known, connecting the client to bootstrap instead.", endpointMatchResult.nodeId());
+                target = targetClusterBootstrap;
+            }
+        }
+        else if (endpointMatchResult.matched()) {
+            target = targetClusterBootstrap;
+        }
+        else {
+            throw new RuntimeException(
+                    "Connection to %s:%d cannot be routed to an upstream endpoint".formatted(context.sniHostname() == null ? "" : context.sniHostname(),
+                            targetPort));
+        }
+
+        context.initiateConnect(target.host(), target.port(), filters.toArray(new KrpcFilter[0]));
+    }
+
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/micrometer/MicrometerConfigurationHookContributorManager.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/micrometer/MicrometerConfigurationHookContributorManager.java
index aed1f458..f148a65e 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/micrometer/MicrometerConfigurationHookContributorManager.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/micrometer/MicrometerConfigurationHookContributorManager.java
@@ -8,7 +8,6 @@ package io.kroxylicious.proxy.micrometer;
 import java.util.ServiceLoader;
 
 import io.kroxylicious.proxy.config.BaseConfig;
-import io.kroxylicious.proxy.config.ProxyConfig;
 
 public class MicrometerConfigurationHookContributorManager {
 
@@ -35,9 +34,9 @@ public class MicrometerConfigurationHookContributorManager {
         throw new IllegalArgumentException("No micrometer configuration hook found for name '" + shortName + "'");
     }
 
-    public MicrometerConfigurationHook getHook(String shortName, ProxyConfig proxyConfig, BaseConfig filterConfig) {
+    public MicrometerConfigurationHook getHook(String shortName, BaseConfig filterConfig) {
         for (MicrometerConfigurationHookContributor contributor : contributors) {
-            MicrometerConfigurationHook hook = contributor.getInstance(shortName, proxyConfig, filterConfig);
+            MicrometerConfigurationHook hook = contributor.getInstance(shortName, null, filterConfig);
             if (hook != null) {
                 return hook;
             }
diff --git a/kroxylicious/src/main/resources/META-INF/services/io.kroxylicious.proxy.clusterendpointprovider.ClusterEndpointProviderContributor b/kroxylicious/src/main/resources/META-INF/services/io.kroxylicious.proxy.clusterendpointprovider.ClusterEndpointProviderContributor
new file mode 100644
index 00000000..9b78922e
--- /dev/null
+++ b/kroxylicious/src/main/resources/META-INF/services/io.kroxylicious.proxy.clusterendpointprovider.ClusterEndpointProviderContributor
@@ -0,0 +1,1 @@
+io.kroxylicious.proxy.internal.clusterendpointprovider.BuiltinClusterEndpointProviderContributor
diff --git a/pom.xml b/pom.xml
index db4d6bec..89e47955 100644
--- a/pom.xml
+++ b/pom.xml
@@ -46,9 +46,9 @@
         <netty.kqueue.classifier>osx-x86_64</netty.kqueue.classifier>
         <lombok.version>1.18.26</lombok.version>
         <assertj-core.version>3.24.2</assertj-core.version>
-        <mockito.version>5.2.0</mockito.version>
+        <mockito.version>5.3.0</mockito.version>
         <slf4j.version>2.0.7</slf4j.version>
-        <micrometer.version>1.10.5</micrometer.version>
+        <micrometer.version>1.10.6</micrometer.version>
         <kroxy.extension.version>0.1</kroxy.extension.version>
         <zjsonpatch.version>0.4.14</zjsonpatch.version>
         <sundr-builder-annotations.version>0.94.0</sundr-builder-annotations.version>


Test patch:
diff --git a/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java b/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java
index 48719711..d0f505b3 100644
--- a/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java
+++ b/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java
@@ -8,10 +8,10 @@ package io.kroxylicious.proxy.service;
 import org.junit.jupiter.api.Test;
 
 import io.kroxylicious.proxy.config.BaseConfig;
-import io.kroxylicious.proxy.config.ProxyConfig;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
 
 class BaseContributorTest {
     static class LongConfig extends BaseConfig {
@@ -22,13 +22,6 @@ class BaseContributorTest {
     static class AnotherConfig extends BaseConfig {
     }
 
-    static class LongProxyConfig implements ProxyConfig {
-        @Override
-        public String address() {
-            return "3";
-        }
-    }
-
     @Test
     public void testDefaultConfigClass() {
         BaseContributor.BaseContributorBuilder<Long> builder = BaseContributor.builder();
@@ -45,7 +38,7 @@ class BaseContributorTest {
         builder.add("one", () -> 1L);
         BaseContributor<Long> baseContributor = new BaseContributor<>(builder) {
         };
-        Long instance = baseContributor.getInstance("one", new LongProxyConfig(), new BaseConfig());
+        Long instance = baseContributor.getInstance("one", mock(ClusterEndpointConfigProvider.class), new BaseConfig());
         assertThat(instance).isEqualTo(1L);
     }
 
@@ -65,7 +58,7 @@ class BaseContributorTest {
         builder.add("fromBaseConfig", LongConfig.class, baseConfig -> baseConfig.value);
         BaseContributor<Long> baseContributor = new BaseContributor<>(builder) {
         };
-        Long instance = baseContributor.getInstance("fromBaseConfig", new LongProxyConfig(), new LongConfig());
+        Long instance = baseContributor.getInstance("fromBaseConfig", mock(ClusterEndpointConfigProvider.class), new LongConfig());
         assertThat(instance).isEqualTo(2L);
     }
 
@@ -77,18 +70,8 @@ class BaseContributorTest {
         };
         AnotherConfig incompatibleConfig = new AnotherConfig();
         assertThatThrownBy(() -> {
-            baseContributor.getInstance("fromBaseConfig", new LongProxyConfig(), incompatibleConfig);
+            baseContributor.getInstance("fromBaseConfig", mock(ClusterEndpointConfigProvider.class), incompatibleConfig);
         }).isInstanceOf(IllegalArgumentException.class);
     }
 
-    @Test
-    public void testUsingProxyConfig() {
-        BaseContributor.BaseContributorBuilder<Long> builder = BaseContributor.builder();
-        builder.add("fromProxyConfig", LongConfig.class, (proxyConfig, longConfig) -> Long.valueOf(proxyConfig.address()));
-        BaseContributor<Long> baseContributor = new BaseContributor<>(builder) {
-        };
-        Long instance = baseContributor.getInstance("fromProxyConfig", new LongProxyConfig(), new LongConfig());
-        assertThat(instance).isEqualTo(3L);
-    }
-
 }
\ No newline at end of file
diff --git a/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/HostPortTest.java b/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/HostPortTest.java
new file mode 100644
index 00000000..fb1d89fa
--- /dev/null
+++ b/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/HostPortTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.service;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class HostPortTest {
+
+    @Test
+    public void validBareHost() {
+        var hp = HostPort.parse("localhost:12345");
+        assertThat(hp.host()).isEqualTo("localhost");
+        assertThat(hp.port()).isEqualTo(12345);
+    }
+
+    @Test
+    public void validFQDN() {
+        var hp = HostPort.parse("kafka.example.com:12345");
+        assertThat(hp.host()).isEqualTo("kafka.example.com");
+        assertThat(hp.port()).isEqualTo(12345);
+    }
+
+    @Test
+    public void validIpv4() {
+        var hp = HostPort.parse("192.168.0.1:12345");
+        assertThat(hp.host()).isEqualTo("192.168.0.1");
+        assertThat(hp.port()).isEqualTo(12345);
+    }
+
+    @Test
+    public void validIpv6() {
+        var hp = HostPort.parse("[2001:db8::1]:12345");
+        assertThat(hp.host()).isEqualTo("[2001:db8::1]");
+        assertThat(hp.port()).isEqualTo(12345);
+    }
+
+    @ParameterizedTest
+    @NullAndEmptySource
+    @ValueSource(strings = { " ", "localhost", ":1000", ":onethousand", "something:really:odd" })
+    public void shouldThrowExceptionForMalformedInput(String input) {
+        assertThatThrownBy(() -> {
+            HostPort.parse(input);
+        }).isInstanceOf(IllegalArgumentException.class);
+    }
+}
\ No newline at end of file
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/ClusterEndpointConfigProvider.java b/integrationtests/src/test/java/io/kroxylicious/proxy/ClusterEndpointConfigProvider.java
new file mode 100644
index 00000000..85df3baa
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/ClusterEndpointConfigProvider.java
@@ -0,0 +1,19 @@
+/*
+ * 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;
+
+import java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+import io.sundr.builder.annotations.Buildable;
+
+import static com.fasterxml.jackson.annotation.JsonInclude.Include.NON_EMPTY;
+
+@Buildable(editableEnabled = false)
+public record ClusterEndpointConfigProvider(String type, @JsonInclude(NON_EMPTY) Map<String, Object> config) {
+}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyConfig.java b/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyConfig.java
index 397ab005..827f6331 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyConfig.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyConfig.java
@@ -24,8 +24,7 @@ public class KroxyConfig {
 
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(new YAMLFactory().disable(YAMLGenerator.Feature.WRITE_DOC_START_MARKER));
 
-    private final Proxy proxy;
-    private final Map<String, Cluster> clusters;
+    private final Map<String, VirtualCluster> virtualClusters;
     private final List<Filter> filters;
 
     @JsonInclude(NON_NULL)
@@ -38,9 +37,8 @@ public class KroxyConfig {
         return new KroxyConfigBuilder();
     }
 
-    public KroxyConfig(Proxy proxy, Map<String, Cluster> clusters, List<Filter> filters, AdminHttp adminHttp, List<MicrometerConfig> micrometer) {
-        this.proxy = proxy;
-        this.clusters = clusters;
+    public KroxyConfig(Map<String, VirtualCluster> virtualClusters, List<Filter> filters, AdminHttp adminHttp, List<MicrometerConfig> micrometer) {
+        this.virtualClusters = virtualClusters;
         this.filters = filters;
         this.adminHttp = adminHttp;
         this.micrometer = micrometer;
@@ -55,16 +53,12 @@ public class KroxyConfig {
         }
     }
 
-    public Proxy getProxy() {
-        return proxy;
-    }
-
     public AdminHttp getAdminHttp() {
         return adminHttp;
     }
 
-    public Map<String, Cluster> getClusters() {
-        return clusters;
+    public Map<String, VirtualCluster> getVirtualClusters() {
+        return virtualClusters;
     }
 
     public List<Filter> getFilters() {
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyConfigTest.java b/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyConfigTest.java
index 3f29d260..242cfc62 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyConfigTest.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyConfigTest.java
@@ -23,29 +23,31 @@ public class KroxyConfigTest {
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(new YAMLFactory());
 
     @Test
-    public void testBareConfig() throws Exception {
-        ObjectNode deserializedConfig = serializeAndDeserialize(builder().withNewProxy().withAddress("localhost:9192").endProxy().build());
-        ObjectNode proxyObj = assertObjectField(deserializedConfig, "proxy");
-        assertTextField(proxyObj, "address", "localhost:9192");
-    }
-
-    @Test
-    public void testSslConfig() throws Exception {
+    public void testVirtualClusterConfig() throws Exception {
         ObjectNode deserializedConfig = serializeAndDeserialize(
-                builder().withNewProxy().withAddress("localhost:9192").withKeyPassword("pass").withKeyStoreFile("file").endProxy().build());
-        ObjectNode proxyObj = assertObjectField(deserializedConfig, "proxy");
-        assertTextField(proxyObj, "address", "localhost:9192");
-        assertTextField(proxyObj, "keyStoreFile", "file");
-        assertTextField(proxyObj, "keyPassword", "pass");
+                builder().addToVirtualClusters("demo",
+                        new VirtualClusterBuilder().withNewTargetCluster().withBootstrapServers("localhost:9092").endTargetCluster().build()).build());
+        ObjectNode clusterObj = assertObjectField(deserializedConfig, "virtualClusters");
+        ObjectNode demoObj = assertObjectField(clusterObj, "demo");
+        ObjectNode targetClusterObj = assertObjectField(demoObj, "targetCluster");
+        assertTextField(targetClusterObj, "bootstrap_servers", "localhost:9092");
     }
 
     @Test
-    public void testClusterConfig() throws Exception {
+    public void testVirtualClusterConfigWithClusterEndpointProvider() throws Exception {
         ObjectNode deserializedConfig = serializeAndDeserialize(
-                builder().addToClusters("demo", new ClusterBuilder().withBootstrapServers("localhost:9092").build()).build());
-        ObjectNode clusterObj = assertObjectField(deserializedConfig, "clusters");
+                builder().addToVirtualClusters("demo", new VirtualClusterBuilder()
+                        .withNewClusterEndpointConfigProvider()
+                        .withType("providertype")
+                        .withConfig(Map.of("a", "b"))
+                        .endClusterEndpointConfigProvider()
+                        .build()).build());
+        ObjectNode clusterObj = assertObjectField(deserializedConfig, "virtualClusters");
         ObjectNode demoObj = assertObjectField(clusterObj, "demo");
-        assertTextField(demoObj, "bootstrap_servers", "localhost:9092");
+        ObjectNode endpointAssignerObj = assertObjectField(demoObj, "clusterEndpointConfigProvider");
+        assertTextField(endpointAssignerObj, "type", "providertype");
+        ObjectNode endpointAssignerConfig = assertObjectField(endpointAssignerObj, "config");
+        assertTextField(endpointAssignerConfig, "a", "b");
     }
 
     @Test
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
index 401c1a8d..d7672a65 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
@@ -12,6 +12,9 @@ import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
 
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.Admin;
@@ -23,6 +26,8 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
 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.common.Node;
+import org.apache.kafka.common.TopicPartitionInfo;
 import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
@@ -36,6 +41,7 @@ import org.junit.jupiter.api.io.TempDir;
 
 import io.kroxylicious.proxy.internal.filter.ByteBufferTransformation;
 import io.kroxylicious.testing.kafka.api.KafkaCluster;
+import io.kroxylicious.testing.kafka.common.BrokerCluster;
 import io.kroxylicious.testing.kafka.common.KeytoolCertificateGenerator;
 import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
 
@@ -44,6 +50,8 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
 
 @ExtendWith(KafkaClusterExtension.class)
 public class KrpcFilterIT {
@@ -246,13 +254,14 @@ public class KrpcFilterIT {
         brokerCertificateGenerator.generateTrustStore(brokerCertificateGenerator.getCertFilePath(), "client",
                 clientTrustStore.toAbsolutePath().toString());
 
-        var config = baseConfigBuilder(proxyAddress, cluster.getBootstrapServers())
-                .editProxy()
+        var builder = baseConfigBuilder(proxyAddress, cluster.getBootstrapServers());
+        var demo = builder.getVirtualClusters().get("demo");
+        demo = new VirtualClusterBuilder(demo)
                 .withKeyPassword(brokerCertificateGenerator.getPassword())
                 .withKeyStoreFile(brokerCertificateGenerator.getKeyStoreLocation())
-                .endProxy()
-                .build()
-                .toYaml();
+                .build();
+        builder.addToVirtualClusters("demo", demo);
+        var config = builder.build().toYaml();
 
         try (var proxy = startProxy(config)) {
             try (var admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
@@ -270,9 +279,79 @@ public class KrpcFilterIT {
         }
     }
 
+    @Test
+    public void proxyExposesClusterOfTwoBrokers(@BrokerCluster(numBrokers = 2) KafkaCluster cluster) throws Exception {
+        var proxyAddress = "localhost:9192";
+        var builder = baseConfigBuilder(proxyAddress, cluster.getBootstrapServers());
+        var demo = builder.getVirtualClusters().get("demo");
+        var brokerEndpoints = Map.of(0, "localhost:9193", 1, "localhost:9194");
+        demo = new VirtualClusterBuilder(demo)
+                .editClusterEndpointConfigProvider()
+                .withType("StaticCluster")
+                .withConfig(Map.of("bootstrapAddress", proxyAddress,
+                        "brokers", brokerEndpoints))
+                .endClusterEndpointConfigProvider()
+                .build();
+        builder.addToVirtualClusters("demo", demo);
+        var config = builder.build().toYaml();
+
+        try (var proxy = startProxy(config)) {
+
+            try (var admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress))) {
+                var nodes = admin.describeCluster().nodes().get();
+                assertThat(nodes).hasSize(2);
+                var unique = nodes.stream().collect(Collectors.toMap(Node::id, KrpcFilterIT::toAddress));
+                assertThat(unique).containsExactlyInAnyOrderEntriesOf(brokerEndpoints);
+            }
+
+            // Create a topic with one partition on each broker and publish to all of them..
+            // As kafka client must connect to the partition leader, this verifies that kroxylicious is
+            // indeed connecting to all brokers.
+            int numPartitions = 2;
+            try (var admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress))) {
+                // create topic and ensure that leaders are on different brokers.
+                admin.createTopics(List.of(new NewTopic(TOPIC_1, numPartitions, (short) 1))).all().get();
+                await().atMost(Duration.ofSeconds(5)).until(() -> admin.describeTopics(List.of(TOPIC_1)).topicNameValues().get(TOPIC_1).get()
+                        .partitions().stream().map(TopicPartitionInfo::leader)
+                        .collect(Collectors.toSet()),
+                        leaders -> leaders.size() == numPartitions);
+
+                try (var producer = new KafkaProducer<String, String>(Map.of(
+                        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress,
+                        ProducerConfig.CLIENT_ID_CONFIG, "myclient",
+                        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class))) {
+                    for (int partition = 0; partition < numPartitions; partition++) {
+                        try {
+                            var send = producer.send(new ProducerRecord<>(TOPIC_1, partition, "key", "value"));
+                            send.get(10, TimeUnit.SECONDS);
+                        }
+                        catch (TimeoutException e) {
+                            // Need to explicitly close the producer otherwise it will await forever awaiting the in-flight send to complete.
+                            producer.close(Duration.ofSeconds(1));
+                            fail("send for partition %d timed-out (check logs for root cause)".formatted(partition), e);
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    private static String toAddress(Node n) {
+        return n.host() + ":" + n.port();
+    }
+
     private static KroxyConfigBuilder baseConfigBuilder(String proxyAddress, String bootstrapServers) {
-        return KroxyConfig.builder().withNewProxy().withAddress(proxyAddress).endProxy()
-                .addToClusters("demo", new ClusterBuilder().withBootstrapServers(bootstrapServers).build())
+        return KroxyConfig.builder()
+                .addToVirtualClusters("demo", new VirtualClusterBuilder()
+                        .withNewTargetCluster()
+                        .withBootstrapServers(bootstrapServers)
+                        .endTargetCluster()
+                        .withNewClusterEndpointConfigProvider()
+                        .withType("StaticCluster")
+                        .withConfig(Map.of("bootstrapAddress", proxyAddress))
+                        .endClusterEndpointConfigProvider()
+                        .build())
                 .addNewFilter().withType("ApiVersions").endFilter()
                 .addNewFilter().withType("BrokerAddress").endFilter();
     }
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java
index 095e135b..bf60f834 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java
@@ -123,7 +123,15 @@ public class MetricsIT {
     }
 
     private static KroxyConfigBuilder baseConfigBuilder(String proxyAddress, String bootstrapServers) {
-        return KroxyConfig.builder().withNewProxy().withAddress(proxyAddress).endProxy()
-                .addToClusters("demo", new ClusterBuilder().withBootstrapServers(bootstrapServers).build());
+        return KroxyConfig.builder()
+                .addToVirtualClusters("demo", new VirtualClusterBuilder()
+                        .withNewTargetCluster()
+                        .withBootstrapServers(bootstrapServers)
+                        .endTargetCluster()
+                        .withNewClusterEndpointConfigProvider()
+                        .withType("StaticCluster")
+                        .withConfig(Map.of("bootstrapAddress", proxyAddress))
+                        .endClusterEndpointConfigProvider()
+                        .build());
     }
 }
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java b/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
index 891f34f7..0c28cccf 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
@@ -126,8 +126,16 @@ public class ProxyRpcTest {
     }
 
     private static String configure(MockServer mockServer) {
-        return KroxyConfig.builder().withNewProxy().withAddress(ProxyRpcTest.PROXY_ADDRESS).endProxy()
-                .addToClusters("demo", new ClusterBuilder().withBootstrapServers("localhost:" + mockServer.port()).build())
+        return KroxyConfig.builder()
+                .addToVirtualClusters("demo", new VirtualClusterBuilder()
+                        .withNewTargetCluster()
+                        .withBootstrapServers("localhost:" + mockServer.port())
+                        .endTargetCluster()
+                        .withNewClusterEndpointConfigProvider()
+                        .withType("StaticCluster")
+                        .withConfig(Map.of("bootstrapAddress", ProxyRpcTest.PROXY_ADDRESS))
+                        .endClusterEndpointConfigProvider()
+                        .build())
                 .addNewFilter().withType("ApiVersions").endFilter()
                 .addNewFilter().withType("FixedClientId").withConfig(Map.of("clientId", "fixed")).endFilter()
                 .build().toYaml();
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/Cluster.java b/integrationtests/src/test/java/io/kroxylicious/proxy/TargetCluster.java
similarity index 79%
rename from integrationtests/src/test/java/io/kroxylicious/proxy/Cluster.java
rename to integrationtests/src/test/java/io/kroxylicious/proxy/TargetCluster.java
index 3003e4b7..5bf66410 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/Cluster.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/TargetCluster.java
@@ -11,5 +11,5 @@ import com.fasterxml.jackson.annotation.JsonGetter;
 import io.sundr.builder.annotations.Buildable;
 
 @Buildable(editableEnabled = false)
-public record Cluster(@JsonGetter("bootstrap_servers") String bootstrapServers) {
+public record TargetCluster(@JsonGetter("bootstrap_servers") String bootstrapServers) {
 }
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/VirtualCluster.java b/integrationtests/src/test/java/io/kroxylicious/proxy/VirtualCluster.java
new file mode 100644
index 00000000..3a82a7fd
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/VirtualCluster.java
@@ -0,0 +1,13 @@
+/*
+ * 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;
+
+import io.sundr.builder.annotations.Buildable;
+
+@Buildable(editableEnabled = false)
+public record VirtualCluster(TargetCluster targetCluster, ClusterEndpointConfigProvider clusterEndpointConfigProvider, String keyStoreFile, String keyPassword, boolean logNetwork, boolean logFrames, boolean useIoUring) {
+}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
index 906873c1..1decc1fd 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
@@ -63,8 +63,8 @@ import org.junit.jupiter.params.provider.EnumSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.kroxylicious.proxy.ClusterBuilder;
 import io.kroxylicious.proxy.KroxyConfig;
+import io.kroxylicious.proxy.VirtualClusterBuilder;
 import io.kroxylicious.testing.kafka.api.KafkaCluster;
 import io.kroxylicious.testing.kafka.common.KeytoolCertificateGenerator;
 import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
@@ -92,7 +92,7 @@ public class MultiTenantIT {
 
     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 TENANT2_PROXY_ADDRESS = "bar.multitenant.kafka:9193";
     private static final String MY_KEY = "my-key";
     private static final String MY_VALUE = "my-value";
     private TestInfo testInfo;
@@ -104,6 +104,7 @@ public class MultiTenantIT {
     @BeforeEach
     public void beforeEach(TestInfo testInfo) throws Exception {
         this.testInfo = testInfo;
+        // TODO: use a per-tenant server certificate.
         this.certificateGenerator = new KeytoolCertificateGenerator();
         this.certificateGenerator.generateSelfSignedCertificateEntry("test@redhat.com", "*.multitenant.kafka", "KI", "RedHat", null, null, "US");
         this.clientTrustStore = certsDirectory.resolve("kafka.truststore.jks");
@@ -113,7 +114,7 @@ public class MultiTenantIT {
 
     @Test
     public void createAndDeleteTopic(KafkaCluster cluster) throws Exception {
-        String config = getConfig(PROXY_ADDRESS, cluster);
+        String config = getConfig(cluster);
 
         try (var proxy = startProxy(config)) {
             try (var admin = Admin.create(commonConfig(TENANT1_PROXY_ADDRESS, Map.of()))) {
@@ -146,7 +147,7 @@ public class MultiTenantIT {
 
     @Test
     public void describeTopic(KafkaCluster cluster) throws Exception {
-        String config = getConfig(PROXY_ADDRESS, cluster);
+        String config = getConfig(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));
@@ -162,7 +163,7 @@ public class MultiTenantIT {
 
     @Test
     public void publishOne(KafkaCluster cluster) throws Exception {
-        String config = getConfig(PROXY_ADDRESS, cluster);
+        String config = getConfig(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);
@@ -171,7 +172,7 @@ public class MultiTenantIT {
 
     @Test
     public void consumeOne(KafkaCluster cluster) throws Exception {
-        String config = getConfig(PROXY_ADDRESS, cluster);
+        String config = getConfig(cluster);
         try (var proxy = startProxy(config)) {
             var groupId = testInfo.getDisplayName();
             createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
@@ -182,7 +183,7 @@ public class MultiTenantIT {
 
     @Test
     public void consumeOneAndOffsetCommit(KafkaCluster cluster) throws Exception {
-        String config = getConfig(PROXY_ADDRESS, cluster);
+        String config = getConfig(cluster);
         try (var proxy = startProxy(config)) {
             var groupId = testInfo.getDisplayName();
             createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
@@ -195,7 +196,7 @@ public class MultiTenantIT {
 
     @Test
     public void alterOffsetCommit(KafkaCluster cluster) throws Exception {
-        String config = getConfig(PROXY_ADDRESS, cluster);
+        String config = getConfig(cluster);
         try (var proxy = startProxy(config); var admin = Admin.create(commonConfig(TENANT1_PROXY_ADDRESS, Map.of()))) {
             var groupId = testInfo.getDisplayName();
             createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
@@ -213,7 +214,7 @@ public class MultiTenantIT {
 
     @Test
     public void deleteConsumerGroupOffsets(KafkaCluster cluster) throws Exception {
-        String config = getConfig(PROXY_ADDRESS, cluster);
+        String config = getConfig(cluster);
         try (var proxy = startProxy(config); var admin = Admin.create(commonConfig(TENANT1_PROXY_ADDRESS, Map.of()))) {
             var groupId = testInfo.getDisplayName();
             createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
@@ -233,7 +234,7 @@ public class MultiTenantIT {
 
     @Test
     public void tenantTopicIsolation(KafkaCluster cluster) throws Exception {
-        String config = getConfig(PROXY_ADDRESS, cluster);
+        String config = getConfig(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));
@@ -251,7 +252,7 @@ public class MultiTenantIT {
     @ParameterizedTest
     @EnumSource
     public void tenantConsumeWithGroup(ConsumerStyle consumerStyle, KafkaCluster cluster) throws Exception {
-        String config = getConfig(PROXY_ADDRESS, cluster);
+        String config = getConfig(cluster);
         try (var proxy = startProxy(config)) {
             createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
             runConsumerInOrderToCreateGroup(TENANT1_PROXY_ADDRESS, "Tenant1Group", NEW_TOPIC_1, consumerStyle);
@@ -262,7 +263,7 @@ public class MultiTenantIT {
     @ParameterizedTest
     @EnumSource
     public void tenantGroupIsolation(ConsumerStyle consumerStyle, KafkaCluster cluster) throws Exception {
-        String config = getConfig(PROXY_ADDRESS, cluster);
+        String config = getConfig(cluster);
         try (var proxy = startProxy(config)) {
             createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
             runConsumerInOrderToCreateGroup(TENANT1_PROXY_ADDRESS, "Tenant1Group", NEW_TOPIC_1, consumerStyle);
@@ -275,7 +276,7 @@ public class MultiTenantIT {
 
     @Test
     public void describeGroup(KafkaCluster cluster) throws Exception {
-        String config = getConfig(PROXY_ADDRESS, cluster);
+        String config = getConfig(cluster);
         try (var proxy = startProxy(config)) {
             createTopics(TENANT1_PROXY_ADDRESS, List.of(NEW_TOPIC_1));
             runConsumerInOrderToCreateGroup(TENANT1_PROXY_ADDRESS, "Tenant1Group", NEW_TOPIC_1, ConsumerStyle.ASSIGN);
@@ -455,16 +456,32 @@ public class MultiTenantIT {
         return config;
     }
 
-    private String getConfig(String proxyAddress, KafkaCluster cluster) {
+    private String getConfig(KafkaCluster cluster) {
         return KroxyConfig.builder()
-                .withNewProxy()
-                .withAddress(proxyAddress)
-                .withKeyPassword(certificateGenerator.getPassword())
-                .withKeyStoreFile(certificateGenerator.getKeyStoreLocation())
-                .endProxy()
-                .addToClusters("demo", new ClusterBuilder().withBootstrapServers(cluster.getBootstrapServers()).build())
+                .addToVirtualClusters("foo", new VirtualClusterBuilder()
+                        .withNewTargetCluster()
+                        .withBootstrapServers(cluster.getBootstrapServers())
+                        .endTargetCluster()
+                        .withNewClusterEndpointConfigProvider()
+                        .withType("StaticCluster")
+                        .withConfig(Map.of("bootstrapAddress", TENANT1_PROXY_ADDRESS))
+                        .endClusterEndpointConfigProvider()
+                        .withKeyPassword(certificateGenerator.getPassword())
+                        .withKeyStoreFile(certificateGenerator.getKeyStoreLocation())
+                        .build())
+                .addToVirtualClusters("bar", new VirtualClusterBuilder()
+                        .withNewTargetCluster()
+                        .withBootstrapServers(cluster.getBootstrapServers())
+                        .endTargetCluster()
+                        .withNewClusterEndpointConfigProvider()
+                        .withType("StaticCluster")
+                        .withConfig(Map.of("bootstrapAddress", TENANT2_PROXY_ADDRESS))
+                        .endClusterEndpointConfigProvider()
+                        .withKeyPassword(certificateGenerator.getPassword())
+                        .withKeyStoreFile(certificateGenerator.getKeyStoreLocation())
+                        .build())
                 .addNewFilter().withType("ApiVersions").endFilter()
-                .addNewFilter().withType("BrokerAddress").withConfig(Map.of("addressMapperClazz", "io.kroxylicious.proxy.internal.filter.SniAddressMapping")).endFilter()
+                .addNewFilter().withType("BrokerAddress").endFilter()
                 .addNewFilter().withType("MultiTenant").endFilter()
                 .build().toYaml();
     }
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusterendpointprovider/StaticClusterEndpointConfigProviderTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusterendpointprovider/StaticClusterEndpointConfigProviderTest.java
new file mode 100644
index 00000000..793f47cf
--- /dev/null
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusterendpointprovider/StaticClusterEndpointConfigProviderTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.clusterendpointprovider;
+
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import io.kroxylicious.proxy.internal.clusterendpointprovider.StaticClusterEndpointConfigProvider.StaticClusterEndpointProviderConfig;
+import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider.EndpointMatchResult;
+import io.kroxylicious.proxy.service.HostPort;
+
+import static io.kroxylicious.proxy.service.HostPort.parse;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class StaticClusterEndpointConfigProviderTest {
+
+    @Test
+    void noBrokerAddresses() {
+        assertThrows(IllegalArgumentException.class, () -> new StaticClusterEndpointProviderConfig(parse("good:1235"), Map.of()));
+    }
+
+    @Test
+    void rejectsDuplicatedBrokerAddresses() {
+        assertThrows(IllegalArgumentException.class,
+                () -> new StaticClusterEndpointProviderConfig(parse("good:1235"), Map.of(0, parse("duplicate:1234"), 1, parse("duplicate:1234"))));
+    }
+
+    @Test
+    void rejectsDuplicatedBrokerPorts() {
+        assertThrows(IllegalArgumentException.class,
+                () -> new StaticClusterEndpointProviderConfig(parse("foo:1235"), Map.of(0, parse("bar:1234"), 1, parse("baz:1234"))));
+    }
+
+    @Test
+    void validArguments() {
+        var provider = new StaticClusterEndpointConfigProvider(
+                new StaticClusterEndpointProviderConfig(parse("boot:1235"), Map.of(0, parse("good:1234"), 1, parse("good:2345"))));
+        assertThat(provider.getClusterBootstrapAddress()).isEqualTo(parse("boot:1235"));
+        assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("good:1234"));
+        assertThat(provider.getBrokerAddress(1)).isEqualTo(parse("good:2345"));
+        assertThat(provider.getNumberOfBrokerEndpointsToPrebind()).isEqualTo(2);
+    }
+
+    @Test
+    void defaultsBrokerZero() {
+        var provider = new StaticClusterEndpointConfigProvider(new StaticClusterEndpointProviderConfig(parse("boot:1235"), null));
+        HostPort parse1 = parse("boot:1235");
+        assertThat(provider.getClusterBootstrapAddress()).isEqualTo(parse1);
+        HostPort parse = parse("boot:1235");
+        assertThat(provider.getBrokerAddress(0)).isEqualTo(parse);
+        assertThat(provider.getNumberOfBrokerEndpointsToPrebind()).isEqualTo(1);
+    }
+
+    @ParameterizedTest
+    @NullAndEmptySource
+    @ValueSource(strings = { "idontcare" })
+    void findsBootstrapMatchByPort(String sniHostname) {
+        var provider = new StaticClusterEndpointConfigProvider(new StaticClusterEndpointProviderConfig(parse("boot:1234"), null));
+        assertThat(provider.hasMatchingEndpoint(sniHostname, 1234)).isEqualTo(new EndpointMatchResult(true, null));
+        assertThat(provider.hasMatchingEndpoint(sniHostname, 1235)).isEqualTo(new EndpointMatchResult(false, null));
+    }
+
+    @ParameterizedTest
+    @NullAndEmptySource
+    @ValueSource(strings = { "idontcare" })
+    void findsBrokerMatchByPort(String sniHostname) {
+        var provider = new StaticClusterEndpointConfigProvider(
+                new StaticClusterEndpointProviderConfig(parse("boot:1234"), Map.of(0, parse("broker:1235"), 1, parse("broker:1236"))));
+        assertThat(provider.hasMatchingEndpoint(sniHostname, 1235)).isEqualTo(new EndpointMatchResult(true, 0));
+        assertThat(provider.hasMatchingEndpoint(sniHostname, 1236)).isEqualTo(new EndpointMatchResult(true, 1));
+        assertThat(provider.hasMatchingEndpoint(sniHostname, 1237)).isEqualTo(new EndpointMatchResult(false, null));
+    }
+}