Repository: kroxylicious/kroxylicious (java)

Issue:
portNumber pattern is superfluous in PortPerBrokerClusterNetworkAddressConfigProvider
`PortPerBrokerClusterNetworkAddressConfigProvider` currently requires that brokerAddressPattern is a string in the form:

`host:$(portNumber)`

The replacement token `$(portNumber)` seems superfluous.  We could just define `brokerAddressPattern` as providing the `host` portion only.  The mechanics of the `#getBrokerAddress` can build the broker's using the HostPort(host, port) constructor.  It doesn't need substitution.

At the moment the implementation allows weird things like: `host:9$(portNumber)`, which really was never intended.

https://github.com/kroxylicious/kroxylicious/blob/bd6e7b1bda023a5f9d1010f58460689024031838/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProvider.java#L110C128-L110C147


Any objections to refactoring brokerAddressPattern to be a host name pattern only?

Pull request:
Fix #412: remove $(portNumber) replacement pattern in the PortPerBroker scheme.
### Type of change

_Select the type of your PR_

- Refactoring

### Description

Removes the `$(portNumber)` pattern from `brokerAddressPattern` parameter `PortPerBroker` scheme.

Why: it was superfluous configuration.  The only permitted value was `$(portNumber)`.  It was substituted for the
port number allocated to the broker from the port range.  Removal gives us simpler configuration without loss of function.

Also in the `SniRouting` case, we now validate that there is no port specified.   `SniRouting` always uses the port number
assigned to the virtual cluster's bootstrap. If a port number was specified, it actually lead to a failure later (client would choke on a badly formed addressed host:xxxx:xxxx. 

### Additional Context

_Why are you making this pull request?_

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [x] Write tests
- [x] Make sure all tests pass
- [x] Make sure all Sonar-Lint warnings are addressed or are justifiably ignored.
- [x] Update documentation
- [x] Reference relevant issue(s) and close them after merging
- [x] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4bf1ba19..0b02eb3d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,7 +6,8 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 
 ## 0.3.0
 
