Repository: kroxylicious/kroxylicious
Pull request #440 (2023-07-05 22:10:39)

Use command rather than which
### Type of change


- Bugfix
- Enhancement / new feature

### Description

Extend @k-wall's change to use `command -v` so that it handles the failure to resolve the command.

### Additional Context

When testing https://github.com/kroxylicious/kroxylicious/pull/436 I discovered things silently fail if one of the commands is missing. So this change pulls out a function to test for the command and exit with an error if its missing.

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [ ] Write tests
- [ ] Make sure all tests pass
- [ ] 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).



Patch:
diff --git a/.idea/compiler.xml b/.idea/compiler.xml
index 5ac0716f..77446034 100644
--- a/.idea/compiler.xml
+++ b/.idea/compiler.xml
@@ -1,6 +1,9 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
   <component name="CompilerConfiguration">
+    <excludeFromCompile>
+      <directory url="file://$PROJECT_DIR$/api/kroxylicious-filter-api-bom/src/main/resources/archetype-resources" includeSubdirectories="true" />
+    </excludeFromCompile>
     <annotationProcessing>
       <profile default="true" name="Default" enabled="true" />
       <profile name="Maven default annotation processors profile" enabled="true">
@@ -15,6 +18,7 @@
         <module name="kroxylicious-unit-test" />
         <module name="kroxylicious" />
         <module name="integrationtests" />
+        <module name="kroxylicious-sample" />
         <module name="integrationtests.test" />
         <module name="kroxylicious-krpc-plugin.main" />
         <module name="kroxylicious-schema-validation" />
@@ -25,6 +29,10 @@
         <module name="kroxylicious-krpc-plugin.test" />
       </profile>
     </annotationProcessing>
+    <bytecodeTargetLevel>
+      <module name="kroxylicious-bom" target="1.5" />
+      <module name="kroxylicious-filter-api-bom" target="1.5" />
+    </bytecodeTargetLevel>
   </component>
   <component name="JavacSettings">
     <option name="ADDITIONAL_OPTIONS_OVERRIDE">
@@ -40,6 +48,7 @@
       <module name="kroxylicious-krpc-plugin.test" options="-parameters" />
       <module name="kroxylicious-multitenant" options="-parameters" />
       <module name="kroxylicious-parent" options="-parameters" />
+      <module name="kroxylicious-sample" options="-parameters" />
       <module name="kroxylicious-schema-validation" options="-parameters" />
       <module name="kroxylicious-test-tools" options="-parameters" />
       <module name="kroxylicious-unit-test" options="-parameters" />
diff --git a/.idea/encodings.xml b/.idea/encodings.xml
index be37c2eb..00dc6788 100644
--- a/.idea/encodings.xml
+++ b/.idea/encodings.xml
@@ -14,6 +14,8 @@
     <file url="file://$PROJECT_DIR$/kroxylicious-additional-filters/kroxylicious-schema-validation/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/kroxylicious-additional-filters/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/kroxylicious-additional-filters/src/main/resources" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/kroxylicious-sample/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/kroxylicious-sample/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/kroxylicious-test-tools/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/kroxylicious-test-tools/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/kroxylicious-test-tools/target/generated-sources/krpc" charset="UTF-8" />
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0b02eb3d..4bf1ba19 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,8 +6,7 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 
 ## 0.3.0
 
-* [#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.
+* [#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 
@@ -26,8 +25,3 @@ 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 13f741ac..c267558c 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 # <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:$(portNumber) # <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,9 +79,12 @@ 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 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.
+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.
 
 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
deleted file mode 100644
index ebdfa482..00000000
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/BrokerAddressPatternUtils.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-
-package io.kroxylicious.proxy.internal.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 6709d939..a5e15032 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,6 +10,7 @@ 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;
 
@@ -19,19 +20,15 @@ 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 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 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 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>
@@ -40,8 +37,13 @@ import static io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider
 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;
 
@@ -54,7 +56,7 @@ public class PortPerBrokerClusterNetworkAddressConfigProvider implements Cluster
         this.bootstrapAddress = config.bootstrapAddress;
         this.brokerAddressPattern = config.brokerAddressPattern;
         this.brokerStartPort = config.brokerStartPort;
-        int numberOfBrokerPorts = config.numberOfBrokerPorts;
+        this.numberOfBrokerPorts = config.numberOfBrokerPorts;
         this.brokerEndPortExclusive = brokerStartPort + numberOfBrokerPorts;
 
         var exclusivePorts = IntStream.range(brokerStartPort, brokerEndPortExclusive).boxed().collect(Collectors.toCollection(HashSet::new));
@@ -81,8 +83,7 @@ public class PortPerBrokerClusterNetworkAddressConfigProvider implements Cluster
                                     brokerEndPortExclusive - 1,
                                     bootstrapAddress));
         }
