Repository: kroxylicious/kroxylicious
Pull request #498 (2023-07-30 23:39:04)

Clean name that cluster
### Type of change

- Enhancement / new feature

### Description

Includes the a name in the Virtual Cluster configuration.

### Additional Context

We need a programatic way of identifying which cluster is being used. This allows us to:
- share state at a cluster level.
- Improve logging so its easier to identify which clusters are being used

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [ ] Write tests
- [ ] Make sure all tests pass
- [ ] Review performance test results. Ensure that any degradations to performance numbers are understood and justified.
- [ ] Make sure all Sonar-Lint warnings are addressed or are justifiably ignored.
- [ ] Update documentation
- [ ] Reference relevant issue(s) and close them after merging
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=498)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=CODE_SMELL) [0 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=498) No Coverage information  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=498&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=498&metric=new_duplicated_lines_density&view=list)


robobario: Should we validate that the VCs are uniquely named somewhere?
sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=498)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=CODE_SMELL) [0 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=498&resolved=false&types=CODE_SMELL)

[![No Coverage information](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/NoCoverageInfo-16px.png 'No Coverage information')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=498) No Coverage information  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=498&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=498&metric=new_duplicated_lines_density&view=list)


SamBarker: > Should we validate that the VCs are uniquely named somewhere?

Yeah for operator sanity/error handling. Currently the last (as seen determined by jackson) cluster with the name will win. 


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 909a3f51..acbf54fd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,7 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 
 ## 0.3.0
 