-* [#414](https://github.com/kroxylicious/kroxylicious/pull/414): Add kubernetes sample illustrating SNI based routing, downstream/upstream TLS and the use of certificates from cert-manager. 
+* [#412](https://github.com/kroxylicious/kroxylicious/issues/412): Remove $(portNumber) pattern from brokerAddressPattern for SniRouting and PortPerBroker schemes
+* [#414](https://github.com/kroxylicious/kroxylicious/pull/414): Add kubernetes sample illustrating SNI based routing, downstream/upstream TLS and the use of certificates from cert-manager.
 * [#392](https://github.com/kroxylicious/kroxylicious/pull/392): Introduce CompositeFilters
 * [#401](https://github.com/kroxylicious/kroxylicious/pull/401): Fix netty buffer leak when doing a short-circuit response
 * [#409](https://github.com/kroxylicious/kroxylicious/pull/409): Bump netty.version from 4.1.93.Final to 4.1.94.Final #409 
@@ -25,3 +26,8 @@ Filter Authors can now implement CompositeFilter if they want a single configura
 to the Filter chain. This enables them to write smaller, more focused Filter implementations but deliver them as a whole
 behaviour with a single block of configuration in the Kroxylicious configuration yaml. This interface is mutually exclusive
 with RequestFilter, ResponseFilter or any specific message Filter interfaces.
+
+In the kroxylicious config, the brokerAddressPattern parameter for the PortPerBroker scheme no longer accepts or requires
+:$(portNumber) suffix.  In addition, for the SniRouting scheme the config now enforces that there is no port specifier
+present on the brokerAddressPattern parameter. Previously, it was accepted but would lead to a failure later.
+
diff --git a/docs/deploying.adoc b/docs/deploying.adoc
index c267558c..13f741ac 100644
--- a/docs/deploying.adoc
+++ b/docs/deploying.adoc
@@ -66,11 +66,11 @@ The `PortPerBroker` scheme can be used with either clear text or TLS downstream
 clusterNetworkAddressConfigProvider:
   type: PortPerBroker
   config:
-    bootstrapAddress: mycluster.kafka.com:9192                                 # <1>
-    brokerAddressPattern: mybroker-$(nodeId).mycluster.kafka.com:$(portNumber) # <2>
-    brokerStartPort: 9193                                                      # <3>
-    numberOfBrokerPorts: 3                                                     # <4>
-    bindAddress: 192.168.0.1                                                   # <5>
+    bootstrapAddress: mycluster.kafka.com:9192                   # <1>
+    brokerAddressPattern: mybroker-$(nodeId).mycluster.kafka.com # <2>
+    brokerStartPort: 9193                                        # <3>
+    numberOfBrokerPorts: 3                                       # <4>
+    bindAddress: 192.168.0.1                                     # <5>
 ----
 <1> The hostname and port of the bootstrap that will be used by the Kafka Clients.
 <2> (Optional) The broker address pattern used to form the broker addresses.  If not defined, it defaults to the
@@ -79,12 +79,9 @@ hostname part of the `bootstrapAddress` and the port number allocated to the bro
 <4> (Optional) The _maximum_ number of brokers of ports that are permitted.  Defaults to 3.
 <5> (Optional) The bind address used when binding the ports. If undefined, all network interfaces will be bound.
 
-The `brokerAddressPattern` configuration parameters understands two replacement tokens.
-
-* this `$(portNumber)` is mandatory.  It is replaced by the port number computed for this broker.
-* the `$(nodeId)` token is optional.  If present, it will be replaced by the
-https://kafka.apache.org/documentation/#brokerconfigs_node.id[`node.id`] (or `broker.id`) assigned to the broker
-of the target cluster.
+The `brokerAddressPattern` configuration parameter understands the replacement token `$(nodeId)`. It is optional.
+If present, it will be replaced by the https://kafka.apache.org/documentation/#brokerconfigs_node.id[`node.id`] (or
+`broker.id`) assigned to the broker of the target cluster.
 
 For example if your configuration looks like the above and your cluster has three brokers, your Kafka Client will receive
 broker address information like this:
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/BrokerAddressPatternUtils.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/BrokerAddressPatternUtils.java
new file mode 100644
index 00000000..ebdfa482
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/BrokerAddressPatternUtils.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.internal.clusternetworkaddressconfigprovider;
+
+import java.util.Objects;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+final class BrokerAddressPatternUtils {
+    private static final String LITERAL_NODE_ID = "$(nodeId)";
+    private static final Pattern LITERAL_NODE_ID_PATTERN = Pattern.compile(Pattern.quote(LITERAL_NODE_ID));
+    private static final Pattern PORT_SPECIFIER_RE = Pattern.compile(":([0-9]+)$");
+    private static final Pattern TOKEN_RE = Pattern.compile("(\\$\\([^)]+\\))");
+    public static Set<String> EXPECTED_TOKEN_SET = Set.of(LITERAL_NODE_ID);
+
+    private BrokerAddressPatternUtils() {
+    }
+
+    private static Set<String> extractTokens(String stringWithTokens) {
+        var tokenMatcher = TOKEN_RE.matcher(stringWithTokens);
+        var tokens = new TreeSet<String>();
+        while (tokenMatcher.find()) {
+            var token = tokenMatcher.group(1);
+            tokens.add(token);
+        }
+        return tokens;
+
+    }
+
+    static void validateStringContainsOnlyExpectedTokens(String stringWithPossibleTokens,
+                                                         Set<String> allowedTokens,
+                                                         Consumer<String> disallowedTokenConsumer) {
+        Objects.requireNonNull(stringWithPossibleTokens);
+        Objects.requireNonNull(allowedTokens);
+        Objects.requireNonNull(disallowedTokenConsumer);
+        var tokens = extractTokens(stringWithPossibleTokens);
+        tokens.removeAll(allowedTokens);
+        tokens.forEach(disallowedTokenConsumer);
+    }
+
+    static void validateStringContainsRequiredTokens(String stringWithTokens,
+                                                     Set<String> requiredTokens,
+                                                     Consumer<String> requiredTokenAbsentConsumer) {
+        Objects.requireNonNull(stringWithTokens);
+        Objects.requireNonNull(requiredTokens);
+        Objects.requireNonNull(requiredTokenAbsentConsumer);
+
+        var tokens = extractTokens(stringWithTokens);
+        var requiredTokensCopy = new TreeSet<>(requiredTokens);
+        requiredTokensCopy.removeAll(tokens);
+
+        if (!requiredTokensCopy.isEmpty()) {
+            requiredTokensCopy.forEach(requiredTokenAbsentConsumer);
+        }
+    }
+
+    static void validatePortSpecifier(String address, Consumer<String> portNumber) {
+        var portMatcher = PORT_SPECIFIER_RE.matcher(address);
+        if (portMatcher.find()) {
+            portNumber.accept(portMatcher.group(1));
+        }
+    }
+
+    static String replaceLiteralNodeId(String brokerAddressPattern, int nodeId) {
+        return LITERAL_NODE_ID_PATTERN.matcher(brokerAddressPattern).replaceAll(Integer.toString(nodeId));
+    }
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProvider.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProvider.java
index a5e15032..6709d939 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProvider.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProvider.java
@@ -10,7 +10,6 @@ import java.util.Collections;
 import java.util.HashSet;
 import java.util.Objects;
 import java.util.Set;
-import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -20,15 +19,19 @@ import io.kroxylicious.proxy.config.BaseConfig;
 import io.kroxylicious.proxy.service.ClusterNetworkAddressConfigProvider;
 import io.kroxylicious.proxy.service.HostPort;
 
+import static io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.BrokerAddressPatternUtils.EXPECTED_TOKEN_SET;
+import static io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.BrokerAddressPatternUtils.validatePortSpecifier;
+import static io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.BrokerAddressPatternUtils.validateStringContainsOnlyExpectedTokens;
+
 /**
  * A ClusterNetworkAddressConfigProvider implementation that uses a separate port per broker endpoint.
  * <br/>
  * The following configuration is supported:
  * <ul>
  *    <li>{@code bootstrapAddress} (required) a {@link HostPort} defining the host and port of the bootstrap address.</li>
- *    <li>{@code brokerAddressPattern} (optional) an address pattern used to form broker addresses.  It is these address that are returned to the kafka client
- *    in the Metadata response so must be resolvable by the client.  Two patterns are supported: {@code $(portNumber)} (mandatory) and {@code $(nodeId)}
- *    which provide the port number and node id respectively.  If brokerAddressPattern is omitted, it defaulted it based on the host name of {@code bootstrapAddress}.</li>
+ *    <li>{@code brokerAddressPattern} (optional) an address pattern used to form broker addresses.  It is addresses made from this pattern that are returned to the kafka
+ *    client in the Metadata response so must be resolvable by the client.  One pattern is supported: {@code $(nodeId)} which interpolates the node id into the address.
+ *    If brokerAddressPattern is omitted, it defaulted it based on the host name of {@code bootstrapAddress}.</li>
  *    <li>{@code brokerStartPort} (optional) defines the starting range of port number that will be assigned to the brokers.  If omitted, it is defaulted to
  *    the port number of {@code bootstrapAddress + 1}.</li>
  *    <li>{@code numberOfBrokerPorts} (optional) defines the maximum number of broker ports that will be permitted. If omitted, it is defaulted to {$code 3}.</li>
@@ -37,13 +40,8 @@ import io.kroxylicious.proxy.service.HostPort;
 public class PortPerBrokerClusterNetworkAddressConfigProvider implements ClusterNetworkAddressConfigProvider {
 
     private final HostPort bootstrapAddress;
-
-    private static final String LITERAL_PORT_NUMBER = "$(portNumber)";
-    private static final String LITERAL_NODE_ID = "$(nodeId)";
-    private static final Pattern ANCHORED_PORT_NUMBER_TOKEN_RE = Pattern.compile(":" + Pattern.quote(LITERAL_PORT_NUMBER) + "$");
     private final String brokerAddressPattern;
     private final int brokerStartPort;
-    private final int numberOfBrokerPorts;
     private final Set<Integer> exclusivePorts;
     private final int brokerEndPortExclusive;
 
@@ -56,7 +54,7 @@ public class PortPerBrokerClusterNetworkAddressConfigProvider implements Cluster
         this.bootstrapAddress = config.bootstrapAddress;
         this.brokerAddressPattern = config.brokerAddressPattern;
         this.brokerStartPort = config.brokerStartPort;
-        this.numberOfBrokerPorts = config.numberOfBrokerPorts;
+        int numberOfBrokerPorts = config.numberOfBrokerPorts;
         this.brokerEndPortExclusive = brokerStartPort + numberOfBrokerPorts;
 
         var exclusivePorts = IntStream.range(brokerStartPort, brokerEndPortExclusive).boxed().collect(Collectors.toCollection(HashSet::new));
@@ -83,7 +81,8 @@ public class PortPerBrokerClusterNetworkAddressConfigProvider implements Cluster
                                     brokerEndPortExclusive - 1,
                                     bootstrapAddress));
         }
-        return HostPort.parse(brokerAddressPattern.replace(LITERAL_PORT_NUMBER, Integer.toString(port)).replace(LITERAL_NODE_ID, Integer.toString(nodeId)));
+
+        return new HostPort(BrokerAddressPatternUtils.replaceLiteralNodeId(brokerAddressPattern, nodeId), port);
     }
 
     @Override
@@ -107,10 +106,18 @@ public class PortPerBrokerClusterNetworkAddressConfigProvider implements Cluster
             Objects.requireNonNull(bootstrapAddress, "bootstrapAddress cannot be null");
 
             this.bootstrapAddress = bootstrapAddress;
-            this.brokerAddressPattern = brokerAddressPattern != null ? brokerAddressPattern : (bootstrapAddress.host() + ":" + LITERAL_PORT_NUMBER);
+            this.brokerAddressPattern = brokerAddressPattern != null ? brokerAddressPattern : bootstrapAddress.host();
             this.brokerStartPort = brokerStartPort != null ? brokerStartPort : (bootstrapAddress.port() + 1);
             this.numberOfBrokerPorts = numberOfBrokerPorts != null ? numberOfBrokerPorts : 3;
 
+            if (this.brokerAddressPattern.isBlank()) {
+                throw new IllegalArgumentException("brokerAddressPattern cannot be blank");
+            }
+
+            validatePortSpecifier(this.brokerAddressPattern, s -> {
+                throw new IllegalArgumentException("brokerAddressPattern cannot have port specifier.  Found port : " + s + " within " + this.brokerAddressPattern);
+            });
+
             if (this.brokerStartPort < 1) {
                 throw new IllegalArgumentException("brokerStartPort cannot be less than 1");
             }
@@ -122,16 +129,9 @@ public class PortPerBrokerClusterNetworkAddressConfigProvider implements Cluster
                 throw new IllegalArgumentException("the port used by the bootstrap address (%d) collides with the broker port range".formatted(bootstrapAddress.port()));
             });
 
-            var matcher = ANCHORED_PORT_NUMBER_TOKEN_RE.matcher(this.brokerAddressPattern);
-            if (!matcher.find()) {
-                throw new IllegalArgumentException("brokerAddressPattern must contain exactly one replacement pattern " + LITERAL_PORT_NUMBER + ". Found none.");
-            }
-            var stripped = matcher.replaceFirst("");
-            matcher = ANCHORED_PORT_NUMBER_TOKEN_RE.matcher(stripped);
-            if (matcher.find()) {
-                throw new IllegalArgumentException("brokerAddressPattern must contain exactly one replacement pattern " + LITERAL_PORT_NUMBER + ". Found too many.");
-            }
-
+            validateStringContainsOnlyExpectedTokens(this.brokerAddressPattern, EXPECTED_TOKEN_SET, (token) -> {
+                throw new IllegalArgumentException("brokerAddressPattern contains an unexpected replacement token '" + token + "'");
+            });
         }
 
         public HostPort getBootstrapAddress() {
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProvider.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProvider.java
index dbf60397..abffd843 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProvider.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProvider.java
@@ -7,7 +7,6 @@
 package io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider;
 
 import java.util.Set;
-import java.util.regex.Pattern;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 
@@ -15,15 +14,24 @@ import io.kroxylicious.proxy.config.BaseConfig;
 import io.kroxylicious.proxy.service.ClusterNetworkAddressConfigProvider;
 import io.kroxylicious.proxy.service.HostPort;
 
+import static io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.BrokerAddressPatternUtils.EXPECTED_TOKEN_SET;
+import static io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.BrokerAddressPatternUtils.validatePortSpecifier;
+import static io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.BrokerAddressPatternUtils.validateStringContainsOnlyExpectedTokens;
+import static io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.BrokerAddressPatternUtils.validateStringContainsRequiredTokens;
+
 /**
- * A ClusterNetworkAddressConfigProvider implementation that uses a single port for bootstrap and
+ * A ClusterNetworkAddressConfigProvider implementation that uses a single, shared, port for bootstrap and
  * all brokers.  SNI information is used to route the connection to the correct target.
- *
+ * <br/>
+ * The following configuration is required:
+ * <ul>
+ *    <li>{@code bootstrapAddress} a {@link HostPort} defining the host and port of the bootstrap address.</li>
+ *    <li>{@code brokerAddressPattern} an address pattern used to form broker addresses.  It is addresses made from this pattern that are returned to the kafka
+ *  *    client in the Metadata response so must be resolvable by the client.  One pattern is supported: {@code $(nodeId)} which interpolates the node id into the address.
+ * </ul>
  */
 public class SniRoutingClusterNetworkAddressConfigProvider implements ClusterNetworkAddressConfigProvider {
 
-    private static final String LITERAL_NODE_ID = "$(nodeId)";
-    private static final Pattern NODE_ID_TOKEN_RE = Pattern.compile(Pattern.quote(LITERAL_NODE_ID));
     private final HostPort bootstrapAddress;
     private final String brokerAddressPattern;
 
@@ -49,7 +57,7 @@ public class SniRoutingClusterNetworkAddressConfigProvider implements ClusterNet
             throw new IllegalArgumentException("nodeId cannot be less than zero");
         }
         // TODO: consider introducing an cache (LRU?)
-        return new HostPort(brokerAddressPattern.replace(LITERAL_NODE_ID, Integer.toString(nodeId)), bootstrapAddress.port());
+        return new HostPort(BrokerAddressPatternUtils.replaceLiteralNodeId(brokerAddressPattern, nodeId), bootstrapAddress.port());
     }
 
     @Override
@@ -80,16 +88,17 @@ public class SniRoutingClusterNetworkAddressConfigProvider implements ClusterNet
                 throw new IllegalArgumentException("brokerAddressPattern cannot be null");
             }
 
-            var matcher = NODE_ID_TOKEN_RE.matcher(brokerAddressPattern);
-            if (!matcher.find()) {
-                throw new IllegalArgumentException("brokerAddressPattern must contain exactly one nodeId replacement pattern " + LITERAL_NODE_ID + ". Found none.");
-            }
+            validatePortSpecifier(brokerAddressPattern, s -> {
+                throw new IllegalArgumentException("brokerAddressPattern cannot have port specifier.  Found port : " + s + " within " + brokerAddressPattern);
+            });
 
-            var stripped = matcher.replaceFirst("");
-            matcher = NODE_ID_TOKEN_RE.matcher(stripped);
-            if (matcher.find()) {
-                throw new IllegalArgumentException("brokerAddressPattern must contain exactly one nodeId replacement pattern " + LITERAL_NODE_ID + ". Found too many.");
-            }
+            validateStringContainsOnlyExpectedTokens(brokerAddressPattern, EXPECTED_TOKEN_SET, (tok) -> {
+                throw new IllegalArgumentException("brokerAddressPattern contains an unexpected replacement token '" + tok + "'");
+            });
+
+            validateStringContainsRequiredTokens(brokerAddressPattern, EXPECTED_TOKEN_SET, (tok) -> {
+                throw new IllegalArgumentException("brokerAddressPattern must contain at least one nodeId replacement pattern '" + tok + "'");
+            });
 
             this.bootstrapAddress = bootstrapAddress;
             this.brokerAddressPattern = brokerAddressPattern;
diff --git a/kubernetes-examples/portperbroker_plain/base/kroxylicious-config.yaml b/kubernetes-examples/portperbroker_plain/base/kroxylicious-config.yaml
index c4aec545..350bf541 100644
--- a/kubernetes-examples/portperbroker_plain/base/kroxylicious-config.yaml
+++ b/kubernetes-examples/portperbroker_plain/base/kroxylicious-config.yaml
@@ -22,7 +22,7 @@ data:
           type: PortPerBroker
           config:
             bootstrapAddress: localhost:9292
-            brokerAddressPattern: kroxylicious-service:$(portNumber)
+            brokerAddressPattern: kroxylicious-service
         logNetwork: false
         logFrames: false
     filters:
diff --git a/pom.xml b/pom.xml
index feb62bf3..2c284b1d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -48,7 +48,7 @@
         <slf4j.version>2.0.7</slf4j.version>
         <micrometer.version>1.11.1</micrometer.version>
         <zjsonpatch.version>0.4.14</zjsonpatch.version>
-        <sundr-builder-annotations.version>0.95.0</sundr-builder-annotations.version>
+        <sundr-builder-annotations.version>0.100.0</sundr-builder-annotations.version>
         <findbugs.annotations.version>3.0.1u2</findbugs.annotations.version>
     </properties>
 


Test patch:
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 f2d07898..5cf87065 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
@@ -40,7 +40,7 @@ class ConfigurationTest {
                               config:
                                 bootstrapAddress: cluster1:9192
                                 numberOfBrokerPorts: 1
-                                brokerAddressPattern: localhost:$(portNumber)
+                                brokerAddressPattern: localhost
                                 brokerStartPort: 9193
                         """),
                 Arguments.of("Virtual cluster (SniRouting)", """
@@ -52,7 +52,7 @@ class ConfigurationTest {
                               type: SniRouting
                               config:
                                 bootstrapAddress: cluster1:9192
-                                brokerAddressPattern: broker-$(nodeId):$(portNumber)
+                                brokerAddressPattern: broker-$(nodeId)
                         """),
                 Arguments.of("Downstream/Upstream TLS", """
                         virtualClusters:
@@ -75,7 +75,7 @@ class ConfigurationTest {
                               type: SniRouting
                               config:
                                 bootstrapAddress: cluster1:9192
-                                brokerAddressPattern: broker-$(nodeId):$(portNumber)
+                                brokerAddressPattern: broker-$(nodeId)
                         """),
                 Arguments.of("Filters", """
                         filters:
@@ -138,7 +138,7 @@ class ConfigurationTest {
                                         .endTargetCluster()
                                         .withClusterNetworkAddressConfigProvider(
                                                 new ClusterNetworkAddressConfigProviderDefinitionBuilder("SniRouting")
-                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId):$(portNumber)")
+                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId)")
                                                         .build())
                                         .build())
                                 .build(),
@@ -151,7 +151,7 @@ class ConfigurationTest {
                                       type: SniRouting
                                       config:
                                         bootstrapAddress: cluster1:9192
-                                        brokerAddressPattern: broker-$(nodeId):$(portNumber)
+                                        brokerAddressPattern: broker-$(nodeId)
                                 """),
                 Arguments.of("Downstream TLS",
                         new ConfigurationBuilder()
@@ -168,7 +168,7 @@ class ConfigurationTest {
                                         .endTls()
                                         .withClusterNetworkAddressConfigProvider(
                                                 new ClusterNetworkAddressConfigProviderDefinitionBuilder("SniRouting")
-                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId):$(portNumber)")
+                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId)")
                                                         .build())
                                         .build())
                                 .build(),
@@ -187,7 +187,7 @@ class ConfigurationTest {
                                       type: SniRouting
                                       config:
                                         bootstrapAddress: cluster1:9192
-                                        brokerAddressPattern: broker-$(nodeId):$(portNumber)
+                                        brokerAddressPattern: broker-$(nodeId)
                                 """),
                 Arguments.of("Upstream TLS - platform trust",
                         new ConfigurationBuilder()
@@ -199,7 +199,7 @@ class ConfigurationTest {
                                         .endTargetCluster()
                                         .withClusterNetworkAddressConfigProvider(
                                                 new ClusterNetworkAddressConfigProviderDefinitionBuilder("SniRouting")
-                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId):$(portNumber)")
+                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId)")
                                                         .build())
                                         .build())
                                 .build(),
@@ -213,7 +213,7 @@ class ConfigurationTest {
                                       type: SniRouting
                                       config:
                                         bootstrapAddress: cluster1:9192
-                                        brokerAddressPattern: broker-$(nodeId):$(portNumber)
+                                        brokerAddressPattern: broker-$(nodeId)
                                 """),
                 Arguments.of("Upstream TLS - trust from truststore",
                         new ConfigurationBuilder()
@@ -230,7 +230,7 @@ class ConfigurationTest {
                                         .endTargetCluster()
                                         .withClusterNetworkAddressConfigProvider(
                                                 new ClusterNetworkAddressConfigProviderDefinitionBuilder("SniRouting")
-                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId):$(portNumber)")
+                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId)")
                                                         .build())
                                         .build())
                                 .build(),
@@ -249,7 +249,7 @@ class ConfigurationTest {
                                       type: SniRouting
                                       config:
                                         bootstrapAddress: cluster1:9192
-                                        brokerAddressPattern: broker-$(nodeId):$(portNumber)
+                                        brokerAddressPattern: broker-$(nodeId)
                                 """),
                 Arguments.of("Upstream TLS - insecure",
                         new ConfigurationBuilder()
@@ -262,7 +262,7 @@ class ConfigurationTest {
                                         .endTargetCluster()
                                         .withClusterNetworkAddressConfigProvider(
                                                 new ClusterNetworkAddressConfigProviderDefinitionBuilder("SniRouting")
-                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId):$(portNumber)")
+                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId)")
                                                         .build())
                                         .build())
                                 .build(),
@@ -278,7 +278,7 @@ class ConfigurationTest {
                                       type: SniRouting
                                       config:
                                         bootstrapAddress: cluster1:9192
-                                        brokerAddressPattern: broker-$(nodeId):$(portNumber)
+                                        brokerAddressPattern: broker-$(nodeId)
                                 """)
 
         );
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProviderTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProviderTest.java
index 6f212670..e68a1f48 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProviderTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProviderTest.java
@@ -25,7 +25,7 @@ class PortPerBrokerClusterNetworkAddressConfigProviderTest {
     void badBrokerPortDefinition(int brokerStartPort, int numberOfBrokerPorts) {
         Assertions.assertThrows(IllegalArgumentException.class, () -> {
             new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"),
-                    "localhost:$(portNumber)", brokerStartPort, numberOfBrokerPorts);
+                    "localhost", brokerStartPort, numberOfBrokerPorts);
         });
     }
 
@@ -34,24 +34,32 @@ class PortPerBrokerClusterNetworkAddressConfigProviderTest {
     void bootstrapBrokerAddressCollision(int brokerStartPort, int numberOfBrokerPorts) {
         Assertions.assertThrows(IllegalArgumentException.class, () -> {
             new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"),
-                    "localhost:$(portNumber)", brokerStartPort, numberOfBrokerPorts);
+                    "localhost", brokerStartPort, numberOfBrokerPorts);
         });
     }
 
     @ParameterizedTest
-    @ValueSource(strings = { "notgood:$(foo)", "toomany$(portNumber)$(portNumber)", "toomanyrecursive$(portNumber(portNumber))", "shouty$(PORTNUMBER)badtoo",
-            "not.at.$(portNumber).end:1234" })
+    @ValueSource(strings = { "portnotallowedinbrokerpattern:1234", "unrecognizedpattern$(foo)", "badpatterncapitalisedbadtoo$(NODEID)" })
     @EmptySource
-    void invalidBrokerPattern(String input) {
+    void invalidBrokerAddressPatterns(String input) {
         assertThrows(IllegalArgumentException.class,
                 () -> new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("good:1235"), input, 1, 5));
     }
 
+    @ParameterizedTest
+    @ValueSource(strings = { "localhost", "127.0.0.1", "[2001:db8::1]", "kafka.example.com", "mybroker$(nodeId)", "mybroker$(nodeId).example.com",
+            "twice$(nodeId)allowed$(nodeId)too" })
+    void validBrokerAddressPatterns(String input) {
+        var config = new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("good:1235"), input, 1, 5);
+        assertThat(config).isNotNull();
+
+    }
+
     @Test
     void portsExhausted() {
         var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
                 new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"),
-                        "localhost:$(portNumber)", 1236, 1));
+                        "localhost", 1236, 1));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("localhost:1236"));
         Assertions.assertThrows(IllegalArgumentException.class, () -> {
             provider.getBrokerAddress(1);
@@ -91,7 +99,7 @@ class PortPerBrokerClusterNetworkAddressConfigProviderTest {
     void definesExclusiveAndSharedCorrectly() {
         var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
                 new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"),
-                        "localhost:$(portNumber)", 1236, 2));
+                        "localhost", 1236, 2));
         assertThat(provider.getExclusivePorts()).containsExactlyInAnyOrder(1235, 1236, 1237);
         assertThat(provider.getSharedPorts()).isEmpty();
     }
@@ -100,7 +108,7 @@ class PortPerBrokerClusterNetworkAddressConfigProviderTest {
     void generatesBrokerAddresses() {
         var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
                 new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"),
-                        "localhost:$(portNumber)", 1236, 3));
+                        "localhost", 1236, 3));
         assertThat(provider.getClusterBootstrapAddress()).isEqualTo(parse("localhost:1235"));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("localhost:1236"));
         assertThat(provider.getBrokerAddress(1)).isEqualTo(parse("localhost:1237"));
@@ -110,7 +118,7 @@ class PortPerBrokerClusterNetworkAddressConfigProviderTest {
     void fullyQualifiedHostNames() {
         var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
                 new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("cluster.kafka.example.com:1235"),
-                        "broker.kafka.example.com:$(portNumber)", 1236,
+                        "broker.kafka.example.com", 1236,
                         1238));
         assertThat(provider.getClusterBootstrapAddress()).isEqualTo(parse("cluster.kafka.example.com:1235"));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("broker.kafka.example.com:1236"));
@@ -121,7 +129,7 @@ class PortPerBrokerClusterNetworkAddressConfigProviderTest {
     void fullyQualifiedHostNamesWithNodeInterpolation() {
         var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
                 new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("cluster.kafka.example.com:1235"),
-                        "broker$(nodeId).kafka.example.com:$(portNumber)",
+                        "broker$(nodeId).kafka.example.com",
                         1236, 1238));
         assertThat(provider.getClusterBootstrapAddress()).isEqualTo(parse("cluster.kafka.example.com:1235"));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("broker0.kafka.example.com:1236"));
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProviderTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProviderTest.java
index f0d503cb..1eaae013 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProviderTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProviderTest.java
@@ -28,13 +28,20 @@ class SniRoutingClusterNetworkAddressConfigProviderTest {
     }
 
     @ParameterizedTest
-    @ValueSource(strings = { "notgood", "toomany$(nodeId)$(nodeId)", "toomanyrecursive$(nodeId$(nodeId))", "shouty$(NODEID)badtoo" })
+    @ValueSource(strings = { "nonodetoken", "recursive$(nodeId$(nodeId))", "capitalisedrejected$(NODEID)", "noportalloweed$(nodeId):1234" })
     @NullAndEmptySource
-    void invalidBrokerPattern(String input) {
+    void invalidBrokerAddressPattern(String input) {
         assertThrows(IllegalArgumentException.class,
                 () -> new SniRoutingClusterNetworkAddressConfigProviderConfig(parse("good:1235"), input));
     }
 
+    @ParameterizedTest
+    @ValueSource(strings = { "broker$(nodeId)", "twice$(nodeId)allowed$(nodeId)too", "broker$(nodeId).kafka.com" })
+    void validBrokerAddressPatterns(String input) {
+        var config = new SniRoutingClusterNetworkAddressConfigProviderConfig(parse("good:1235"), input);
+        assertThat(config).isNotNull();
+    }
+
     @Test
     void getBrokerAddress() {
         var provider = new SniRoutingClusterNetworkAddressConfigProvider(