-
-        return new HostPort(BrokerAddressPatternUtils.replaceLiteralNodeId(brokerAddressPattern, nodeId), port);
+        return HostPort.parse(brokerAddressPattern.replace(LITERAL_PORT_NUMBER, Integer.toString(port)).replace(LITERAL_NODE_ID, Integer.toString(nodeId)));
     }
 
     @Override
@@ -106,18 +107,10 @@ public class PortPerBrokerClusterNetworkAddressConfigProvider implements Cluster
             Objects.requireNonNull(bootstrapAddress, "bootstrapAddress cannot be null");
 
             this.bootstrapAddress = bootstrapAddress;
-            this.brokerAddressPattern = brokerAddressPattern != null ? brokerAddressPattern : bootstrapAddress.host();
+            this.brokerAddressPattern = brokerAddressPattern != null ? brokerAddressPattern : (bootstrapAddress.host() + ":" + LITERAL_PORT_NUMBER);
             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");
             }
@@ -129,9 +122,16 @@ 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()));
             });
 
-            validateStringContainsOnlyExpectedTokens(this.brokerAddressPattern, EXPECTED_TOKEN_SET, (token) -> {
-                throw new IllegalArgumentException("brokerAddressPattern contains an unexpected replacement token '" + token + "'");
-            });
+            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.");
+            }
+
         }
 
         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 abffd843..dbf60397 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,6 +7,7 @@
 package io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider;
 
 import java.util.Set;
+import java.util.regex.Pattern;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 
@@ -14,24 +15,15 @@ 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, shared, port for bootstrap and
+ * A ClusterNetworkAddressConfigProvider implementation that uses a single 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;
 
@@ -57,7 +49,7 @@ public class SniRoutingClusterNetworkAddressConfigProvider implements ClusterNet
             throw new IllegalArgumentException("nodeId cannot be less than zero");
         }
         // TODO: consider introducing an cache (LRU?)
-        return new HostPort(BrokerAddressPatternUtils.replaceLiteralNodeId(brokerAddressPattern, nodeId), bootstrapAddress.port());
+        return new HostPort(brokerAddressPattern.replace(LITERAL_NODE_ID, Integer.toString(nodeId)), bootstrapAddress.port());
     }
 
     @Override
@@ -88,17 +80,16 @@ public class SniRoutingClusterNetworkAddressConfigProvider implements ClusterNet
                 throw new IllegalArgumentException("brokerAddressPattern cannot be null");
             }
 
-            validatePortSpecifier(brokerAddressPattern, s -> {
-                throw new IllegalArgumentException("brokerAddressPattern cannot have port specifier.  Found port : " + s + " within " + brokerAddressPattern);
-            });
-
-            validateStringContainsOnlyExpectedTokens(brokerAddressPattern, EXPECTED_TOKEN_SET, (tok) -> {
-                throw new IllegalArgumentException("brokerAddressPattern contains an unexpected replacement token '" + tok + "'");
-            });
+            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.");
+            }
 
-            validateStringContainsRequiredTokens(brokerAddressPattern, EXPECTED_TOKEN_SET, (tok) -> {
-                throw new IllegalArgumentException("brokerAddressPattern must contain at least one nodeId replacement pattern '" + tok + "'");
-            });
+            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.");
+            }
 
             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 350bf541..c4aec545 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
+            brokerAddressPattern: kroxylicious-service:$(portNumber)
         logNetwork: false
         logFrames: false
     filters:
diff --git a/pom.xml b/pom.xml
index 2c284b1d..feb62bf3 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.100.0</sundr-builder-annotations.version>
+        <sundr-builder-annotations.version>0.95.0</sundr-builder-annotations.version>
         <findbugs.annotations.version>3.0.1u2</findbugs.annotations.version>
     </properties>
 
diff --git a/scripts/common.sh b/scripts/common.sh
index 7f7db1a9..c40858b0 100644
--- a/scripts/common.sh
+++ b/scripts/common.sh
@@ -1,29 +1,44 @@
-#!/bin/bash
+#!/usr/bin/env bash
 #
 # Copyright Kroxylicious Authors.
 #
 # Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 #
 
+set -Eu
+
 OS=$(uname)
 