+* [#498](https://github.com/kroxylicious/kroxylicious/issues/498): Include the cluster name from the configuration node in the config model.
 * [#488](https://github.com/kroxylicious/kroxylicious/pull/488): Kroxylicious Bill Of Materials 
 * [#480](https://github.com/kroxylicious/kroxylicious/issues/480): Multi-tenant - add suport for the versions of OffsetFetch, FindCoordinator, and DeleteTopics used by Sarama client v1.38.1
 * [#472](https://github.com/kroxylicious/kroxylicious/issues/472): Respect logFrame/logNetwork options in virtualcluster config
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
index 22b200cb..6715d442 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
@@ -11,7 +11,6 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -41,9 +40,7 @@ import io.kroxylicious.proxy.internal.KafkaProxyInitializer;
 import io.kroxylicious.proxy.internal.MeterRegistries;
 import io.kroxylicious.proxy.internal.PortConflictDetector;
 import io.kroxylicious.proxy.internal.admin.AdminHttpInitializer;
-import io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.ClusterNetworkAddressConfigProviderContributorManager;
 import io.kroxylicious.proxy.internal.net.DefaultNetworkBindingOperationProcessor;
-import io.kroxylicious.proxy.internal.net.Endpoint;
 import io.kroxylicious.proxy.internal.net.EndpointRegistry;
 import io.kroxylicious.proxy.internal.net.NetworkBindingOperationProcessor;
 import io.kroxylicious.proxy.internal.util.Metrics;
@@ -69,7 +66,7 @@ public final class KafkaProxy implements AutoCloseable {
     private final Configuration config;
     private final AdminHttpConfiguration adminHttpConfig;
     private final List<MicrometerDefinition> micrometerConfig;
-    private final Map<String, VirtualCluster> virtualClusterMap;
+    private final List<VirtualCluster> virtualClusters;
     private final AtomicBoolean running = new AtomicBoolean();
     private final CompletableFuture<Void> shutdown = new CompletableFuture<>();
     private EventGroupConfig adminEventGroup;
@@ -78,9 +75,7 @@ public final class KafkaProxy implements AutoCloseable {
 
     public KafkaProxy(Configuration config) {
         this.config = config;
-        this.virtualClusterMap = config.virtualClusters().entrySet().stream()
-                .collect(Collectors.toMap(Map.Entry::getKey,
-                        e -> toVirtualClusterModel(e.getValue())));
+        this.virtualClusters = config.virtualClusterModel();
         this.adminHttpConfig = config.adminHttpConfig();
         this.micrometerConfig = config.getMicrometer();
     }
@@ -98,7 +93,7 @@ public final class KafkaProxy implements AutoCloseable {
 
         var portConflictDefector = new PortConflictDetector();
         Optional<HostPort> adminHttpHostPort = Optional.ofNullable(shouldBindAdminEndpoint() ? new HostPort(adminHttpConfig.host(), adminHttpConfig.port()) : null);
-        portConflictDefector.validate(virtualClusterMap, adminHttpHostPort);
+        portConflictDefector.validate(virtualClusters, adminHttpHostPort);
 
         var availableCores = Runtime.getRuntime().availableProcessors();
         meterRegistries = new MeterRegistries(micrometerConfig);
@@ -114,12 +109,8 @@ public final class KafkaProxy implements AutoCloseable {
         bindingOperationProcessor.start(plainServerBootstrap, tlsServerBootstrap);
 
         // TODO: startup/shutdown should return a completionstage
-        List<CompletableFuture<Endpoint>> futures = new ArrayList<>();
-        virtualClusterMap.forEach((n, vc) -> {
-            futures.add(endpointRegistry.registerVirtualCluster(vc).toCompletableFuture());
-        });
-
-        CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join();
+        CompletableFuture.allOf(virtualClusters.stream().map(vc -> endpointRegistry.registerVirtualCluster(vc).toCompletableFuture()).toArray(CompletableFuture[]::new))
+                .join();
 
         // 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
@@ -253,13 +244,4 @@ public final class KafkaProxy implements AutoCloseable {
         }
     }
 
-    private static VirtualCluster toVirtualClusterModel(io.kroxylicious.proxy.config.VirtualCluster configModel) {
-        return new VirtualCluster(configModel.targetCluster(),
-                ClusterNetworkAddressConfigProviderContributorManager.getInstance()
-                        .getClusterEndpointConfigProvider(configModel.clusterNetworkAddressConfigProvider().type(),
-                                configModel.clusterNetworkAddressConfigProvider().config()),
-                configModel.tls(),
-                configModel.logNetwork(), configModel.logFrames());
-    }
-
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ConfigParser.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ConfigParser.java
index c5872474..6dc81b00 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ConfigParser.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ConfigParser.java
@@ -11,6 +11,7 @@ import java.io.InputStream;
 import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -52,7 +53,7 @@ public class ConfigParser {
             return MAPPER.writeValueAsString(configuration);
         }
         catch (JsonProcessingException e) {
-            throw new RuntimeException(e);
+            throw new IllegalArgumentException("Failed to encode configuration as YAML", e);
         }
     }
 
@@ -67,6 +68,7 @@ public class ConfigParser {
                 .setConstructorDetector(ConstructorDetector.USE_PROPERTIES_BASED)
                 .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
                 .configure(DeserializationFeature.FAIL_ON_MISSING_EXTERNAL_TYPE_ID_PROPERTY, false)
+                .configure(JsonParser.Feature.STRICT_DUPLICATE_DETECTION, true)
                 .setSerializationInclusion(JsonInclude.Include.NON_DEFAULT);
     }
 }
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 d665f864..36ef6e2a 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/Configuration.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/Configuration.java
@@ -26,4 +26,10 @@ public record Configuration(AdminHttpConfiguration adminHttp,
     public boolean isUseIoUring() {
         return useIoUring();
     }
+
+    public List<io.kroxylicious.proxy.model.VirtualCluster> virtualClusterModel() {
+        return virtualClusters.entrySet().stream()
+                .map(entry -> entry.getValue().toVirtualClusterModel(entry.getKey()))
+                .toList();
+    }
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/VirtualCluster.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/VirtualCluster.java
index aa61822c..d8b2f23d 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/VirtualCluster.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/VirtualCluster.java
@@ -10,12 +10,26 @@ import java.util.Optional;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 import io.kroxylicious.proxy.config.tls.Tls;
+import io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.ClusterNetworkAddressConfigProviderContributorManager;
+import io.kroxylicious.proxy.service.ClusterNetworkAddressConfigProvider;
 
 public record VirtualCluster(TargetCluster targetCluster,
                              @JsonProperty(required = true) ClusterNetworkAddressConfigProviderDefinition clusterNetworkAddressConfigProvider,
 
-                             @JsonProperty(required = false) Optional<Tls> tls,
+                             @JsonProperty() Optional<Tls> tls,
                              boolean logNetwork,
                              boolean logFrames
 ) {
+    public io.kroxylicious.proxy.model.VirtualCluster toVirtualClusterModel(String virtualClusterNodeName) {
+        return new io.kroxylicious.proxy.model.VirtualCluster(virtualClusterNodeName,
+                targetCluster(),
+                toClusterNetworkAddressConfigProviderModel(),
+                tls(),
+                logNetwork(), logFrames());
+    }
+
+    private ClusterNetworkAddressConfigProvider toClusterNetworkAddressConfigProviderModel() {
+        return ClusterNetworkAddressConfigProviderContributorManager.getInstance()
+                .getClusterEndpointConfigProvider(clusterNetworkAddressConfigProvider().type(), clusterNetworkAddressConfigProvider().config());
+    }
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/PortConflictDetector.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/PortConflictDetector.java
index b6748218..18c7e71b 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/PortConflictDetector.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/PortConflictDetector.java
@@ -6,6 +6,8 @@
 
 package io.kroxylicious.proxy.internal;
 
+import java.util.Collection;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -21,9 +23,10 @@ import io.kroxylicious.proxy.service.HostPort;
 /**
  * Detects potential for port conflicts arising between virtual cluster configurations.
  */
+@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
 public class PortConflictDetector {
 
-    private final static Optional<String> ANY_INTERFACE = Optional.empty();
+    private static final Optional<String> ANY_INTERFACE = Optional.empty();
     private static final String ANY_STRING = "<any>";
 
     private enum BindingScope {
@@ -37,112 +40,141 @@ public class PortConflictDetector {
      * @param virtualClusterMap map of virtual clusters.
      * @param otherExclusivePort an optional exclusive port that should conflict with virtual cluster ports
      */
-    public void validate(Map<String, VirtualCluster> virtualClusterMap, Optional<HostPort> otherExclusivePort) {
+    public void validate(Collection<VirtualCluster> virtualClusterMap, Optional<HostPort> otherExclusivePort) {
 
         Set<String> seenVirtualClusters = new HashSet<>();
 
         Map<Optional<String>, Set<Integer>> inUseExclusivePorts = new HashMap<>();
         Map<Optional<String>, Map<Integer, Boolean>> inUseSharedPorts = new HashMap<>();
 
-        virtualClusterMap.entrySet().stream().sorted(Map.Entry.comparingByKey()).forEach(e -> {
-            var name = e.getKey();
-            var cluster = e.getValue();
-            var proposedSharedPorts = cluster.getSharedPorts();
-            var proposedExclusivePorts = cluster.getExclusivePorts();
-
-            if (otherExclusivePort.isPresent()) {
-                Optional<String> otherInterface = otherExclusivePort.map(hostPort -> hostPort.host().equals("0.0.0.0") ? null : hostPort.host());
-                var checkPorts = cluster.getBindAddress().isEmpty() || otherInterface.isEmpty() || cluster.getBindAddress().equals(otherInterface);
-                if (checkPorts) {
-                    checkForConflictsWithOtherExclusivePort(otherExclusivePort.get(), name, cluster, proposedExclusivePorts, BindingScope.EXCLUSIVE);
-                    checkForConflictsWithOtherExclusivePort(otherExclusivePort.get(), name, cluster, proposedSharedPorts, BindingScope.SHARED);
-                }
-            }
-
-            // if this virtual cluster is binding to <any>, we need to check for conflicts on the <any> interface and *all* specific interfaces,
-            // otherwise we just check for conflicts on <any> and the specified specific interface
-            var exclusiveCheckSet = cluster.getBindAddress().isEmpty() ? inUseExclusivePorts.keySet() : Set.of(ANY_INTERFACE, cluster.getBindAddress());
-
-            // check the proposed *exclusive ports* for conflicts with the *exclusive ports* seen so far.
-
-            inUseExclusivePorts.entrySet().stream()
-                    .filter(interfacePortsEntry -> exclusiveCheckSet.contains(interfacePortsEntry.getKey()))
-                    .forEach((entry) -> {
-                        var bindingInterface = entry.getKey();
-                        var ports = entry.getValue();
-                        var conflicts = getSortedPortConflicts(ports, proposedExclusivePorts);
-                        if (!conflicts.isEmpty()) {
-                            throw buildPortConflictException(name, seenVirtualClusters, conflicts,
-                                    cluster.getBindAddress(), BindingScope.EXCLUSIVE,
-                                    bindingInterface, BindingScope.EXCLUSIVE);
-                        }
-                    });
-
-            // check the proposed *shared ports* for conflicts with the *exclusive ports* seen so far.
-
-            inUseExclusivePorts.entrySet().stream()
-                    .filter(interfacePortsEntry -> exclusiveCheckSet.contains(interfacePortsEntry.getKey()))
-                    .forEach((entry) -> {
-                        var bindingInterface = entry.getKey();
-                        var ports = entry.getValue();
-                        var conflicts = getSortedPortConflicts(ports, proposedSharedPorts);
-                        if (!conflicts.isEmpty()) {
-                            throw buildPortConflictException(name, seenVirtualClusters, conflicts,
-                                    cluster.getBindAddress(), BindingScope.SHARED,
-                                    bindingInterface, BindingScope.EXCLUSIVE);
-                        }
-                    });
-
-            // check the proposed *exclusive ports* for conflicts with the *shared ports* seen so far.
-
-            inUseSharedPorts.entrySet().stream()
-                    .filter(interfacePortsEntry -> exclusiveCheckSet.contains(interfacePortsEntry.getKey()))
-                    .forEach((entry) -> {
-                        var bindingInterface = entry.getKey();
-                        var ports = entry.getValue().keySet();
-                        var conflicts = getSortedPortConflicts(ports, proposedExclusivePorts);
-                        if (!conflicts.isEmpty()) {
-                            throw buildPortConflictException(name, seenVirtualClusters, conflicts,
-                                    cluster.getBindAddress(), BindingScope.EXCLUSIVE,
-                                    bindingInterface, BindingScope.SHARED);
-                        }
-                    });
-
-            // check the proposed *shared ports* for conflicts with the shared ports seen so far on *different* interfaces.
-
-            var sharedCheckSet = new HashSet<>(exclusiveCheckSet);
-            sharedCheckSet.remove(cluster.getBindAddress());
-
-            inUseSharedPorts.entrySet().stream()
-                    .filter(interfacePortsEntry -> sharedCheckSet.contains(interfacePortsEntry.getKey()))
-                    .forEach((entry) -> {
-                        var bindingInterface = entry.getKey();
-                        var ports = entry.getValue().keySet();
-                        var conflicts = getSortedPortConflicts(ports, proposedSharedPorts);
-                        if (!conflicts.isEmpty()) {
-                            throw buildPortConflictException(name, seenVirtualClusters, conflicts,
-                                    cluster.getBindAddress(), BindingScope.SHARED,
-                                    bindingInterface, BindingScope.SHARED);
+        virtualClusterMap.stream()
+                .sorted(Comparator.comparing(VirtualCluster::getClusterName))
+                .forEach(virtualCluster -> {
+                    var name = virtualCluster.getClusterName();
+                    var proposedSharedPorts = virtualCluster.getSharedPorts();
+                    var proposedExclusivePorts = virtualCluster.getExclusivePorts();
+
+                    if (otherExclusivePort.isPresent()) {
+                        Optional<String> otherInterface = otherExclusivePort.map(hostPort -> hostPort.host().equals("0.0.0.0") ? null : hostPort.host());
+                        var checkPorts = virtualCluster.getBindAddress().isEmpty() || otherInterface.isEmpty() || virtualCluster.getBindAddress().equals(otherInterface);
+                        if (checkPorts) {
+                            checkForConflictsWithOtherExclusivePort(otherExclusivePort.get(), name, virtualCluster, proposedExclusivePorts, BindingScope.EXCLUSIVE);
+                            checkForConflictsWithOtherExclusivePort(otherExclusivePort.get(), name, virtualCluster, proposedSharedPorts, BindingScope.SHARED);
                         }
-                    });
+                    }
+
+                    // if this virtual cluster is binding to <any>, we need to check for conflicts on the <any> interface and *all* specific interfaces,
+                    // otherwise we just check for conflicts on <any> and the specified specific interface
+                    var exclusiveCheckSet = virtualCluster.getBindAddress().isEmpty() ? inUseExclusivePorts.keySet()
+                            : Set.of(ANY_INTERFACE, virtualCluster.getBindAddress());
+
+                    // check the proposed *exclusive ports* for conflicts with the *exclusive ports* seen so far.
+                    assertExclusivePortsAreMutuallyExclusive(seenVirtualClusters, inUseExclusivePorts, virtualCluster, name, proposedExclusivePorts, exclusiveCheckSet);
+
+                    // check the proposed *shared ports* for conflicts with the *exclusive ports* seen so far.
+                    assertSharedPortsDoNotOverlapWithExlusivePorts(seenVirtualClusters, inUseExclusivePorts, virtualCluster, name, proposedSharedPorts,
+                            exclusiveCheckSet);
+
+                    // check the proposed *exclusive ports* for conflicts with the *shared ports* seen so far.
+                    assertExclusivePortsDoNotOverlapWithExclusivePorts(seenVirtualClusters, inUseSharedPorts, virtualCluster, name, proposedExclusivePorts,
+                            exclusiveCheckSet);
+
+                    // check the proposed *shared ports* for conflicts with the shared ports seen so far on *different* interfaces.
+                    assertSharedPortsAreExclusiveAcrossInterfaces(seenVirtualClusters, inUseSharedPorts, virtualCluster, name, proposedSharedPorts, exclusiveCheckSet);
 
-            // check for proposed shared ports for differing TLS configuration
+                    // check for proposed shared ports for differing TLS configuration
+                    assertSharedPortsHaveMatchingTlsConfiguration(seenVirtualClusters, inUseSharedPorts, virtualCluster, name, proposedSharedPorts);
 
-            proposedSharedPorts.forEach(p -> {
-                var tls = inUseSharedPorts.getOrDefault(cluster.getBindAddress(), Map.of()).get(p);
-                if (tls != null && !tls.equals(cluster.isUseTls())) {
-                    throw buildOverviewException(name, seenVirtualClusters, buildTlsConflictException(p, cluster.getBindAddress()));
-                }
-            });
+                    seenVirtualClusters.add(name);
+                    inUseExclusivePorts.computeIfAbsent(virtualCluster.getBindAddress(), k -> new HashSet<>()).addAll(proposedExclusivePorts);
+                    var sharedMap = inUseSharedPorts.computeIfAbsent(virtualCluster.getBindAddress(), k -> new HashMap<>());
+                    proposedSharedPorts.forEach(p -> sharedMap.put(p, virtualCluster.isUseTls()));
 
-            seenVirtualClusters.add(name);
-            inUseExclusivePorts.computeIfAbsent(cluster.getBindAddress(), (k) -> new HashSet<>()).addAll(proposedExclusivePorts);
-            var sharedMap = inUseSharedPorts.computeIfAbsent(cluster.getBindAddress(), (k) -> new HashMap<>());
-            proposedSharedPorts.forEach(p -> sharedMap.put(p, cluster.isUseTls()));
+                });
+    }
 
+    private void assertSharedPortsHaveMatchingTlsConfiguration(Set<String> seenVirtualClusters, Map<Optional<String>, Map<Integer, Boolean>> inUseSharedPorts,
+                                                               VirtualCluster virtualCluster, String name,
+                                                               Set<Integer> proposedSharedPorts) {
+        proposedSharedPorts.forEach(p -> {
+            var tls = inUseSharedPorts.getOrDefault(virtualCluster.getBindAddress(), Map.of()).get(p);
+            if (tls != null && !tls.equals(virtualCluster.isUseTls())) {
+                throw buildOverviewException(name, seenVirtualClusters, buildTlsConflictException(p, virtualCluster.getBindAddress()));
+            }
         });
     }
 
+    private void assertSharedPortsAreExclusiveAcrossInterfaces(Set<String> seenVirtualClusters, Map<Optional<String>, Map<Integer, Boolean>> inUseSharedPorts,
+                                                               VirtualCluster virtualCluster, String name,
+                                                               Set<Integer> proposedSharedPorts, Set<Optional<String>> exclusiveCheckSet) {
+        var sharedCheckSet = new HashSet<>(exclusiveCheckSet);
+        sharedCheckSet.remove(virtualCluster.getBindAddress());
+
+        inUseSharedPorts.entrySet().stream()
+                .filter(interfacePortsEntry -> sharedCheckSet.contains(interfacePortsEntry.getKey()))
+                .forEach(entry -> {
+                    var bindingInterface = entry.getKey();
+                    var ports = entry.getValue().keySet();
+                    var conflicts = getSortedPortConflicts(ports, proposedSharedPorts);
+                    if (!conflicts.isEmpty()) {
+                        throw buildPortConflictException(name, seenVirtualClusters, conflicts,
+                                virtualCluster.getBindAddress(), BindingScope.SHARED,
+                                bindingInterface, BindingScope.SHARED);
+                    }
+                });
+    }
+
+    private void assertExclusivePortsDoNotOverlapWithExclusivePorts(Set<String> seenVirtualClusters, Map<Optional<String>, Map<Integer, Boolean>> inUseSharedPorts,
+                                                                    VirtualCluster virtualCluster, String name,
+                                                                    Set<Integer> proposedExclusivePorts, Set<Optional<String>> exclusiveCheckSet) {
+        inUseSharedPorts.entrySet().stream()
+                .filter(interfacePortsEntry -> exclusiveCheckSet.contains(interfacePortsEntry.getKey()))
+                .forEach(entry -> {
+                    var bindingInterface = entry.getKey();
+                    var ports = entry.getValue().keySet();
+                    var conflicts = getSortedPortConflicts(ports, proposedExclusivePorts);
+                    if (!conflicts.isEmpty()) {
+                        throw buildPortConflictException(name, seenVirtualClusters, conflicts,
+                                virtualCluster.getBindAddress(), BindingScope.EXCLUSIVE,
+                                bindingInterface, BindingScope.SHARED);
+                    }
+                });
+    }
+
+    private void assertSharedPortsDoNotOverlapWithExlusivePorts(Set<String> seenVirtualClusters, Map<Optional<String>, Set<Integer>> inUseExclusivePorts,
+                                                                VirtualCluster virtualCluster, String name,
+                                                                Set<Integer> proposedSharedPorts, Set<Optional<String>> exclusiveCheckSet) {
+        inUseExclusivePorts.entrySet().stream()
+                .filter(interfacePortsEntry -> exclusiveCheckSet.contains(interfacePortsEntry.getKey()))
+                .forEach(entry -> {
+                    var bindingInterface = entry.getKey();
+                    var ports = entry.getValue();
+                    var conflicts = getSortedPortConflicts(ports, proposedSharedPorts);
+                    if (!conflicts.isEmpty()) {
+                        throw buildPortConflictException(name, seenVirtualClusters, conflicts,
+                                virtualCluster.getBindAddress(), BindingScope.SHARED,
+                                bindingInterface, BindingScope.EXCLUSIVE);
+                    }
+                });
+    }
+
+    private void assertExclusivePortsAreMutuallyExclusive(Set<String> seenVirtualClusters, Map<Optional<String>, Set<Integer>> inUseExclusivePorts,
+                                                          VirtualCluster virtualCluster, String name,
+                                                          Set<Integer> proposedExclusivePorts, Set<Optional<String>> exclusiveCheckSet) {
+        inUseExclusivePorts.entrySet().stream()
+                .filter(interfacePortsEntry -> exclusiveCheckSet.contains(interfacePortsEntry.getKey()))
+                .forEach(entry -> {
+                    var bindingInterface = entry.getKey();
+                    var ports = entry.getValue();
+                    var conflicts = getSortedPortConflicts(ports, proposedExclusivePorts);
+                    if (!conflicts.isEmpty()) {
+                        throw buildPortConflictException(name, seenVirtualClusters, conflicts,
+                                virtualCluster.getBindAddress(), BindingScope.EXCLUSIVE,
+                                bindingInterface, BindingScope.EXCLUSIVE);
+                    }
+                });
+    }
+
     private void checkForConflictsWithOtherExclusivePort(HostPort otherHostPort, String name, VirtualCluster cluster, Set<Integer> proposedExclusivePorts,
                                                          BindingScope scope) {
         var ports = Set.of(otherHostPort.port());
@@ -151,7 +183,7 @@ public class PortConflictDetector {
             Optional<String> proposedBindingInterface = cluster.getBindAddress();
             var portConflicts = conflicts.stream().map(String::valueOf).collect(Collectors.joining(","));
 
-            throw new IllegalStateException(("The %s bind of port(s) %s for virtual cluster '%s' to %s would conflict with another (non-cluster) port binding").formatted(
+            throw new IllegalStateException("The %s bind of port(s) %s for virtual cluster '%s' to %s would conflict with another (non-cluster) port binding".formatted(
                     scope.name().toLowerCase(Locale.ROOT),
                     portConflicts,
                     name,
@@ -160,7 +192,7 @@ public class PortConflictDetector {
     }
 
     private List<Integer> getSortedPortConflicts(Set<Integer> ports, Set<Integer> candidates) {
-        return ports.stream().filter(candidates::contains).sorted().collect(Collectors.toList());
+        return ports.stream().filter(candidates::contains).sorted().toList();
     }
 
     private IllegalStateException buildPortConflictException(String virtualClusterName, Set<String> seenVirtualClusters, List<Integer> conflicts,
@@ -168,7 +200,7 @@ public class PortConflictDetector {
                                                              Optional<String> existingBindingInterface, BindingScope existingBindingScope) {
         var portConflicts = conflicts.stream().map(String::valueOf).collect(Collectors.joining(","));
 
-        var underlying = new IllegalStateException(("The %s bind of port(s) %s to %s would conflict with existing %s port bindings on %s.").formatted(
+        var underlying = new IllegalStateException("The %s bind of port(s) %s to %s would conflict with existing %s port bindings on %s.".formatted(
                 proposedScope.name().toLowerCase(Locale.ROOT),
                 portConflicts,
                 proposedBindingInterface.orElse(ANY_STRING),
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/model/VirtualCluster.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/model/VirtualCluster.java
index fdfecdfe..15db2add 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/model/VirtualCluster.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/model/VirtualCluster.java
@@ -22,7 +22,9 @@ import io.kroxylicious.proxy.config.tls.Tls;
 import io.kroxylicious.proxy.service.ClusterNetworkAddressConfigProvider;
 import io.kroxylicious.proxy.service.HostPort;
 
+@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
 public class VirtualCluster implements ClusterNetworkAddressConfigProvider {
+    private final String clusterName;
 
     private final TargetCluster targetCluster;
 
@@ -32,33 +34,36 @@ public class VirtualCluster implements ClusterNetworkAddressConfigProvider {
     private final boolean logFrames;
 
     private final ClusterNetworkAddressConfigProvider clusterNetworkAddressConfigProvider;
+
     private final Optional<SslContext> upstreamSslContext;
+
     private final Optional<SslContext> downstreamSslContext;
 
-    public VirtualCluster(TargetCluster targetCluster,
+    public VirtualCluster(String clusterName,
+                          TargetCluster targetCluster,
                           ClusterNetworkAddressConfigProvider clusterNetworkAddressConfigProvider,
                           Optional<Tls> tls,
-                          boolean logNetwork, boolean logFrames) {
+                          boolean logNetwork,
+                          boolean logFrames) {
+        this.clusterName = clusterName;
         this.tls = tls;
-        if (clusterNetworkAddressConfigProvider.requiresTls() && (tls.isEmpty() || !tls.get().definesKey())) {
-            throw new IllegalStateException("Cluster endpoint provider requires server TLS, but this virtual cluster does not define it.");
-        }
-        var conflicts = clusterNetworkAddressConfigProvider.getExclusivePorts().stream().filter(p -> clusterNetworkAddressConfigProvider.getSharedPorts().contains(p))
-                .collect(Collectors.toSet());
-        if (!conflicts.isEmpty()) {
-            throw new IllegalStateException(
-                    "The set of exclusive ports described by the cluster endpoint provider must be distinct from those described as shared. Intersection: " + conflicts);
-        }
         this.targetCluster = targetCluster;
         this.logNetwork = logNetwork;
         this.logFrames = logFrames;
         this.clusterNetworkAddressConfigProvider = clusterNetworkAddressConfigProvider;
 
+        validateTLsSettings(clusterNetworkAddressConfigProvider, tls);
+        validatePortUsage(clusterNetworkAddressConfigProvider);
+
         // TODO: https://github.com/kroxylicious/kroxylicious/issues/104 be prepared to reload the SslContext at runtime.
         this.upstreamSslContext = buildUpstreamSslContext();
         this.downstreamSslContext = buildDownstreamSslContext();
     }
 
+    public String getClusterName() {
+        return clusterName;
+    }
+
     public TargetCluster targetCluster() {
         return targetCluster;
     }
@@ -79,40 +84,18 @@ public class VirtualCluster implements ClusterNetworkAddressConfigProvider {
         return tls.isPresent();
     }
 
-    private Optional<SslContext> buildDownstreamSslContext() {
-        return tls.map(tls -> {
-            try {
-                return Optional.of(tls.key()).map(KeyProvider::forServer).orElseThrow().build();
-            }
-            catch (SSLException e) {
-                throw new UncheckedIOException(e);
-            }
-        });
-    }
-
-    private Optional<SslContext> buildUpstreamSslContext() {
-        return targetCluster.tls().map(tls -> {
-            try {
-                var sslContextBuilder = SslContextBuilder.forClient();
-                Optional.ofNullable(tls.trust()).ifPresent(tp -> tp.apply(sslContextBuilder));
-                return sslContextBuilder.build();
-            }
-            catch (SSLException e) {
-                throw new UncheckedIOException(e);
-            }
-        });
-    }
-
     @Override
     public String toString() {
-        final StringBuilder sb = new StringBuilder("VirtualCluster [");
-        sb.append("targetCluster=").append(targetCluster);
-        sb.append(", clusterNetworkAddressConfigProvider=").append(clusterNetworkAddressConfigProvider);
-        sb.append(", tls=").append(tls.map(Tls::toString).orElse(null));
-        sb.append(", logNetwork=").append(logNetwork);
-        sb.append(", logFrames=").append(logFrames);
-        sb.append(']');
-        return sb.toString();
+        return "VirtualCluster{" +
+                "clusterName='" + clusterName + '\'' +
+                ", targetCluster=" + targetCluster +
+                ", tls=" + tls +
+                ", logNetwork=" + logNetwork +
+                ", logFrames=" + logFrames +
+                ", clusterNetworkAddressConfigProvider=" + clusterNetworkAddressConfigProvider +
+                ", upstreamSslContext=" + upstreamSslContext +
+                ", downstreamSslContext=" + downstreamSslContext +
+                '}';
     }
 
     @Override
@@ -163,4 +146,42 @@ public class VirtualCluster implements ClusterNetworkAddressConfigProvider {
         return upstreamSslContext;
     }
 
+    private Optional<SslContext> buildDownstreamSslContext() {
+        return tls.map(tls -> {
+            try {
+                return Optional.of(tls.key()).map(KeyProvider::forServer).orElseThrow().build();
+            }
+            catch (SSLException e) {
+                throw new UncheckedIOException(e);
+            }
+        });
+    }
+
+    private Optional<SslContext> buildUpstreamSslContext() {
+        return targetCluster.tls().map(tls -> {
+            try {
+                var sslContextBuilder = SslContextBuilder.forClient();
+                Optional.ofNullable(tls.trust()).ifPresent(tp -> tp.apply(sslContextBuilder));
+                return sslContextBuilder.build();
+            }
+            catch (SSLException e) {
+                throw new UncheckedIOException(e);
+            }
+        });
+    }
+
+    private static void validatePortUsage(ClusterNetworkAddressConfigProvider clusterNetworkAddressConfigProvider) {
+        var conflicts = clusterNetworkAddressConfigProvider.getExclusivePorts().stream().filter(p -> clusterNetworkAddressConfigProvider.getSharedPorts().contains(p))
+                .collect(Collectors.toSet());
+        if (!conflicts.isEmpty()) {
+            throw new IllegalStateException(
+                    "The set of exclusive ports described by the cluster endpoint provider must be distinct from those described as shared. Intersection: " + conflicts);
+        }
+    }
+
+    private static void validateTLsSettings(ClusterNetworkAddressConfigProvider clusterNetworkAddressConfigProvider, Optional<Tls> tls) {
+        if (clusterNetworkAddressConfigProvider.requiresTls() && (tls.isEmpty() || !tls.get().definesKey())) {
+            throw new IllegalStateException("Cluster endpoint provider requires server TLS, but this virtual cluster does not define it.");
+        }
+    }
 }
diff --git a/pom.xml b/pom.xml
index 07afcf42..7fc7b935 100644
--- a/pom.xml
+++ b/pom.xml
@@ -38,7 +38,7 @@
         <kroxy.extension.version>0.4.0</kroxy.extension.version>
         <log4j.version>2.20.0</log4j.version>
         <picocli.version>4.7.4</picocli.version>
-        <netty.version>4.1.96.Final</netty.version>
+        <netty.version>4.1.95.Final</netty.version>
         <netty.io_uring.version>0.0.21.Final</netty.io_uring.version>
         <netty.epoll.classifier>linux-x86_64</netty.epoll.classifier>
         <netty.io_uring.classifier>linux-x86_64</netty.io_uring.classifier>


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/ExpositionIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/ExpositionIT.java
index a54a1fdc..0b5d5fa8 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/ExpositionIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/ExpositionIT.java
@@ -84,10 +84,10 @@ public class ExpositionIT {
 
     @ParameterizedTest(name = "{0}")
     @MethodSource("virtualClusterConfigurations")
-    public void exposesSingleUpstreamClusterOverTls(String name,
-                                                    VirtualClusterBuilder virtualClusterBuilder,
-                                                    Map<String, Object> clientSecurityProtocolConfig,
-                                                    @BrokerCluster(numBrokers = 2) KafkaCluster cluster) {
+    void exposesSingleUpstreamClusterOverTls(String name,
+                                             VirtualClusterBuilder virtualClusterBuilder,
+                                             Map<String, Object> clientSecurityProtocolConfig,
+                                             @BrokerCluster(numBrokers = 2) KafkaCluster cluster) {
         virtualClusterBuilder.editOrNewTargetCluster().withBootstrapServers(cluster.getBootstrapServers()).endTargetCluster();
         var builder = new ConfigurationBuilder()
                 .addToVirtualClusters("demo", virtualClusterBuilder.build());
@@ -106,7 +106,7 @@ public class ExpositionIT {
     }
 
     @Test
-    public void exposesTwoClusterOverPlainWithSeparatePorts(KafkaCluster cluster) {
+    void exposesTwoClusterOverPlainWithSeparatePorts(KafkaCluster cluster) {
         List<String> clusterProxyAddresses = List.of("localhost:9192", "localhost:9294");
 
         var builder = new ConfigurationBuilder();
@@ -139,7 +139,7 @@ public class ExpositionIT {
     }
 
     @Test
-    public void exposesTwoSeparateUpstreamClustersUsingSniRouting(KafkaCluster cluster) throws Exception {
+    void exposesTwoSeparateUpstreamClustersUsingSniRouting(KafkaCluster cluster) throws Exception {
         var keystoreTrustStoreList = new ArrayList<KeystoreTrustStorePair>();
         var virtualClusterCommonNamePattern = IntegrationTestInetAddressResolverProvider.generateFullyQualifiedDomainName(".virtualcluster%d");
         var virtualClusterBootstrapPattern = "bootstrap" + virtualClusterCommonNamePattern;
@@ -191,7 +191,7 @@ public class ExpositionIT {
     }
 
     @Test
-    public void exposesClusterOfTwoBrokers(@BrokerCluster(numBrokers = 2) KafkaCluster cluster) throws Exception {
+    void exposesClusterOfTwoBrokers(@BrokerCluster(numBrokers = 2) KafkaCluster cluster) throws Exception {
         var builder = new ConfigurationBuilder()
                 .addToVirtualClusters("demo", new VirtualClusterBuilder()
                         .withNewTargetCluster()
@@ -268,10 +268,10 @@ public class ExpositionIT {
      */
     @ParameterizedTest(name = "{0}")
     @MethodSource(value = "virtualClusterConfigurations")
-    public void connectToExposedBrokerEndpointsDirectlyAfterKroxyliciousRestart(String name,
-                                                                                VirtualClusterBuilder virtualClusterBuilder,
-                                                                                Map<String, Object> clientSecurityProtocolConfig,
-                                                                                @BrokerCluster(numBrokers = 2) KafkaCluster cluster) {
+    void connectToExposedBrokerEndpointsDirectlyAfterKroxyliciousRestart(String name,
+                                                                         VirtualClusterBuilder virtualClusterBuilder,
+                                                                         Map<String, Object> clientSecurityProtocolConfig,
+                                                                         @BrokerCluster(numBrokers = 2) KafkaCluster cluster) {
         connectToExposedBrokerEndpointsDirectlyAfterKroxyliciousRestart(virtualClusterBuilder, clientSecurityProtocolConfig, cluster);
     }
 
@@ -284,10 +284,10 @@ public class ExpositionIT {
      */
     @ParameterizedTest(name = "{0}")
     @MethodSource(value = "virtualClusterConfigurations")
-    public void connectToExposedBrokerEndpointsDirectlyAfterKroxyliciousRestart_Sasl(String name,
-                                                                                     VirtualClusterBuilder virtualClusterBuilder,
-                                                                                     Map<String, Object> clientSecurityProtocolConfig,
-                                                                                     @BrokerCluster(numBrokers = 2) @SaslPlainAuth(user = SASL_USER, password = SASL_PASSWORD) KafkaCluster cluster) {
+    void connectToExposedBrokerEndpointsDirectlyAfterKroxyliciousRestart_Sasl(String name,
+                                                                              VirtualClusterBuilder virtualClusterBuilder,
+                                                                              Map<String, Object> clientSecurityProtocolConfig,
+                                                                              @BrokerCluster(numBrokers = 2) @SaslPlainAuth(user = SASL_USER, password = SASL_PASSWORD) KafkaCluster cluster) {
 
         var securityProtocol = virtualClusterBuilder.hasTls() ? SecurityProtocol.SASL_SSL : SecurityProtocol.SASL_PLAINTEXT;
         clientSecurityProtocolConfig = new HashMap<>(clientSecurityProtocolConfig);
@@ -350,10 +350,10 @@ public class ExpositionIT {
 
     @ParameterizedTest(name = "{0}")
     @MethodSource(value = "virtualClusterConfigurations")
-    public void connectToDiscoveryAddress(String name,
-                                          VirtualClusterBuilder virtualClusterBuilder,
-                                          Map<String, Object> clientSecurityProtocolConfig,
-                                          @BrokerCluster KafkaCluster cluster) {
+    void connectToDiscoveryAddress(String name,
+                                   VirtualClusterBuilder virtualClusterBuilder,
+                                   Map<String, Object> clientSecurityProtocolConfig,
+                                   @BrokerCluster KafkaCluster cluster) {
         virtualClusterBuilder.editOrNewTargetCluster().withBootstrapServers(cluster.getBootstrapServers()).endTargetCluster();
         var builder = new ConfigurationBuilder()
                 .addToVirtualClusters("demo", virtualClusterBuilder.build());
@@ -405,7 +405,7 @@ public class ExpositionIT {
     }
 
     @Test
-    public void targetClusterDynamicallyAddsBroker(@BrokerCluster KafkaCluster cluster) throws Exception {
+    void targetClusterDynamicallyAddsBroker(@BrokerCluster KafkaCluster cluster) throws Exception {
         var builder = new ConfigurationBuilder()
                 .addToVirtualClusters("demo", new VirtualClusterBuilder()
                         .withNewTargetCluster()
@@ -437,7 +437,7 @@ public class ExpositionIT {
     }
 
     @Test
-    public void targetClusterDynamicallyRemovesBroker(@BrokerCluster(numBrokers = 2) KafkaCluster cluster) throws Exception {
+    void targetClusterDynamicallyRemovesBroker(@BrokerCluster(numBrokers = 2) KafkaCluster cluster) throws Exception {
         var builder = new ConfigurationBuilder()
                 .addToVirtualClusters("demo", new VirtualClusterBuilder()
                         .withNewTargetCluster()
@@ -463,6 +463,7 @@ public class ExpositionIT {
                         n -> n.size() == cluster.getNumOfBrokers());
 
                 assertThat(updatedNodes).describedAs("removed node must not appear in the describeCluster response")
+                        .isNotEmpty()
                         .allSatisfy(n -> assertThat(n.id()).isNotEqualTo(removedNodeId));
             }
 
diff --git a/kroxylicious-test-tools/src/test/java/io/kroxylicious/proxy/config/ConfigurationTest.java b/kroxylicious-test-tools/src/test/java/io/kroxylicious/proxy/config/ConfigurationTest.java
index 5cf87065..421828d2 100644
--- a/kroxylicious-test-tools/src/test/java/io/kroxylicious/proxy/config/ConfigurationTest.java
+++ b/kroxylicious-test-tools/src/test/java/io/kroxylicious/proxy/config/ConfigurationTest.java
@@ -6,12 +6,15 @@
 
 package io.kroxylicious.proxy.config;
 
+import java.util.List;
 import java.util.stream.Stream;
 
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import com.fasterxml.jackson.databind.JsonMappingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
@@ -20,6 +23,7 @@ import com.flipkart.zjsonpatch.JsonDiff;
 import io.kroxylicious.proxy.internal.filter.ProduceRequestTransformationFilter;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 class ConfigurationTest {
 
@@ -101,7 +105,7 @@ class ConfigurationTest {
 
     @ParameterizedTest(name = "{0}")
     @MethodSource
-    public void yamlDeserializeSerializeFidelity(String name, String config) throws Exception {
+    void yamlDeserializeSerializeFidelity(String name, String config) throws Exception {
 
         var configuration = configParser.parseConfiguration(config);
         var roundTripped = configParser.toYaml(configuration);
@@ -286,7 +290,7 @@ class ConfigurationTest {
 
     @ParameterizedTest(name = "{0}")
     @MethodSource
-    public void fluentApiConfigYamlFidelity(String name, Configuration config, String expected) throws Exception {
+    void fluentApiConfigYamlFidelity(String name, Configuration config, String expected) throws Exception {
         var yaml = configParser.toYaml(config);
         var actualJson = MAPPER.reader().readValue(yaml, JsonNode.class);
         var expectedJson = MAPPER.reader().readValue(expected, JsonNode.class);
@@ -294,4 +298,63 @@ class ConfigurationTest {
         assertThat(diff).isEmpty();
 
     }
+
+    @Test
+    void shouldConfigureClusterNameFromNodeName() {
+        // Given
+        final Configuration configurationModel = configParser.parseConfiguration("""
+                virtualClusters:
+                  myAwesomeCluster:
+                    targetCluster:
+                      bootstrap_servers: kafka.example:1234
+                    clusterNetworkAddressConfigProvider:
+                      type: PortPerBroker
+                      config:
+                        bootstrapAddress: cluster1:9192
+                        numberOfBrokerPorts: 1
+                        brokerAddressPattern: localhost
+                        brokerStartPort: 9193
+                """);
+        // When
+        final List<io.kroxylicious.proxy.model.VirtualCluster> actualValidClusters = configurationModel.virtualClusterModel();
+
+        // Then
+        assertThat(actualValidClusters).singleElement().extracting("clusterName").isEqualTo("myAwesomeCluster");
+    }
+
+    @Test
+    void shouldDetectDuplicateClusterNodeNames() {
+        // Given
+        assertThatThrownBy(() ->
+        // When
+        configParser.parseConfiguration("""
+                virtualClusters:
+                  demo1:
+                    targetCluster:
+                      bootstrap_servers: kafka.example:1234
+                    clusterNetworkAddressConfigProvider:
+                      type: PortPerBroker
+                      config:
+                        bootstrapAddress: cluster1:9192
+                        numberOfBrokerPorts: 1
+                        brokerAddressPattern: localhost
+                        brokerStartPort: 9193
+                  demo1:
+                    targetCluster:
+                      bootstrap_servers: magic-kafka.example:1234
+                    clusterNetworkAddressConfigProvider:
+                      type: PortPerBroker
+                      config:
+                        bootstrapAddress: cluster2:9193
+                        numberOfBrokerPorts: 1
+                        brokerAddressPattern: localhost
+                        brokerStartPort: 10193
+                """))
+                // Then
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasCauseInstanceOf(JsonMappingException.class) // Debatable to enforce the wrapped JsonMappingException
+                .cause()
+                .hasMessageStartingWith("Duplicate field 'demo1'");
+
+    }
 }
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/PortConflictDetectorTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/PortConflictDetectorTest.java
index 20babf69..1cd1c474 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/PortConflictDetectorTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/PortConflictDetectorTest.java
@@ -6,7 +6,7 @@
 
 package io.kroxylicious.proxy.internal;
 
-import java.util.Map;
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Stream;
@@ -36,121 +36,121 @@ class PortConflictDetectorTest {
         var privateUse = "192.168.0.1";
         return Stream.of(
                 Arguments.of("any:single conflict",
-                        createMockVirtualCluster(Set.of(9080), Set.of(), any),
-                        createMockVirtualCluster(Set.of(9080), Set.of(), any),
+                        createMockVirtualCluster("one", Set.of(9080), Set.of(), any),
+                        createMockVirtualCluster("two", Set.of(9080), Set.of(), any),
                         "The exclusive bind of port(s) 9080 to <any> would conflict with existing exclusive port bindings on <any>.",
                         null),
                 Arguments.of("any:many exclusive conflicts",
-                        createMockVirtualCluster(Set.of(9080, 9081), Set.of(), any),
-                        createMockVirtualCluster(Set.of(9080, 9081), Set.of(), any),
+                        createMockVirtualCluster("one", Set.of(9080, 9081), Set.of(), any),
+                        createMockVirtualCluster("two", Set.of(9080, 9081), Set.of(), any),
                         "The exclusive bind of port(s) 9080,9081 to <any> would conflict with existing exclusive port bindings on <any>.",
                         null),
                 Arguments.of("any:overlap produces exclusive conflict ",
-                        createMockVirtualCluster(Set.of(9080, 9081), Set.of(), any),
-                        createMockVirtualCluster(Set.of(9081, 9082), Set.of(), any),
+                        createMockVirtualCluster("one", Set.of(9080, 9081), Set.of(), any),
+                        createMockVirtualCluster("two", Set.of(9081, 9082), Set.of(), any),
                         "The exclusive bind of port(s) 9081 to <any> would conflict with existing exclusive port bindings on <any>.",
                         null),
                 Arguments.of("loopback:single exclusive conflict",
-                        createMockVirtualCluster(Set.of(9080), Set.of(), loopback),
-                        createMockVirtualCluster(Set.of(9080), Set.of(), loopback),
+                        createMockVirtualCluster("one", Set.of(9080), Set.of(), loopback),
+                        createMockVirtualCluster("two", Set.of(9080), Set.of(), loopback),
                         "The exclusive bind of port(s) 9080 to 127.0.0.1 would conflict with existing exclusive port bindings on 127.0.0.1.",
                         null),
                 Arguments.of("loopback/any:single exclusive conflict",
-                        createMockVirtualCluster(Set.of(9080), Set.of(), loopback),
-                        createMockVirtualCluster(Set.of(9080), Set.of(), any),
+                        createMockVirtualCluster("one", Set.of(9080), Set.of(), loopback),
+                        createMockVirtualCluster("two", Set.of(9080), Set.of(), any),
                         "The exclusive bind of port(s) 9080 to <any> would conflict with existing exclusive port bindings on 127.0.0.1.",
                         null),
                 Arguments.of("any/loopback:single exclusive conflict",
-                        createMockVirtualCluster(Set.of(9080), Set.of(), any),
-                        createMockVirtualCluster(Set.of(9080), Set.of(), loopback),
+                        createMockVirtualCluster("one", Set.of(9080), Set.of(), any),
+                        createMockVirtualCluster("two", Set.of(9080), Set.of(), loopback),
                         "The exclusive bind of port(s) 9080 to 127.0.0.1 would conflict with existing exclusive port bindings on <any>.",
                         null),
                 Arguments.of("any/loopback:single shared conflict",
-                        createMockVirtualCluster(Set.of(), Set.of(9080), any),
-                        createMockVirtualCluster(Set.of(), Set.of(9080), loopback),
+                        createMockVirtualCluster("one", Set.of(), Set.of(9080), any),
+                        createMockVirtualCluster("two", Set.of(), Set.of(9080), loopback),
                         "The shared bind of port(s) 9080 to 127.0.0.1 would conflict with existing shared port bindings on <any>.",
                         null),
                 Arguments.of("loopback/any:single shared conflict",
-                        createMockVirtualCluster(Set.of(), Set.of(9080), loopback),
-                        createMockVirtualCluster(Set.of(), Set.of(9080), any),
+                        createMockVirtualCluster("one", Set.of(), Set.of(9080), loopback),
+                        createMockVirtualCluster("two", Set.of(), Set.of(9080), any),
                         "The shared bind of port(s) 9080 to <any> would conflict with existing shared port bindings on 127.0.0.1.",
                         null),
                 Arguments.of("shared/exclusivity mismatch",
-                        createMockVirtualCluster(Set.of(), Set.of(9080), any),
-                        createMockVirtualCluster(Set.of(9080), Set.of(), any),
+                        createMockVirtualCluster("one", Set.of(), Set.of(9080), any),
+                        createMockVirtualCluster("two", Set.of(9080), Set.of(), any),
                         "The exclusive bind of port(s) 9080 to <any> would conflict with existing shared port bindings on <any>.",
                         null),
                 Arguments.of("exclusivity/shared mismatch",
-                        createMockVirtualCluster(Set.of(9080), Set.of(), any),
-                        createMockVirtualCluster(Set.of(), Set.of(9080), any),
+                        createMockVirtualCluster("one", Set.of(9080), Set.of(), any),
+                        createMockVirtualCluster("one", Set.of(), Set.of(9080), any),
                         "The shared bind of port(s) 9080 to <any> would conflict with existing exclusive port bindings on <any>.",
                         null),
                 Arguments.of("shared tls mismatch",
-                        createMockVirtualCluster(Set.of(), Set.of(9080), any, true),
-                        createMockVirtualCluster(Set.of(), Set.of(9080), any, false),
+                        createMockVirtualCluster("one", Set.of(), Set.of(9080), any, true),
+                        createMockVirtualCluster("two", Set.of(), Set.of(9080), any, false),
                         "The shared bind of port 9080 to <any> has conflicting TLS settings with existing port on the same interface.",
                         null),
                 Arguments.of("shared tls mismatch",
-                        createMockVirtualCluster(Set.of(), Set.of(9080), loopback, false),
-                        createMockVirtualCluster(Set.of(), Set.of(9080), loopback, true),
+                        createMockVirtualCluster("one", Set.of(), Set.of(9080), loopback, false),
+                        createMockVirtualCluster("two", Set.of(), Set.of(9080), loopback, true),
                         "The shared bind of port 9080 to 127.0.0.1 has conflicting TLS settings with existing port on the same interface.",
                         null),
                 Arguments.of("different exclusive ports",
-                        createMockVirtualCluster(Set.of(9080), Set.of(), any),
-                        createMockVirtualCluster(Set.of(9081), Set.of(), any),
+                        createMockVirtualCluster("one", Set.of(9080), Set.of(), any),
+                        createMockVirtualCluster("two", Set.of(9081), Set.of(), any),
                         null,
                         null),
                 Arguments.of("same shared ports",
-                        createMockVirtualCluster(Set.of(), Set.of(9080), any),
-                        createMockVirtualCluster(Set.of(), Set.of(9080), any),
+                        createMockVirtualCluster("one", Set.of(), Set.of(9080), any),
+                        createMockVirtualCluster("two", Set.of(), Set.of(9080), any),
                         null,
                         null),
                 Arguments.of("same shared shared ports different interfaces with differing tls",
-                        createMockVirtualCluster(Set.of(), Set.of(9080), loopback, true),
-                        createMockVirtualCluster(Set.of(), Set.of(9080), privateUse, false),
+                        createMockVirtualCluster("one", Set.of(), Set.of(9080), loopback, true),
+                        createMockVirtualCluster("two", Set.of(), Set.of(9080), privateUse, false),
                         null,
                         null),
                 Arguments.of("same exclusive port different interfaces",
-                        createMockVirtualCluster(Set.of(9080), Set.of(), privateUse),
-                        createMockVirtualCluster(Set.of(9080), Set.of(), loopback),
+                        createMockVirtualCluster("one", Set.of(9080), Set.of(), privateUse),
+                        createMockVirtualCluster("two", Set.of(9080), Set.of(), loopback),
                         null,
                         null),
                 Arguments.of("different exclusive ports, collides with other exclusive port on any interface",
-                        createMockVirtualCluster(Set.of(9080), Set.of(), any),
-                        createMockVirtualCluster(Set.of(9081), Set.of(), any),
+                        createMockVirtualCluster("one", Set.of(9080), Set.of(), any),
+                        createMockVirtualCluster("two", Set.of(9081), Set.of(), any),
                         "The exclusive bind of port(s) 9080 for virtual cluster 'one' to <any> would conflict with another (non-cluster) port binding",
                         new HostPort("0.0.0.0", 9080)),
                 Arguments.of("different exclusive ports on any interface, collides with other exclusive port on specific interface",
-                        createMockVirtualCluster(Set.of(9080), Set.of(), any),
-                        createMockVirtualCluster(Set.of(9081), Set.of(), any),
+                        createMockVirtualCluster("one", Set.of(9080), Set.of(), any),
+                        createMockVirtualCluster("two", Set.of(9081), Set.of(), any),
                         "The exclusive bind of port(s) 9080 for virtual cluster 'one' to <any> would conflict with another (non-cluster) port binding",
                         new HostPort(loopback, 9080)),
                 Arguments.of("different exclusive ports on specific interface, collides with other exclusive port on any interface",
-                        createMockVirtualCluster(Set.of(9080), Set.of(), loopback),
-                        createMockVirtualCluster(Set.of(9081), Set.of(), loopback),
+                        createMockVirtualCluster("one", Set.of(9080), Set.of(), loopback),
+                        createMockVirtualCluster("two", Set.of(9081), Set.of(), loopback),
                         "The exclusive bind of port(s) 9080 for virtual cluster 'one' to 127.0.0.1 would conflict with another (non-cluster) port binding",
                         new HostPort("0.0.0.0", 9080)),
                 Arguments.of("same shared ports, collides with other exclusive port on any interface",
-                        createMockVirtualCluster(Set.of(), Set.of(9080), any),
-                        createMockVirtualCluster(Set.of(), Set.of(9080), any),
+                        createMockVirtualCluster("one", Set.of(), Set.of(9080), any),
+                        createMockVirtualCluster("two", Set.of(), Set.of(9080), any),
                         "The shared bind of port(s) 9080 for virtual cluster 'one' to <any> would conflict with another (non-cluster) port binding",
                         new HostPort("0.0.0.0", 9080)),
                 Arguments.of("same shared shared ports different interfaces with differing tls, collides with other exclusive port",
-                        createMockVirtualCluster(Set.of(), Set.of(9080), loopback, true),
-                        createMockVirtualCluster(Set.of(), Set.of(9080), privateUse, false),
+                        createMockVirtualCluster("one", Set.of(), Set.of(9080), loopback, true),
+                        createMockVirtualCluster("two", Set.of(), Set.of(9080), privateUse, false),
                         "The shared bind of port(s) 9080 for virtual cluster 'one' to 127.0.0.1 would conflict with another (non-cluster) port binding",
                         new HostPort(loopback, 9080)),
                 Arguments.of("same exclusive port different interfaces, collides with other exclusive port",
-                        createMockVirtualCluster(Set.of(9080), Set.of(), privateUse),
-                        createMockVirtualCluster(Set.of(9080), Set.of(), loopback),
+                        createMockVirtualCluster("one", Set.of(9080), Set.of(), privateUse),
+                        createMockVirtualCluster("two", Set.of(9080), Set.of(), loopback),
                         "The exclusive bind of port(s) 9080 for virtual cluster 'two' to 127.0.0.1 would conflict with another (non-cluster) port binding",
                         new HostPort(loopback, 9080)));
     }
 
     @ParameterizedTest(name = "{0}")
     @MethodSource
-    public void portConflict(String name, VirtualCluster virtualCluster1, VirtualCluster virtualCluster2, String expectedMessageSuffix, HostPort otherExclusivePort) {
-        var clusters = Map.of("one", virtualCluster1, "two", virtualCluster2);
+    void portConflict(String name, VirtualCluster virtualCluster1, VirtualCluster virtualCluster2, String expectedMessageSuffix, HostPort otherExclusivePort) {
+        var clusters = List.of(virtualCluster1, virtualCluster2);
         Optional<HostPort> maybeOtherExclusivePort = Optional.ofNullable(otherExclusivePort);
         if (expectedMessageSuffix == null) {
             detector.validate(clusters, maybeOtherExclusivePort);
@@ -161,12 +161,13 @@ class PortConflictDetectorTest {
         }
     }
 
-    private static VirtualCluster createMockVirtualCluster(Set<Integer> exclusivePorts, Set<Integer> sharedPorts, String bindAddress) {
-        return createMockVirtualCluster(exclusivePorts, sharedPorts, bindAddress, false);
+    private static VirtualCluster createMockVirtualCluster(String clusterName, Set<Integer> exclusivePorts, Set<Integer> sharedPorts, String bindAddress) {
+        return createMockVirtualCluster(clusterName, exclusivePorts, sharedPorts, bindAddress, false);
     }
 
-    private static VirtualCluster createMockVirtualCluster(Set<Integer> exclusivePorts, Set<Integer> sharedPorts, String bindAddress, boolean tls) {
+    private static VirtualCluster createMockVirtualCluster(String clusterName, Set<Integer> exclusivePorts, Set<Integer> sharedPorts, String bindAddress, boolean tls) {
         VirtualCluster cluster = mock(VirtualCluster.class);
+        when(cluster.getClusterName()).thenReturn(clusterName);
         when(cluster.getExclusivePorts()).thenReturn(exclusivePorts);
         when(cluster.getSharedPorts()).thenReturn(sharedPorts);
         when(cluster.getBindAddress()).thenReturn(Optional.ofNullable(bindAddress));
@@ -174,4 +175,4 @@ class PortConflictDetectorTest {
         return cluster;
     }
 
-}
\ No newline at end of file
+}