+resolveCommand () {
+  local targetCommand=${1}
+  local resolvedCommand
+  resolvedCommand=$(command -v "${targetCommand}")
+  if [[ -z ${resolvedCommand} ]]; then
+    >&2 echo -e "\033[0;31m Unable to resolve path to ${targetCommand}\033[0m"
+    exit 127
+  else
+    echo "${resolvedCommand}"
+  fi
+}
+
 if [[ -z $CONTAINER_ENGINE ]]; then
   echo "Setting CONTAINER_ENGINE to default: docker"
-  CONTAINER_ENGINE=$(which docker)
+  CONTAINER_ENGINE=$(resolveCommand docker)
   export CONTAINER_ENGINE
 fi
-KUBECTL=$(which kubectl)
+
+KUBECTL=$(resolveCommand kubectl)
 export KUBECTL
-MINIKUBE=$(which minikube)
+MINIKUBE=$(resolveCommand minikube)
 export MINIKUBE
-KUSTOMIZE=$(which kustomize)
+KUSTOMIZE=$(resolveCommand kustomize)
 export KUSTOMIZE
 
 if [ "$OS" = 'Darwin' ]; then
   # for MacOS
-  SED=$(which gsed)
+  SED=$(resolveCommand gsed)
 else
   # for Linux and Windows
-  SED=$(which sed)
+  SED=$(resolveCommand sed)
 fi
 export SED
diff --git a/scripts/run-with-strimzi.sh b/scripts/run-with-strimzi.sh
index 98ffe3d5..b12bf16b 100755
--- a/scripts/run-with-strimzi.sh
+++ b/scripts/run-with-strimzi.sh
@@ -1,4 +1,4 @@
-#!/bin/bash
+#!/usr/bin/env bash
 #
 # Copyright Kroxylicious Authors.
 #


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 5cf87065..f2d07898 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
+                                brokerAddressPattern: localhost:$(portNumber)
                                 brokerStartPort: 9193
                         """),
                 Arguments.of("Virtual cluster (SniRouting)", """
@@ -52,7 +52,7 @@ class ConfigurationTest {
                               type: SniRouting
                               config:
                                 bootstrapAddress: cluster1:9192
-                                brokerAddressPattern: broker-$(nodeId)
+                                brokerAddressPattern: broker-$(nodeId):$(portNumber)
                         """),
                 Arguments.of("Downstream/Upstream TLS", """
                         virtualClusters:
@@ -75,7 +75,7 @@ class ConfigurationTest {
                               type: SniRouting
                               config:
                                 bootstrapAddress: cluster1:9192
-                                brokerAddressPattern: broker-$(nodeId)
+                                brokerAddressPattern: broker-$(nodeId):$(portNumber)
                         """),
                 Arguments.of("Filters", """
                         filters:
@@ -138,7 +138,7 @@ class ConfigurationTest {
                                         .endTargetCluster()
                                         .withClusterNetworkAddressConfigProvider(
                                                 new ClusterNetworkAddressConfigProviderDefinitionBuilder("SniRouting")
-                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId)")
+                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId):$(portNumber)")
                                                         .build())
                                         .build())
                                 .build(),
@@ -151,7 +151,7 @@ class ConfigurationTest {
                                       type: SniRouting
                                       config:
                                         bootstrapAddress: cluster1:9192
-                                        brokerAddressPattern: broker-$(nodeId)
+                                        brokerAddressPattern: broker-$(nodeId):$(portNumber)
                                 """),
                 Arguments.of("Downstream TLS",
                         new ConfigurationBuilder()
@@ -168,7 +168,7 @@ class ConfigurationTest {
                                         .endTls()
                                         .withClusterNetworkAddressConfigProvider(
                                                 new ClusterNetworkAddressConfigProviderDefinitionBuilder("SniRouting")
-                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId)")
+                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId):$(portNumber)")
                                                         .build())
                                         .build())
                                 .build(),
@@ -187,7 +187,7 @@ class ConfigurationTest {
                                       type: SniRouting
                                       config:
                                         bootstrapAddress: cluster1:9192
-                                        brokerAddressPattern: broker-$(nodeId)
+                                        brokerAddressPattern: broker-$(nodeId):$(portNumber)
                                 """),
                 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)")
+                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId):$(portNumber)")
                                                         .build())
                                         .build())
                                 .build(),
@@ -213,7 +213,7 @@ class ConfigurationTest {
                                       type: SniRouting
                                       config:
                                         bootstrapAddress: cluster1:9192
-                                        brokerAddressPattern: broker-$(nodeId)
+                                        brokerAddressPattern: broker-$(nodeId):$(portNumber)
                                 """),
                 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)")
+                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId):$(portNumber)")
                                                         .build())
                                         .build())
                                 .build(),
@@ -249,7 +249,7 @@ class ConfigurationTest {
                                       type: SniRouting
                                       config:
                                         bootstrapAddress: cluster1:9192
-                                        brokerAddressPattern: broker-$(nodeId)
+                                        brokerAddressPattern: broker-$(nodeId):$(portNumber)
                                 """),
                 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)")
+                                                        .withConfig("bootstrapAddress", "cluster1:9192", "brokerAddressPattern", "broker-$(nodeId):$(portNumber)")
                                                         .build())
                                         .build())
                                 .build(),
@@ -278,7 +278,7 @@ class ConfigurationTest {
                                       type: SniRouting
                                       config:
                                         bootstrapAddress: cluster1:9192
-                                        brokerAddressPattern: broker-$(nodeId)
+                                        brokerAddressPattern: broker-$(nodeId):$(portNumber)
                                 """)
 
         );
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 e68a1f48..6f212670 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", brokerStartPort, numberOfBrokerPorts);
+                    "localhost:$(portNumber)", brokerStartPort, numberOfBrokerPorts);
         });
     }
 
@@ -34,32 +34,24 @@ class PortPerBrokerClusterNetworkAddressConfigProviderTest {
     void bootstrapBrokerAddressCollision(int brokerStartPort, int numberOfBrokerPorts) {
         Assertions.assertThrows(IllegalArgumentException.class, () -> {
             new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"),
-                    "localhost", brokerStartPort, numberOfBrokerPorts);
+                    "localhost:$(portNumber)", brokerStartPort, numberOfBrokerPorts);
         });
     }
 
     @ParameterizedTest
-    @ValueSource(strings = { "portnotallowedinbrokerpattern:1234", "unrecognizedpattern$(foo)", "badpatterncapitalisedbadtoo$(NODEID)" })
+    @ValueSource(strings = { "notgood:$(foo)", "toomany$(portNumber)$(portNumber)", "toomanyrecursive$(portNumber(portNumber))", "shouty$(PORTNUMBER)badtoo",
+            "not.at.$(portNumber).end:1234" })
     @EmptySource
-    void invalidBrokerAddressPatterns(String input) {
+    void invalidBrokerPattern(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", 1236, 1));
+                        "localhost:$(portNumber)", 1236, 1));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("localhost:1236"));
         Assertions.assertThrows(IllegalArgumentException.class, () -> {
             provider.getBrokerAddress(1);
@@ -99,7 +91,7 @@ class PortPerBrokerClusterNetworkAddressConfigProviderTest {
     void definesExclusiveAndSharedCorrectly() {
         var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
                 new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"),
-                        "localhost", 1236, 2));
+                        "localhost:$(portNumber)", 1236, 2));
         assertThat(provider.getExclusivePorts()).containsExactlyInAnyOrder(1235, 1236, 1237);
         assertThat(provider.getSharedPorts()).isEmpty();
     }
@@ -108,7 +100,7 @@ class PortPerBrokerClusterNetworkAddressConfigProviderTest {
     void generatesBrokerAddresses() {
         var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
                 new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"),
-                        "localhost", 1236, 3));
+                        "localhost:$(portNumber)", 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"));
@@ -118,7 +110,7 @@ class PortPerBrokerClusterNetworkAddressConfigProviderTest {
     void fullyQualifiedHostNames() {
         var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
                 new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("cluster.kafka.example.com:1235"),
-                        "broker.kafka.example.com", 1236,
+                        "broker.kafka.example.com:$(portNumber)", 1236,
                         1238));
         assertThat(provider.getClusterBootstrapAddress()).isEqualTo(parse("cluster.kafka.example.com:1235"));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("broker.kafka.example.com:1236"));
@@ -129,7 +121,7 @@ class PortPerBrokerClusterNetworkAddressConfigProviderTest {
     void fullyQualifiedHostNamesWithNodeInterpolation() {
         var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
                 new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("cluster.kafka.example.com:1235"),
-                        "broker$(nodeId).kafka.example.com",
+                        "broker$(nodeId).kafka.example.com:$(portNumber)",
                         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 1eaae013..f0d503cb 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,20 +28,13 @@ class SniRoutingClusterNetworkAddressConfigProviderTest {
     }
 
     @ParameterizedTest
-    @ValueSource(strings = { "nonodetoken", "recursive$(nodeId$(nodeId))", "capitalisedrejected$(NODEID)", "noportalloweed$(nodeId):1234" })
+    @ValueSource(strings = { "notgood", "toomany$(nodeId)$(nodeId)", "toomanyrecursive$(nodeId$(nodeId))", "shouty$(NODEID)badtoo" })
     @NullAndEmptySource
-    void invalidBrokerAddressPattern(String input) {
+    void invalidBrokerPattern(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(