Repository: kroxylicious/kroxylicious
Pull request #468 (2023-07-20 11:31:26)

Split up README.md into separate files.
### Type of change

- Refactoring
- Documentation

### Description

REAME.md is getting a bit unwieldy, let's refactor it to help the user find what they need.

### Additional Context

_Why are you making this pull request?_

### 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.
- [x] 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).


sonatype-lift[bot]: ## Sonatype Lift is retiring
Sonatype Lift will be retiring on <b>Sep 12, 2023</b>, with its analysis stopping on Aug 12, 2023. We understand that this news may come as a disappointment, and Sonatype is committed to helping you transition off it seamlessly. If you’d like to retain your data, please export your issues from the web console.
We are extremely grateful and thank you for your support over the years.

<b>[📖 Read about the impacts and timeline](https://www.sonatype.com/lift-retirement)</b>

Patch:
diff --git a/.idea/compiler.xml b/.idea/compiler.xml
index 77446034..13a191e9 100644
--- a/.idea/compiler.xml
+++ b/.idea/compiler.xml
@@ -1,9 +1,6 @@
 <?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">
@@ -23,7 +20,6 @@
         <module name="kroxylicious-krpc-plugin.main" />
         <module name="kroxylicious-schema-validation" />
         <module name="kroxylicious.test" />
-        <module name="kroxylicious-filter-api" />
         <module name="kroxylicious-krpc-plugin" />
         <module name="performance-tests" />
         <module name="kroxylicious-krpc-plugin.test" />
@@ -42,7 +38,6 @@
       <module name="kroxylicious" options="-parameters" />
       <module name="kroxylicious-additional-filters" options="-parameters" />
       <module name="kroxylicious-api" options="-parameters" />
-      <module name="kroxylicious-filter-api" options="-parameters" />
       <module name="kroxylicious-krpc-plugin" options="-parameters" />
       <module name="kroxylicious-krpc-plugin.main" options="-parameters" />
       <module name="kroxylicious-krpc-plugin.test" options="-parameters" />
diff --git a/.idea/encodings.xml b/.idea/encodings.xml
index 00dc6788..bd3cb76e 100644
--- a/.idea/encodings.xml
+++ b/.idea/encodings.xml
@@ -1,11 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
   <component name="Encoding">
-    <file url="file://$PROJECT_DIR$/api/kroxylicious-api/src/main/java" charset="UTF-8" />
-    <file url="file://$PROJECT_DIR$/api/kroxylicious-api/src/main/resources" charset="UTF-8" />
-    <file url="file://$PROJECT_DIR$/api/kroxylicious-filter-api/src/main/java" charset="UTF-8" />
-    <file url="file://$PROJECT_DIR$/api/kroxylicious-filter-api/src/main/resources" charset="UTF-8" />
-    <file url="file://$PROJECT_DIR$/api/kroxylicious-filter-api/target/generated-sources/krpc" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/integrationtests/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/integrationtests/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/kroxylicious-additional-filters/kroxylicious-multitenant/src/main/java" charset="UTF-8" />
@@ -14,6 +9,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-api/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/kroxylicious-api/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" />
diff --git a/CHANGELOG.md b/CHANGELOG.md
index fb92de2b..abb5b49b 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
 
+* [#416](https://github.com/kroxylicious/kroxylicious/issues/416): Eagerly expose broker endpoints on startup to allow existing client to reconnect (without connecting to bootstrap).
 * [#463](https://github.com/kroxylicious/kroxylicious/issues/463): deregister micrometer hooks, meters and the registry on shutdown
 * [#443](https://github.com/kroxylicious/kroxylicious/pull/443): Obtain upstream ApiVersions when proxy is not SASL offloading
 * [#412](https://github.com/kroxylicious/kroxylicious/issues/412): Remove $(portNumber) pattern from brokerAddressPattern for SniRouting and PortPerBroker schemes
diff --git a/Contributors.md b/CONTRIBUTORS.md
similarity index 100%
rename from Contributors.md
rename to CONTRIBUTORS.md
diff --git a/DEV_GUIDE.md b/DEV_GUIDE.md
new file mode 100644
index 00000000..c2fb49c0
--- /dev/null
+++ b/DEV_GUIDE.md
@@ -0,0 +1,291 @@
+# Development Guide for Kroxylicious
+
+This document gives a detailed breakdown of the various build processes and options for building the Kroxylicious from source.
+
+<!-- TOC -->
+* [Development Guide for Kroxylicious](#development-guide-for-kroxylicious)
+  * [Build Prerequisites](#build-prerequisites)
+  * [Prerequistes to run the kubernetes-examples](#prerequistes-to-run-the-kubernetes-examples)
+  * [Build](#build)
+    * [Formatting the Code](#formatting-the-code)
+  * [Run](#run)
+    * [Debugging](#debugging)
+  * [Running the kubernetes-examples](#running-the-kubernetes-examples)
+  * [IDE setup](#ide-setup)
+    * [Intellij](#intellij)
+  * [Setting Up in Windows Using WSL](#setting-up-in-windows-using-wsl)
+    * [Installing WSL](#installing-wsl)
+    * [Ensure appropriate tooling available](#ensure-appropriate-tooling-available)
+  * [Running Integration Tests on Podman](#running-integration-tests-on-podman)
+    * [DOCKER_HOST environment variable](#dockerhost-environment-variable)
+    * [Podman/Testcontainers incompatibility](#podmantestcontainers-incompatibility)
+    * [MacOS X](#macos-x)
+    * [Linux](#linux)
+    * [Verify that the fix is effective](#verify-that-the-fix-is-effective)
+  * [Rendering documentation](#rendering-documentation)
+<!-- TOC -->
+
+## Build Prerequisites
+
+- [JDK](https://openjdk.org/projects/jdk/17/) (version 17 and above) - Maven CLI
+- [`mvn`](https://maven.apache.org/index.html) (version 3.5 and above) - Maven CLI
+- [`docker`](https://docs.docker.com/install/) or [`podman`](https://podman.io/docs/installation) - Docker or Podman
+
+> :warning: **If you are using Podman please see [these notes](#running-integration-tests-on-podman) below**
+
+## Prerequistes to run the kubernetes-examples
+
+* User must have a [quay.io](https://www.quay.io) account and create a public repository named `kroxylicious`
+* Minikube [installed](https://minikube.sigs.k8s.io/docs/start)
+* kubectl [installed](https://kubernetes.io/docs/tasks/tools)
+* kustomize [installed](https://kubectl.docs.kubernetes.io/installation/kustomize/)
+* OSX users must have `gsed` [installed](https://formulae.brew.sh/formula/gnu-sed)
+* Docker engine [installed](https://docs.docker.com/engine/install) or [podman](https://podman.io/docs/installation) 
+
+
+## Build
+
+JDK version 20 or newer, and Apache Maven are required for building this project.
+
+Kroxylicious targets language level 17, except for the `integrationtests` module
+which targets 20 to access some new language features.
+
+Build the project like this:
+
+```
+$ mvn clean install
+```
+
+The running of the tests can be controlled with the following Maven properties:
+
+| property           | description                                                                               |
+|--------------------|-------------------------------------------------------------------------------------------|
+| `-DskipUTs=true`   | skip unit tests                                                                           |
+| `-DskipITs=true`   | skip integration tests                                                                    |
+| `-DskipTests=true` | skip all tests                                                                            |
+| `-Pdebug`          | enables logging so you can see what the Kafka clients, Proxy and in VM brokers are up to. |
+
+The kafka environment used by the integrations tests can be _defaulted_ with these two environment variables.
+
+| env var                       | default | description                                                                                                                             |
+|-------------------------------|---------|-----------------------------------------------------------------------------------------------------------------------------------------|
+| `TEST_CLUSTER_EXECUTION_MODE` | `IN_VM` | `IN_VM` or `CONTAINER`. if `IN_VM`, kafka will be run same virtual machines as the integration test. Otherwise containers will be used. |
+| `TEST_CLUSTER_KRAFT_MODE`     | `true`  | if true, kafka will be run in kraft mode.                                                                                               |
+
+When the integration-tests are run in `CONTAINER` mode, the kafka/zookeeper logs are written to a location specified by
+the `container.logs.dir`  system property. When run through Maven this is defaulted to `integrationtests/target/container-logs`.
+
+Pass the `-Dquick` option to skip all tests and non-essential plug-ins and create the output artifact as quickly as possible:
+
+```
+$ mvn clean verify -Dquick
+```
+
+Run the following command to format the source code and organize the imports as per the project's conventions:
+
+```
+$ mvn process-sources
+```
+
+Build with the `dist` profile for creating an executable JAR:
+
+```
+$ mvn clean verify -Pdist -Dquick
+```
+
+Run the following to add missing license headers e.g. when adding new source files:
+
+```
+$ mvn org.commonjava.maven.plugins:directory-maven-plugin:highest-basedir@resolve-rootdir license:format
+```
+
+### Formatting the Code
+No one likes to argue about code formatting in pull requests, as project we take the stance that if we can't automate the formatting we are not going to argue about it either. Having said that we don't want a mishmash of conflicting styles! So we attack this from multiple angles.
+
+1. Shared Code formatter settings. Included in the repo are code formatter settings for `Eclipse`, `InjtellJ` and `.editorconfig`.
+2. The Continuous Integration (CI) job building Pull Requests will fail if there is formatting which doesn't pass our agreed conventions
+3. We apply [Checkstyle](https://checkstyle.org/) validation to the project as well. You can find our [agreed ruleset](etc/checkstyle-custom_checks.xml) in the `etc` folder. We bind checkstyle to the `verify` phase of the build so `mvn clean verify` will validate the code is acceptable. 
+4. We also employ [impsort-maven-plugin](https://code.revelc.net/impsort-maven-plugin/) to keep import order consistent which will re-order imports as part of the maven build.
+5. We also have [formatter-maven-plugin](https://code.revelc.net/formatter-maven-plugin/) which will apply the project code style rules, this is driven from the Eclipse code formatter, as part of the maven build cycle.
+
+## Run
+
+Build with the `dist` profile as shown above, then execute this:
+
+```shell
+$ java -jar kroxylicious/target/kroxylicious-*-SNAPSHOT.jar --config {path-to-kroxylicious-config}
+```
+
+Or, to run with your own class path, run this instead:
+
+```shell
+$ java -cp {path-to-your-class-path}:kroxylicious/target/kroxylicious-*-SNAPSHOT.jar io.kroxylicious.proxy.Kroxylicious --config {path-to-kroxylicious-config}
+```
+
+To prevent the [following error](https://www.slf4j.org/codes.html#StaticLoggerBinder):
+
+```shell
+Failed to load class org.slf4j.impl.StaticLoggerBinder
+```
+
+Make sure to follow the [suggestions here](https://www.slf4j.org/codes.html#StaticLoggerBinder) to include one (and only one) of the suggested jars on the classpath.
+
+### Debugging
+Logging is turned off by default for better performance. In case you want to debug, logging should be turned on in the `example-proxy-config.yml` file:
+```yaml
+  logNetwork: true
+  logFrames: true
+```
+
+## Running the kubernetes-examples
+
+Kroxylicious can be containerised and run on Minikube against a [Strimzi](https://strimzi.io) managed Kafka cluster.
+
+Running:
+
+```shell
+minikube delete && QUAY_ORG=$your_quay_username$ ./scripts/run-with-strimzi.sh $kubernetes example directory$
+```
+where `$kubernetes example directory$` is replaced by a path to an example directory e.g. `./kubernetes-examples/portperbroker_plain`.
+
+This `run-with-strimzi.sh` script does the following:
+1. builds and pushes a kroxylicious image to quay.io
+2. starts minikube
+3. installs cert manager and strimzi
+4. installs a 3-node Kafka cluster using Strimzi into minikube
+5. installs kroxylicious into minikube, configured to proxy the cluster
+
+If you want to only build and push an image to quay.io you can run `PUSH_IMAGE=y QUAY_ORG=$your_quay_username$ ./scripts/deploy-image.sh`
+
+To change the container engine to podman set `CONTAINER_ENGINE=podman`
+
+## IDE setup
+
+### Intellij
+
+Open the root `pom.xml` as a project.
+
+## Setting Up in Windows Using WSL
+While Kroxylicious is a java application we've had reports of issues running the build natively on Windows and thus suggest using the Windows Subsystem for Linux (WSL) for development.
+### Installing WSL
+
+1. Enable the Windows Subsystem for Linux feature: To enable WSL, you need to enable the Windows Subsystem for Linux
+   feature in the Windows Features dialog.
+2. Install a Linux distribution from the Microsoft Store. The rest of these instructions assume a distribution (such as
+   Ubuntu) which uses `apt` for package management, but the process should be similar for distributions using other
+   package managers/ such as Fedora/`dnf`.
+3. Launch the Linux distribution and Configure the Linux environment: After launching the Linux distribution, you can
+   configure the environment by creating a user account and setting the password.
+   With these steps, you should now have WSL installed and configured on your Windows system.
+   For further assistance please see the [Microsoft documentation](https://learn.microsoft.com/en-us/windows/wsl/install)
+
+### Ensure appropriate tooling available
+1. Open the WSL window.
+2. Update the packages using
+    ```bash
+    sudo apt update
+    sudo apt upgrade
+    ```
+3. 
+    1. Check the Java version by typing
+      ```bash
+      java --version
+      ```
+      Expect output similar to: 
+      ```bash
+      > java --version
+   openjdk 19.0.2 2023-01-17
+   OpenJDK Runtime Environment Temurin-19.0.2+7 (build 19.0.2+7)
+   OpenJDK 64-Bit Server VM Temurin-19.0.2+7 (build 19.0.2+7, mixed mode, sharing)
+   ```
+    2. Update if needed: sample update command like:
+    ```bash
+    sudo apt update
+    sudo apt upgrade
+    sudo apt install openjdk-18-jre-headless
+    ```
+4. Ensure GIT is available
+   1. ```bash
+      git --version
+      ```
+      Expect a version string similar to `git version 2.37.1 (Apple Git-137.1)`
+   2. Follow the [WSL-git tutorial](https://learn.microsoft.com/en-us/windows/wsl/tutorials/wsl-git) if needed.
+5. Checkout Kroxylicious to `${kroxylicious-checkout}`
+6. Build & develop following the [standard build](#Build) instructions
+
+## Running Integration Tests on Podman
+
+### DOCKER_HOST environment variable
+
+On Linux, it maybe necessary to configure the `DOCKER_HOST` environment variable to allow the tests to correctly use test containers.
+
+```bash
+DOCKER_HOST=unix://$(podman info --format '{{.Host.RemoteSocket.Path}}')
+export DOCKER_HOST
+```
+
+### Podman/Testcontainers incompatibility
+
+There is an incompatibility between HTTP connection timeout expectations of 
+[testcontainers-java](https://github.com/testcontainers/testcontainers-java) and the Podman API. This
+can result in sporadic test failures when running the Integration Tests under Podman.  It manifests as
+failed or hanging REST API calls that leads to test failures and test hangs.
+
+It affects Linux and Mac OS X.
+On Linux it manifests as Http calls failing with a `Broken Pipe` exception. 
+Similarly on MacOS we see a `localhost:XXX failed to respond`.
+
+To workaround around the issue, tune the `service_timeout` so that the timeout is in sympathy with the
+expectations of the underlying HttpClient defaults.
+
+Do so by following these instructions.
+
+### MacOS X
+
+Start the `podman` machine as normal, then:
+
+```shell
+echo 'mkdir -p /etc/containers/containers.conf.d && printf "[engine]\nservice_timeout=91\n" > /etc/containers/containers.conf.d/service-timeout.conf' && systemctl restart podman.socket |  podman machine ssh --username root --
+```
+
+### Linux
+
+As a privileged user:
+
+```shell
+mkdir -p /etc/containers/containers.conf.d && printf "[engine]\nservice_timeout=91\n" > /etc/containers/containers.conf.d/service-timeout.conf && systemctl restart podman.socket
+```
+
+### Verify that the fix is effective
+
+On Linux, start this command:
+```shell
+socat - UNIX-CONNECT:$(podman info --format '{{.Host.RemoteSocket.Path}}')
+```
+On Mac OSX, Start this command:
+```shell
+time socat - UNIX-CONNECT:/var/run/docker.sock
+```
+the send this input (including the empty line):
+```
+GET /version HTTP/1.1
+Host: www.example.com
+
+```
+
+You'll see an API response.  If the service_timeout change is effective, the socat
+will continue for 3 mins.  If `socat` terminates after about 10 seconds, the workaround
+has been applied ineffectively.
+
+
+## Rendering documentation
+
+The `docs` directory has some user documentation written in [AsciiDoc](https://docs.asciidoctor.org/asciidoc/latest/) format.
+You can render it to HTML using:
+
+```bash
+mvn org.asciidoctor:asciidoctor-maven-plugin:process-asciidoc@convert-to-html
+```
+
+The output will be in `target/html/master.html`. 
+
diff --git a/benchmarking.md b/PERFORMANCE.md
similarity index 92%
rename from benchmarking.md
rename to PERFORMANCE.md
index 5b95e15a..6c192a08 100644
--- a/benchmarking.md
+++ b/PERFORMANCE.md
@@ -1,4 +1,4 @@
-# Benchmarking
+# Performance
 
 This document describes how to run some basic performance tests for the proxy.
 
@@ -146,3 +146,17 @@ new ProduceRecordTransformationFilter(
 10000000 records sent, 60679.611650 records/sec (59.26 MB/sec), 25.00 ms avg latency, 1080.00 ms max latency, 1 ms 50th, 170 ms 95th, 601 ms 99th, 992 ms 99.9th.
 
 ```
+
+### Jenkins pipeline for performance
+
+When a PR is created and the performance tests are needed, if you are a member of
+[Developers](https://github.com/orgs/kroxylicious/teams/developers), you may add the following comment into the PR to trigger the run.
+
+```
+@strimzi-ci run perf
+```
+
+It will launch the `kroxylicious-performance-tests-pr` build, that will insert a comment with a summary into the PR comparing the results with the previous execution.
+
+In case a manual execution needs to be done, this job is ready for that `kroxylicious-performance-tests`, where
+some parameters can be configurable (NUMBER_OF_MESSAGES, MESSAGE_SIZE, PRODUCER_PROPERTIES)
diff --git a/README.md b/README.md
index b2d18c0a..f846efe7 100644
--- a/README.md
+++ b/README.md
@@ -6,23 +6,12 @@ addressing use cases such as multi-tenancy, schema validation, or encryption.
 <!-- TOC -->
 * [Kroxylicious](#kroxylicious)
   * [Quick Links](#quick-links)
-  * [Build](#build)
-    * [Formatting](#formatting)
-  * [Run](#run)
-    * [Run natively](#run-natively)
-      * [Debugging](#debugging)
-    * [Run on Minikube](#run-on-minikube)
-  * [Samples](#samples)
-  * [Rendering documentation](#rendering-documentation)
+  * [License](#license)
+  * [Developer Guide](#developer-guide)
+  * [Releasing this project](#releasing-this-project)
   * [Performance Testing](#performance-testing)
-    * [Jenkins pipeline for performance](#jenkins-pipeline-for-performance)
+  * [Kroxylicious Samples](#kroxylicious-samples)
   * [Architecture Monitoring](#architecture-monitoring)
-  * [IntelliJ setup](#intellij-setup)
-  * [License](#license)
-  * [Setting Up in Windows Using WSL](#setting-up-in-windows-using-wsl)
-    * [Installing WSL](#installing-wsl)
-    * [Ensure appropriate tooling available](#ensure-appropriate-tooling-available)
-  * [Releasing the project](#releasing-the-project)
 <!-- TOC -->
 
 ## Quick Links
@@ -31,162 +20,25 @@ addressing use cases such as multi-tenancy, schema validation, or encryption.
 - [GitHub design and discussion](https://github.com/kroxylicious/design)
 - [Community Slack chat](https://kroxylicious.slack.com/)
 
-## Build
-
-JDK version 20 or newer, and Apache Maven are required for building this project.
-
-Kroxylicious targets language level 17, except for the `integrationtests` module
-which targets 20 to access some new language features.
-
-Build the project like this:
-
-```
-$ mvn clean install
-```
-
-The running of the tests can be controlled with the following Maven properties:
-
-| property           | description                                                                               |
-|--------------------|-------------------------------------------------------------------------------------------|
-| `-DskipUTs=true`   | skip unit tests                                                                           |
-| `-DskipITs=true`   | skip integration tests                                                                    |
-| `-DskipTests=true` | skip all tests                                                                            |
-| `-Pdebug`          | enables logging so you can see what the Kafka clients, Proxy and in VM brokers are up to. |
-
-The kafka environment used by the integrations tests can be _defaulted_ with these two environment variables.
-
-| env var                       | default | description                                                                                                                             |
-|-------------------------------|---------|-----------------------------------------------------------------------------------------------------------------------------------------|
-| `TEST_CLUSTER_EXECUTION_MODE` | `IN_VM` | `IN_VM` or `CONTAINER`. if `IN_VM`, kafka will be run same virtual machines as the integration test. Otherwise containers will be used. |
-| `TEST_CLUSTER_KRAFT_MODE`     | `true`  | if true, kafka will be run in kraft mode.                                                                                               |
-
-When the integration-tests are run in `CONTAINER` mode, the kafka/zookeeper logs are written to a location specified by
-the `container.logs.dir`  system property. When run through Maven this is defaulted to `integrationtests/target/container-logs`.
-
-Pass the `-Dquick` option to skip all tests and non-essential plug-ins and create the output artifact as quickly as possible:
-
-```
-$ mvn clean verify -Dquick
-```
-
-Run the following command to format the source code and organize the imports as per the project's conventions:
-
-```
-$ mvn process-sources
-```
-
-Build with the `dist` profile for creating an executable JAR:
-
-```
-$ mvn clean verify -Pdist -Dquick
-```
-
-Run the following to add missing license headers e.g. when adding new source files:
-
-```
-$ mvn org.commonjava.maven.plugins:directory-maven-plugin:highest-basedir@resolve-rootdir license:format
-```
-
-### Formatting
-No one likes to argue about code formatting in pull requests, as project we take the stance that if we can't automate the formatting we are not going to argue about it either. Having said that we don't want a mishmash of conflicting styles! So we attack this from multiple angles.
-
-1. Shared Code formatter settings. Included in the repo are code formatter settings for `Eclipse`, `InjtellJ` and `.editorconfig`.
-2. The Continuous Integration (CI) job building Pull Requests will fail if there is formatting which doesn't pass our agreed conventions
-3. We apply [Checkstyle](https://checkstyle.org/) validation to the project as well. You can find our [agreed ruleset](etc/checkstyle-custom_checks.xml) in the `etc` folder. We bind checkstyle to the `verify` phase of the build so `mvn clean verify` will validate the code is acceptable. 
-4. We also employ [impsort-maven-plugin](https://code.revelc.net/impsort-maven-plugin/) to keep import order consistent which will re-order imports as part of the maven build.
-5. We also have [formatter-maven-plugin](https://code.revelc.net/formatter-maven-plugin/) which will apply the project code style rules, this is driven from the Eclipse code formatter, as part of the maven build cycle.
-
-## Run
-
-### Run natively
-
-Build with the `dist` profile as shown above, then execute this:
-
-```
-$ java -jar kroxylicious/target/kroxylicious-*-SNAPSHOT.jar --config {path-to-kroxylicious-config}
-```
-
-Or, to run with your own class path, run this instead:
-
-```
-$ java -cp {path-to-your-class-path}:kroxylicious/target/kroxylicious-*-SNAPSHOT.jar io.kroxylicious.proxy.Kroxylicious --config {path-to-kroxylicious-config}
-```
-
-To prevent the [following error](https://www.slf4j.org/codes.html#StaticLoggerBinder):
-
-```
-Failed to load class org.slf4j.impl.StaticLoggerBinder
-```
-
-Make sure to follow the [suggestions here](https://www.slf4j.org/codes.html#StaticLoggerBinder) to include one (and only one) of the suggested jars on the classpath.
-
-#### Debugging
-Logging is turned off by default for better performance. In case you want to debug, logging should be turned on in the `example-proxy-config.yml` file:
-```yaml
-  logNetwork: true
-  logFrames: true
-```
-
-### Run on Minikube
-
-Kroxylicious can be containerised and run on Minikube against a [Strimzi](https://strimzi.io) managed Kafka cluster.
-
-**Prerequisites**
-* User must have a [quay.io](https://www.quay.io) account and create a public repository named `kroxylicious`
-* Minikube [installed](https://minikube.sigs.k8s.io/docs/start)
-* kubectl [installed](https://kubernetes.io/docs/tasks/tools)
-* kustomize [installed](https://kubectl.docs.kubernetes.io/installation/kustomize/)
-* OSX users must have `gsed` [installed](https://formulae.brew.sh/formula/gnu-sed)
-* Docker engine [installed](https://docs.docker.com/engine/install) or [podman](https://podman.io/docs/installation) 
-
-Running:
-
-```bash
-minikube delete && QUAY_ORG=$your_quay_username$ ./scripts/run-with-strimzi.sh $kubernetes example directory$
-```
-where `$kubernetes example directory$` is replaced by a path to an example directory e.g. `./kubernetes-examples/portperbroker_plain`.
-
-This `run-with-strimzi.sh` script does the following:
-1. builds and pushes a kroxylicious image to quay.io
-2. starts minikube
-3. installs cert manager and strimzi
-4. installs a 3-node Kafka cluster using Strimzi into minikube
-5. installs kroxylicious into minikube, configured to proxy the cluster
-
-If you want to only build and push an image to quay.io you can run `PUSH_IMAGE=y QUAY_ORG=$your_quay_username$ ./scripts/deploy-image.sh`
-
-To change the container engine to podman set `CONTAINER_ENGINE=podman`
-
-## Samples
+## License
 
-See [kroxylicious-sample](kroxylicious-sample) to learn more about sample filters. Try them out and customise them for a hands-on introduction to custom filters in Kroxylicious.
+This code base is available under the [Apache License, version 2](LICENSE).
 
-## Rendering documentation
+## Developer Guide
 
-The `docs` directory has some user documentation written in [AsciiDoc](https://docs.asciidoctor.org/asciidoc/latest/) format.
-You can render it to HTML using:
+See the [Developer Guide](DEV_GUIDE.md).
 
-```
-mvn org.asciidoctor:asciidoctor-maven-plugin:process-asciidoc@convert-to-html
-```
+## Releasing this project
 
-The output will be in `target/html/master.html`. 
+See the [Release Guide](RELEASING.md)
 
 ## Performance Testing
 
-See [benchmarking.md](benchmarking.md) for information on running basic performance tests for this proxy.
-
-### Jenkins pipeline for performance
-When a PR is created and the performance tests are needed, the following comment shall be added into the PR:
-
-```
-@strimzi-ci run perf
-```
+See the [Performance Guide](PERFORMANCE.md) for information on running basic performance tests for this proxy.
 
-It will launch the `kroxylicious-performance-tests-pr` build, that will insert a comment with a summary into the PR comparing the results with the previous execution.
+## Kroxylicious Samples
 
-In case a manual execution needs to be done, this job is ready for that `kroxylicious-performance-tests`, where
-some parameters can be configurable (NUMBER_OF_MESSAGES, MESSAGE_SIZE, PRODUCER_PROPERTIES)
+See [kroxylicious-sample](kroxylicious-sample) to learn more about sample filters. Try them out and customise them for a hands-on introduction to custom filters in Kroxylicious.
 
 ## Architecture Monitoring
 
@@ -197,70 +49,12 @@ Any new package relationships need to be registered there.
 
 To verify whether the code base adheres to that target model, run the following:
 
-```
+```bash
 $ mvn clean verify -Dquick -Parchitecture-check -pl kroxylicious
 ```
 
 In case of any architecture violations, the actual architecture can be visualized using GraphViz like so:
 
-```
+```bash
 $ dot -Tpng kroxylicious/target/generated-sources/annotations/deptective.dot > kroxylicious-arch.png
 ```
-
-## IntelliJ setup
-
-Open the root `pom.xml` as a project.
-
-## License
-
-This code base is available under the Apache License, version 2.
-
-## Setting Up in Windows Using WSL
-While Kroxylicious is a java application we've had reports of issues running the build natively on Windows and thus suggest using the Windows Subsystem for Linux (WSL) for development.
-### Installing WSL
-
-1. Enable the Windows Subsystem for Linux feature: To enable WSL, you need to enable the Windows Subsystem for Linux
-   feature in the Windows Features dialog.
-2. Install a Linux distribution from the Microsoft Store. The rest of these instructions assume a distribution (such as
-   Ubuntu) which uses `apt` for package management, but the process should be similar for distributions using other
-   package managers/ such as Fedora/`dnf`.
-3. Launch the Linux distribution and Configure the Linux environment: After launching the Linux distribution, you can
-   configure the environment by creating a user account and setting the password.
-   With these steps, you should now have WSL installed and configured on your Windows system.
-   For further assistance please see the [Microsoft documentation](https://learn.microsoft.com/en-us/windows/wsl/install)
-### Ensure appropriate tooling available
-1. Open the WSL window.
-2. Update the packages using
-    ```bash
-    sudo apt update
-    sudo apt upgrade
-    ```
-3. 
-    1. Check the Java version by typing
-      ```bash
-      java --version
-      ```
-      Expect output similar to: 
-      ```bash
-      > java --version
-   openjdk 19.0.2 2023-01-17
-   OpenJDK Runtime Environment Temurin-19.0.2+7 (build 19.0.2+7)
-   OpenJDK 64-Bit Server VM Temurin-19.0.2+7 (build 19.0.2+7, mixed mode, sharing)
-   ```
-    2. Update if needed: sample update command like:
-    ```bash
-    sudo apt update
-    sudo apt upgrade
-    sudo apt install openjdk-18-jre-headless
-    ```
-4. Ensure GIT is available
-   1. ```bash
-      git --version
-      ```
-      Expect a version string similar to `git version 2.37.1 (Apple Git-137.1)`
-   2. Follow the [WSL-git tutorial](https://learn.microsoft.com/en-us/windows/wsl/tutorials/wsl-git) if needed.
-5. Checkout Kroxylicious to `${kroxylicious-checkout}`
-6. Build & develop following the [standard build](#Build) instructions
-
-## Releasing the project
-See [The release guide](docs/Releasing.adoc)
\ No newline at end of file
diff --git a/docs/Releasing.adoc b/RELEASING.md
similarity index 88%
rename from docs/Releasing.adoc
rename to RELEASING.md
index e05ca5eb..32701b19 100644
--- a/docs/Releasing.adoc
+++ b/RELEASING.md
@@ -1,13 +1,9 @@
 = Releasing Kroxylicious
-3.0, July 29, 2022: AsciiDoc article template
-:toc:
-:icons: font
-:url-quickref: https://docs.asciidoctor.org/asciidoc/latest/syntax-quick-reference/
 
 A quick and dirty guide to actually releasing builds for consumption via maven central.
 
-== Requirements
-=== Sonatype/Maven Central
+## Requirements
+### Sonatype/Maven Central
 
 * A Maven Central/Sonatype account. Accounts are created via their https://issues.sonatype.org/secure/Signup!default.jspa[JIRA signup]
 * Your Sonatype account needs to be granted publication permissions for the Kroxylicous project.
@@ -15,28 +11,28 @@ A quick and dirty guide to actually releasing builds for consumption via maven c
 
 Signing the builds with a GPG key is a requirement for publishing to Maven Central.
 
-=== PGP
+### PGP
 * Ensure your PGP key is published to a public key server (for example keys.openpgp.org)
 * Get your PGP key signed by the Kroxylicious project key. Rather than share a single key between project members (which has issues with revocation and re-issuing) we sign the releases individually using a key signed by the project key as a mark of trust.
 * A copy of your keys short ID.
 
 TIP: To get the short ID for your key `gpg2 --show-keys --keyid-format short $\{YOUR_KEY_LONG_ID}`
 
-=== Development environment
+### Development environment
 
 * Once logged in to the https://s01.oss.sonatype.org/[Sonatype Nexus UI], create a User Token and configure this in the https://maven.apache.org/settings.html#servers[`<servers>`] element o  `.m2/settings.xml`.  This is required in order to allow Maven to push the artifacts to Nexus. 
 * The release scripts will make use of the GitHub CLI https://cli.github.com/ if present, so you may wish to install to reduces the number of chores.
 
 
-== Running a release
+## Running a release
 
 The project is release is two parts, the API and the Framework. There are separate release scripts beneath `./scripts` for each of these.
 
-```
+```shell
 scripts/release-api.sh <RELEASE_VERSION>
 ```
 
-```
+```shell
 scripts/release-framework.sh <RELEASE_VERSION>
 ```
 
diff --git a/api/kroxylicious-api/pom.xml b/api/kroxylicious-api/pom.xml
deleted file mode 100644
index 049cf65d..00000000
--- a/api/kroxylicious-api/pom.xml
+++ /dev/null
@@ -1,51 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-
-    Copyright Kroxylicious Authors.
-
-    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
-
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <modelVersion>4.0.0</modelVersion>
-
-    <parent>
-        <groupId>io.kroxylicious</groupId>
-        <artifactId>kroxylicious-parent</artifactId>
-        <version>0.3.0-SNAPSHOT</version>
-        <relativePath>../../pom.xml</relativePath>
-    </parent>
-
-    <artifactId>kroxylicious-api</artifactId>
-    <version>0.3.0-SNAPSHOT</version>
-    <packaging>jar</packaging>
-
-    <name>Kroxylicious API</name>
-    <description>generic APIs used throughout Kroxylicious</description>
-
-    <dependencies>
-        <!-- third party dependencies - runtime and compile -->
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-annotations</artifactId>
-        </dependency>
-        <!-- third party dependencies - test -->
-        <dependency>
-            <groupId>org.assertj</groupId>
-            <artifactId>assertj-core</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.mockito</groupId>
-            <artifactId>mockito-core</artifactId>
-            <scope>test</scope>
-        </dependency>
-    </dependencies>
-</project>
diff --git a/api/kroxylicious-filter-api/kroxylicious-filter-api.iml b/api/kroxylicious-filter-api/kroxylicious-filter-api.iml
deleted file mode 100644
index 9e3449c9..00000000
--- a/api/kroxylicious-filter-api/kroxylicious-filter-api.iml
+++ /dev/null
@@ -1,6 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<module version="4">
-  <component name="CheckStyle-IDEA-Module" serialisationVersion="2">
-    <option name="activeLocationsIds" />
-  </component>
-</module>
\ No newline at end of file
diff --git a/docs/custom-filters.adoc b/docs/custom-filters.adoc
index 226fd93f..88cbd5af 100644
--- a/docs/custom-filters.adoc
+++ b/docs/custom-filters.adoc
@@ -1,6 +1,5 @@
 :github: https://github.com/kroxylicious/kroxylicious
 :api-javadoc: https://javadoc.io/doc/io.kroxylicious/kroxylicious-api/latest
-:filter-api-javadoc: https://javadoc.io/doc/io.kroxylicious/kroxylicious-filter-api/latest
 :source-highlighter: pygments
 :java-17-javadoc: https://docs.oracle.com/en/java/javase/17/docs/api
 
@@ -25,9 +24,9 @@ You can find them {github}/tree/main/kroxylicious-sample[here] for a hands-on in
 == API docs
 
 Custom Protocol Filters are built by implementing interfaces supplied by the
-{github}/tree/main/api/kroxylicious-filter-api[kroxylicious-filter-api] module
-(https://mvnrepository.com/artifact/io.kroxylicious/kroxylicious-filter-api[io.kroxylicious:kroxylicious-filter-api] on
-maven central). You can view the javadoc {filter-api-javadoc}/io/kroxylicious/proxy/filter/package-summary.html[here].
+{github}/tree/main/api/kroxylicious-api[kroxylicious-api] module
+(https://mvnrepository.com/artifact/io.kroxylicious/kroxylicious-api[io.kroxylicious:kroxylicious-api] on
+maven central). You can view the javadoc {api-javadoc}/io/kroxylicious/proxy/filter/package-summary.html[here].
 
 == Dependencies
 
@@ -76,7 +75,7 @@ intercept only the request, or only the response, or both the request and respon
 ==== Examples
 
 To intercept all Fetch Requests your class would implement
-{filter-api-javadoc}/io/kroxylicious/proxy/filter/FetchRequestFilter.html[FetchRequestFilter]:
+{api-javadoc}/io/kroxylicious/proxy/filter/FetchRequestFilter.html[FetchRequestFilter]:
 
 [source,java]
 ----
@@ -94,7 +93,7 @@ public class FetchRequestClientIdFilter implements FetchRequestFilter {
 ----
 
 To intercept all Fetch Responses your class would implement
-{filter-api-javadoc}/io/kroxylicious/proxy/filter/FetchResponseFilter.html[FetchResponseFilter]:
+{api-javadoc}/io/kroxylicious/proxy/filter/FetchResponseFilter.html[FetchResponseFilter]:
 
 [source,java]
 ----
@@ -113,8 +112,8 @@ public class FetchRequestClientIdFilter implements FetchResponseFilter {
 ----
 
 To intercept all Fetch Requests and all Fetch Responses your class would implement
-{filter-api-javadoc}/io/kroxylicious/proxy/filter/FetchRequestFilter.html[FetchRequestFilter] and
-{filter-api-javadoc}/io/kroxylicious/proxy/filter/FetchResponseFilter.html[FetchResponseFilter]:
+{api-javadoc}/io/kroxylicious/proxy/filter/FetchRequestFilter.html[FetchRequestFilter] and
+{api-javadoc}/io/kroxylicious/proxy/filter/FetchResponseFilter.html[FetchResponseFilter]:
 
 [source,java]
 ----
@@ -148,9 +147,9 @@ interfaces.
 A filter may wish to intercept every message being sent from the Client to the Cluster or from the Cluster
 to the Client. To do this your custom filter will implement:
 
-- {filter-api-javadoc}/io/kroxylicious/proxy/filter/RequestFilter.html[RequestFilter]
+- {api-javadoc}/io/kroxylicious/proxy/filter/RequestFilter.html[RequestFilter]
 to intercept all requests.
-- {filter-api-javadoc}/io/kroxylicious/proxy/filter/ResponseFilter.html[ResponseFilter]
+- {api-javadoc}/io/kroxylicious/proxy/filter/ResponseFilter.html[ResponseFilter]
 to intercept all responses.
 
 Custom filters are free to implement either interface or both interfaces to intercept all messages.
@@ -180,7 +179,7 @@ interfaces.
 === Composite Filters
 
 Sometimes we want to present a chain of multiple Filters as a single cohesive unit with just one entry in the `filters`
-configuration of Kroxylicious. The {github}/blob/main/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/CompositeFilter.java[CompositeFilter]
+configuration of Kroxylicious. The {github}/blob/main/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/CompositeFilter.java[CompositeFilter]
 interface enables you to do this.
 
 An example might look like this:
@@ -406,7 +405,7 @@ public class FetchFilter implements FetchRequestFilter{
 :github-sample-contributor: {github}/blob/main/kroxylicious-sample/src/main/java/io/kroxylicious/sample/SampleContributor.java
 
 For Kroxylicious to instantiate and configure your custom filter we use Java's {java-17-javadoc}/java.base/java/util/ServiceLoader.html[ServiceLoader] API.
-A Custom Filter JAR should provide a {filter-api-javadoc}/io/kroxylicious/proxy/filter/FilterContributor.html[FilterContributor] implementation that can contribute an instance of your custom Filter and
+A Custom Filter JAR should provide a {api-javadoc}/io/kroxylicious/proxy/filter/FilterContributor.html[FilterContributor] implementation that can contribute an instance of your custom Filter and
 optionally declare a configuration class that Kroxylicious will populate (using Jackson) when loading your custom Filter.
 The module must package a `META-INF/services/io.kroxylicious.proxy.filter.FilterContributor` file containing the
 classname of the contributor into the JAR file.
diff --git a/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml b/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml
index eaeee8cd..b71b2545 100644
--- a/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml
+++ b/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml
@@ -31,10 +31,6 @@
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-api</artifactId>
         </dependency>
-        <dependency>
-            <groupId>io.kroxylicious</groupId>
-            <artifactId>kroxylicious-filter-api</artifactId>
-        </dependency>
 
         <!-- project dependencies - test -->
         <dependency>
diff --git a/kroxylicious-additional-filters/kroxylicious-schema-validation/pom.xml b/kroxylicious-additional-filters/kroxylicious-schema-validation/pom.xml
index 38553cc5..81cfe955 100644
--- a/kroxylicious-additional-filters/kroxylicious-schema-validation/pom.xml
+++ b/kroxylicious-additional-filters/kroxylicious-schema-validation/pom.xml
@@ -31,10 +31,6 @@
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-api</artifactId>
         </dependency>
-        <dependency>
-            <groupId>io.kroxylicious</groupId>
-            <artifactId>kroxylicious-filter-api</artifactId>
-        </dependency>
 
         <!-- third party dependencies - runtime and compile -->
         <dependency>
diff --git a/kroxylicious-additional-filters/pom.xml b/kroxylicious-additional-filters/pom.xml
index dbf26ef9..ee712489 100644
--- a/kroxylicious-additional-filters/pom.xml
+++ b/kroxylicious-additional-filters/pom.xml
@@ -46,7 +46,6 @@
                                     </excludes>
                                     <includes>
                                         <include>io.kroxylicious:kroxylicious-api</include>
-                                        <include>io.kroxylicious:kroxylicious-filter-api</include>
                                         <include>io.kroxylicious:kroxylicious-unit-test</include>
                                     </includes>
                                 </bannedDependencies>
diff --git a/api/kroxylicious-api/kroxylicious-api.iml b/kroxylicious-api/kroxylicious-api.iml
similarity index 100%
rename from api/kroxylicious-api/kroxylicious-api.iml
rename to kroxylicious-api/kroxylicious-api.iml
diff --git a/api/kroxylicious-filter-api/pom.xml b/kroxylicious-api/pom.xml
similarity index 92%
rename from api/kroxylicious-filter-api/pom.xml
rename to kroxylicious-api/pom.xml
index d662a23a..9a47dea4 100644
--- a/api/kroxylicious-filter-api/pom.xml
+++ b/kroxylicious-api/pom.xml
@@ -15,10 +15,10 @@
         <groupId>io.kroxylicious</groupId>
         <artifactId>kroxylicious-parent</artifactId>
         <version>0.3.0-SNAPSHOT</version>
-        <relativePath>../../pom.xml</relativePath>
+        <relativePath>../pom.xml</relativePath>
     </parent>
 
-    <artifactId>kroxylicious-filter-api</artifactId>
+    <artifactId>kroxylicious-api</artifactId>
     <version>0.3.0-SNAPSHOT</version>
     <packaging>jar</packaging>
 
@@ -26,16 +26,10 @@
         <libs.dir>libs</libs.dir>
     </properties>
 
-    <name>Kroxylicious Filter API</name>
-    <description>The public interfaces, classes, etc required to author custom filters and configure Kroxylicious to use them</description>
+    <name>Kroxylicious API</name>
+    <description>generic APIs used throughout Kroxylicious</description>
 
     <dependencies>
-        <!-- project dependencies - runtime and compile -->
-        <dependency>
-            <groupId>io.kroxylicious</groupId>
-            <artifactId>kroxylicious-api</artifactId>
-        </dependency>
-
         <!-- third party dependencies - runtime and compile -->
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
@@ -66,8 +60,8 @@
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter-params</artifactId>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
             <scope>test</scope>
         </dependency>
     </dependencies>
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusternetworkaddressconfigprovider/ClusterNetworkAddressConfigProviderContributor.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusternetworkaddressconfigprovider/ClusterNetworkAddressConfigProviderContributor.java
similarity index 100%
rename from api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusternetworkaddressconfigprovider/ClusterNetworkAddressConfigProviderContributor.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusternetworkaddressconfigprovider/ClusterNetworkAddressConfigProviderContributor.java
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/config/BaseConfig.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/config/BaseConfig.java
similarity index 100%
rename from api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/config/BaseConfig.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/config/BaseConfig.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/CompositeFilter.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/CompositeFilter.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/CompositeFilter.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/CompositeFilter.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterAndInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterAndInvoker.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterAndInvoker.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterAndInvoker.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterContributor.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContributor.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterContributor.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContributor.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvoker.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvoker.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvoker.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvokers.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvokers.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvokers.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvokers.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/HandleNothingFilterInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/HandleNothingFilterInvoker.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/HandleNothingFilterInvoker.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/HandleNothingFilterInvoker.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilter.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilter.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilter.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilter.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
similarity index 93%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
index 5bd6c12a..18688b31 100644
--- a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
@@ -84,5 +84,11 @@ public interface KrpcFilterContext {
      */
     void forwardResponse(ApiMessage response);
 
+    /**
+     * Allows a filter to decide to close the connection. The client will be disconnected.  The client is free
+     * to retry the connection.  The client will receive no indication why the connection was closed.
+     */
+    void closeConnection();
+
     // TODO an API to allow a filter to add/remove another filter from the pipeline
 }
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilter.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilter.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilter.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilter.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterInvoker.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterInvoker.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilterInvoker.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/RequestResponseInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestResponseInvoker.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/RequestResponseInvoker.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/RequestResponseInvoker.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilter.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilter.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilter.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilter.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterInvoker.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterInvoker.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilterInvoker.java
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/SafeInvoker.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/SafeInvoker.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/SafeInvoker.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/SafeInvoker.java
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java
similarity index 100%
rename from api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterNetworkAddressConfigProvider.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterNetworkAddressConfigProvider.java
similarity index 58%
rename from api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterNetworkAddressConfigProvider.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterNetworkAddressConfigProvider.java
index 8246ef31..02de1323 100644
--- a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterNetworkAddressConfigProvider.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterNetworkAddressConfigProvider.java
@@ -5,6 +5,7 @@
  */
 package io.kroxylicious.proxy.service;
 
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
@@ -22,8 +23,8 @@ public interface ClusterNetworkAddressConfigProvider {
 
     /**
      * Address of broker with the given node id, includes the port. Note that
-     * {@code nodeId} are generally expected to be consecutively numbered and starting from zero. However, gaps in the sequence can potentially emerge as
-     * the target cluster topology evolves.
+     * {@code nodeId} are generally expected to be consecutively numbered and starting from zero. However,
+     * gaps in the sequence can potentially emerge as the target cluster topology evolves.
      *
      * @param nodeId node identifier
      * @return broker address
@@ -31,6 +32,21 @@ public interface ClusterNetworkAddressConfigProvider {
      */
     HostPort getBrokerAddress(int nodeId) throws IllegalArgumentException;
 
+    /**
+     * Generates the node id implied by the given broker address. This method make sense only for
+     * implementation that embed node id information into the broker address.  This information is
+     * used at startup time to allow a client that already in possession of a broker address
+     * to reconnect to the cluster via Kroxylicious using only that address.
+     * <br/>
+     * This is an optional method. An implementation can return null.
+     *
+     * @param brokerAddress broker address
+     * @return a broker id or null if the broker id cannot be
+     */
+    default Integer getBrokerIdFromBrokerAddress(HostPort brokerAddress) {
+        return null;
+    }
+
     /**
      * Gets the bind address used when binding socket.  Used to restrict
      * listening to particular network interfaces.
@@ -67,4 +83,18 @@ public interface ClusterNetworkAddressConfigProvider {
         return Set.of();
     }
 
+    /**
+     * Map of discovery addresses that will be bound by the virtual cluster on start-up
+     * (before the first reconcile has taken place).  Discovery bindings always
+     * point to a target cluster's bootstrap and are used for purposes of metadata
+     * discovery only.
+     * <br/>
+     * Discovery bindings are replaced by bindings to upstream brokers on reconciliation.
+     *
+     * @return discovery address map
+     */
+    default Map<Integer, HostPort> discoveryAddressMap() {
+        return Map.of();
+    }
+
 }
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/Contributor.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/Contributor.java
similarity index 100%
rename from api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/Contributor.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/Contributor.java
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/HostPort.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/HostPort.java
similarity index 100%
rename from api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/HostPort.java
rename to kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/HostPort.java
diff --git a/api/kroxylicious-filter-api/src/main/templates/Kproxy/Filter.ftl b/kroxylicious-api/src/main/templates/Kproxy/Filter.ftl
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/templates/Kproxy/Filter.ftl
rename to kroxylicious-api/src/main/templates/Kproxy/Filter.ftl
diff --git a/api/kroxylicious-filter-api/src/main/templates/Kproxy/FilterInvoker.ftl b/kroxylicious-api/src/main/templates/Kproxy/FilterInvoker.ftl
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/templates/Kproxy/FilterInvoker.ftl
rename to kroxylicious-api/src/main/templates/Kproxy/FilterInvoker.ftl
diff --git a/api/kroxylicious-filter-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl b/kroxylicious-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl
similarity index 100%
rename from api/kroxylicious-filter-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl
rename to kroxylicious-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl
diff --git a/api/kroxylicious-filter-api/src/test/java/io/kroxylicious/proxy/filter/FilterInvokersTest.java b/kroxylicious-api/src/test/java/io/kroxylicious/proxy/filter/FilterInvokersTest.java
similarity index 100%
rename from api/kroxylicious-filter-api/src/test/java/io/kroxylicious/proxy/filter/FilterInvokersTest.java
rename to kroxylicious-api/src/test/java/io/kroxylicious/proxy/filter/FilterInvokersTest.java
diff --git a/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java b/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java
similarity index 100%
rename from api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java
rename to kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java
diff --git a/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/HostPortTest.java b/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/HostPortTest.java
similarity index 100%
rename from api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/HostPortTest.java
rename to kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/HostPortTest.java
diff --git a/kroxylicious-sample/pom.xml b/kroxylicious-sample/pom.xml
index f2c797d0..23aac130 100644
--- a/kroxylicious-sample/pom.xml
+++ b/kroxylicious-sample/pom.xml
@@ -31,11 +31,6 @@
             <artifactId>kroxylicious-api</artifactId>
             <scope>provided</scope>
         </dependency>
-        <dependency>
-            <groupId>io.kroxylicious</groupId>
-            <artifactId>kroxylicious-filter-api</artifactId>
-            <scope>provided</scope>
-        </dependency>
 
         <!-- project dependencies - test -->
         <dependency>
diff --git a/kroxylicious/pom.xml b/kroxylicious/pom.xml
index 948a1857..2c3596be 100644
--- a/kroxylicious/pom.xml
+++ b/kroxylicious/pom.xml
@@ -33,10 +33,6 @@
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-api</artifactId>
         </dependency>
-        <dependency>
-            <groupId>io.kroxylicious</groupId>
-            <artifactId>kroxylicious-filter-api</artifactId>
-        </dependency>
 
         <!-- project dependencies - test -->
         <dependency>
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
index dd496a2e..3fef507b 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
@@ -209,6 +209,13 @@ class DefaultFilterContext implements KrpcFilterContext {
         }
     }
 
+    @Override
+    public void closeConnection() {
+        this.channelContext.close().addListener(future -> {
+            LOGGER.debug("{} closed.", channelDescriptor());
+        });
+    }
+
     /**
      * In this case we are not forwarding to the proxied broker but responding immediately.
      * We want to check that the ApiMessage is the correct type for the request. Ie if the
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
index 40eaac58..7956a47d 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
@@ -33,6 +33,7 @@ import io.kroxylicious.proxy.internal.codec.KafkaRequestDecoder;
 import io.kroxylicious.proxy.internal.codec.KafkaResponseEncoder;
 import io.kroxylicious.proxy.internal.filter.ApiVersionsFilter;
 import io.kroxylicious.proxy.internal.filter.BrokerAddressFilter;
+import io.kroxylicious.proxy.internal.filter.EagerMetadataLearner;
 import io.kroxylicious.proxy.internal.net.Endpoint;
 import io.kroxylicious.proxy.internal.net.EndpointReconciler;
 import io.kroxylicious.proxy.internal.net.VirtualClusterBinding;
@@ -176,6 +177,9 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
             List<FilterAndInvoker> brokerAddressFilters = FilterAndInvoker.build(new BrokerAddressFilter(virtualCluster, endpointReconciler));
             var filters = new ArrayList<>(apiVersionFilters);
             filters.addAll(customProtocolFilters);
+            if (binding.restrictUpstreamToMetadataDiscovery()) {
+                filters.addAll(FilterAndInvoker.build(new EagerMetadataLearner()));
+            }
             filters.addAll(brokerAddressFilters);
 
             var target = binding.upstreamTarget();
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
index ebdfa482..eabd31fd 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/BrokerAddressPatternUtils.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/BrokerAddressPatternUtils.java
@@ -6,11 +6,13 @@
 
 package io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider;
 
+import java.util.Arrays;
 import java.util.Objects;
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.function.Consumer;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 final class BrokerAddressPatternUtils {
     private static final String LITERAL_NODE_ID = "$(nodeId)";
@@ -70,4 +72,9 @@ final class BrokerAddressPatternUtils {
     static String replaceLiteralNodeId(String brokerAddressPattern, int nodeId) {
         return LITERAL_NODE_ID_PATTERN.matcher(brokerAddressPattern).replaceAll(Integer.toString(nodeId));
     }
+
+    static Pattern createNodeIdCapturingRegex(String brokerAddressPattern) {
+        var partsQuoted = Arrays.stream(brokerAddressPattern.split(Pattern.quote(LITERAL_NODE_ID))).map(Pattern::quote);
+        return Pattern.compile(partsQuoted.collect(Collectors.joining("(\\d+)")), Pattern.CASE_INSENSITIVE);
+    }
 }
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..73a82301 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
@@ -8,8 +8,10 @@ package io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider;
 
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -44,6 +46,7 @@ public class PortPerBrokerClusterNetworkAddressConfigProvider implements Cluster
     private final int brokerStartPort;
     private final Set<Integer> exclusivePorts;
     private final int brokerEndPortExclusive;
+    private final int numberOfBrokerPorts;
 
     /**
      * Creates the provider.
@@ -54,7 +57,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));
@@ -90,6 +93,11 @@ public class PortPerBrokerClusterNetworkAddressConfigProvider implements Cluster
         return this.exclusivePorts;
     }
 
+    @Override
+    public Map<Integer, HostPort> discoveryAddressMap() {
+        return IntStream.range(0, numberOfBrokerPorts).boxed().collect(Collectors.toMap(Function.identity(), this::getBrokerAddress));
+    }
+
     /**
      * Creates the configuration for this provider.
      */
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..48e312c0 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,9 @@
 package io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider;
 
 import java.util.Set;
+import java.util.regex.Pattern;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 import io.kroxylicious.proxy.config.BaseConfig;
@@ -34,6 +36,7 @@ public class SniRoutingClusterNetworkAddressConfigProvider implements ClusterNet
 
     private final HostPort bootstrapAddress;
     private final String brokerAddressPattern;
+    private final Pattern brokerAddressNodeIdCapturingRegex;
 
     /**
      * Creates the provider.
@@ -43,6 +46,7 @@ public class SniRoutingClusterNetworkAddressConfigProvider implements ClusterNet
     public SniRoutingClusterNetworkAddressConfigProvider(SniRoutingClusterNetworkAddressConfigProviderConfig config) {
         this.bootstrapAddress = config.bootstrapAddress;
         this.brokerAddressPattern = config.brokerAddressPattern;
+        this.brokerAddressNodeIdCapturingRegex = config.brokerAddressNodeIdCapturingRegex;
     }
 
     @Override
@@ -60,6 +64,24 @@ public class SniRoutingClusterNetworkAddressConfigProvider implements ClusterNet
         return new HostPort(BrokerAddressPatternUtils.replaceLiteralNodeId(brokerAddressPattern, nodeId), bootstrapAddress.port());
     }
 
+    @Override
+    public Integer getBrokerIdFromBrokerAddress(HostPort brokerAddress) {
+        if (brokerAddress.port() != bootstrapAddress.port()) {
+            return null;
+        }
+        var matcher = brokerAddressNodeIdCapturingRegex.matcher(brokerAddress.host());
+        if (matcher.matches()) {
+            var nodeId = matcher.group(1);
+            try {
+                return Integer.valueOf(nodeId);
+            }
+            catch (NumberFormatException e) {
+                throw new IllegalStateException("unexpected exception parsing : '%s'".formatted(nodeId), e);
+            }
+        }
+        return null;
+    }
+
     @Override
     public Set<Integer> getSharedPorts() {
         return Set.of(bootstrapAddress.port());
@@ -78,6 +100,8 @@ public class SniRoutingClusterNetworkAddressConfigProvider implements ClusterNet
         private final HostPort bootstrapAddress;
 
         private final String brokerAddressPattern;
+        @JsonIgnore
+        private final Pattern brokerAddressNodeIdCapturingRegex;
 
         public SniRoutingClusterNetworkAddressConfigProviderConfig(@JsonProperty(required = true) HostPort bootstrapAddress,
                                                                    @JsonProperty(required = true) String brokerAddressPattern) {
@@ -102,6 +126,8 @@ public class SniRoutingClusterNetworkAddressConfigProvider implements ClusterNet
 
             this.bootstrapAddress = bootstrapAddress;
             this.brokerAddressPattern = brokerAddressPattern;
+            this.brokerAddressNodeIdCapturingRegex = BrokerAddressPatternUtils.createNodeIdCapturingRegex(brokerAddressPattern);
+
         }
 
         public HostPort getBootstrapAddress() {
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearner.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearner.java
new file mode 100644
index 00000000..1d44a150
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearner.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.proxy.internal.filter;
+
+import java.util.Set;
+
+import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+import io.kroxylicious.proxy.filter.RequestFilter;
+
+/**
+ * An internal filter that causes the system to eagerly learn the cluster's topology by spontaneously emitting
+ * an out-of-band Metadata request at the earliest legal point in the Kafka conversation.  The response to allows
+ * the Endpoint reconciliation to take place so that restricted upstream bindings are replaced by true bindings to
+ * the actual upstream brokers.
+ * <br/>
+ * Once the bindings are made, the filter causes the client's connection to close.   This is done
+ * in order to force the client to reconnect, thus ensuring the client has a connection to the intended broker.
+ *
+ * @see io.kroxylicious.proxy.internal.net.EndpointRegistry
+ */
+public class EagerMetadataLearner implements RequestFilter {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(EagerMetadataLearner.class);
+
+    /**
+     * The set of the API keys that are permitted before the client would normally send a METADATA request.
+     */
+    private final static Set<ApiKeys> KAFKA_PRELUDE = Set.of(ApiKeys.API_VERSIONS, ApiKeys.SASL_HANDSHAKE, ApiKeys.SASL_AUTHENTICATE);
+
+    public EagerMetadataLearner() {
+    }
+
+    @Override
+    public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        if (KAFKA_PRELUDE.contains(apiKey)) {
+            filterContext.forwardRequest(header, body);
+        }
+        else {
+            final short apiVersion = determineMetadataApiVersion(header);
+            // Send an out-of-band Metadata request. The response will be intercepted by the in-built BrokerAddressFilter.
+            // By the time control returns to the handler, the upstream addresses will have been reconciled.
+            var unused = filterContext.<MetadataResponseData> sendRequest(apiVersion, new MetadataRequestData())
+                    .thenAccept(metadataResponseData -> {
+                        if (apiKey.equals(ApiKeys.METADATA) && apiVersion == header.requestApiVersion()) {
+                            // The client's requested matched our out-of-band request, so we may as well return the
+                            // response.
+                            filterContext.forwardResponse(metadataResponseData);
+                        }
+                        // closing the connection is important. This client connection is connected to bootstrap (it could
+                        // be any broker or maybe not something else). we must close the connection to force the client to
+                        // connect again.
+                        filterContext.closeConnection();
+                        LOGGER.info("Closing upstream bootstrap connection {} now that endpoint reconciliation is complete.", filterContext.channelDescriptor());
+                    });
+        }
+    }
+
+    private short determineMetadataApiVersion(RequestHeaderData header) {
+        final short apiVersion;
+        if (header.requestApiKey() == ApiKeys.METADATA.id) {
+            apiVersion = header.requestApiVersion();
+        }
+        else {
+            // TODO: use a version appearing the intersection calculated by ApiVersionFilter.
+            apiVersion = MetadataRequestData.LOWEST_SUPPORTED_VERSION;
+        }
+        return apiVersion;
+    }
+
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/EndpointRegistry.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/EndpointRegistry.java
index 9bb579fc..33d4405f 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/EndpointRegistry.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/EndpointRegistry.java
@@ -6,9 +6,14 @@
 
 package io.kroxylicious.proxy.internal.net;
 
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentHashMap;
@@ -117,7 +122,8 @@ public class EndpointRegistry implements EndpointReconciler, VirtualClusterBindi
 
     }
 
-    private record VirtualClusterRecord(CompletionStage<Endpoint> registrationStage, AtomicReference<ReconciliationRecord> reconciliationRecord, AtomicReference<CompletionStage<Void>> deregistrationStage) {
+    private record VirtualClusterRecord(CompletionStage<Endpoint> registrationStage, AtomicReference<ReconciliationRecord> reconciliationRecord,
+                                        AtomicReference<CompletionStage<Void>> deregistrationStage) {
         private VirtualClusterRecord {
             Objects.requireNonNull(registrationStage);
             Objects.requireNonNull(reconciliationRecord);
@@ -180,30 +186,43 @@ public class EndpointRegistry implements EndpointReconciler, VirtualClusterBindi
 
         vcr.reconciliationRecord().set(ReconciliationRecord.createEmptyReconcileRecord());
 
-        var unused = bootstrapEndpointFuture.whenComplete((u, t) -> {
-            var future = vcr.registrationStage.toCompletableFuture();
-            if (t != null) {
-                // Try to roll back any bindings that were successfully made
-                deregisterBinding(virtualCluster, (vcb) -> vcb.virtualCluster().equals(virtualCluster))
-                        .handle((u1, t1) -> {
-                            if (t1 != null) {
-                                LOGGER.warn("Registration error", t);
-                                LOGGER.warn("Secondary error occurred whilst handling a previous registration error: {}", t.getMessage(), t1);
-                            }
-                            registeredVirtualClusters.remove(virtualCluster);
-                            future.completeExceptionally(t);
-                            return null;
-                        });
-            }
-            else {
-                try {
-                    future.complete(bootstrapEndpointFuture.get());
-                }
-                catch (Throwable t1) {
-                    future.completeExceptionally(t1);
-                }
-            }
-        });
+        // bind any discovery binding to the bootstrap address
+        var discoveryAddressesMapStage = allOfStage(Optional.ofNullable(virtualCluster.discoveryAddressMap()).orElse(Map.of())
+                .entrySet()
+                .stream()
+                .sorted(Map.Entry.comparingByKey()) // ordering not functionality important, but simplifies the unit testing
+                .map(e -> {
+                    var nodeId = e.getKey();
+                    var bhp = e.getValue();
+                    return registerBinding(new Endpoint(virtualCluster.getBindAddress(), bhp.port(), virtualCluster.isUseTls()), bhp.host(),
+                            new VirtualClusterBrokerBinding(virtualCluster, upstreamBootstrap, nodeId, true));
+                }));
+
+        var unused = bootstrapEndpointFuture.thenCombine(discoveryAddressesMapStage, (bef, bps) -> bef)
+                .whenComplete((u, t) -> {
+                    var future = vcr.registrationStage.toCompletableFuture();
+                    if (t != null) {
+                        // Try to roll back any bindings that were successfully made
+                        deregisterBinding(virtualCluster, (vcb) -> vcb.virtualCluster().equals(virtualCluster))
+                                .handle((u1, t1) -> {
+                                    if (t1 != null) {
+                                        LOGGER.warn("Registration error", t);
+                                        LOGGER.warn("Secondary error occurred whilst handling a previous registration error: {}", t.getMessage(), t1);
+                                    }
+                                    registeredVirtualClusters.remove(virtualCluster);
+                                    future.completeExceptionally(t);
+                                    return null;
+                                });
+                    }
+                    else {
+                        try {
+                            future.complete(bootstrapEndpointFuture.get());
+                        }
+                        catch (Throwable t1) {
+                            future.completeExceptionally(t1);
+                        }
+                    }
+                });
 
         return vcr.registrationStage();
     }
@@ -314,8 +333,10 @@ public class EndpointRegistry implements EndpointReconciler, VirtualClusterBindi
     private void doReconcile(VirtualCluster virtualCluster, Map<Integer, HostPort> upstreamNodes, CompletableFuture<Void> future, VirtualClusterRecord vcr) {
         var bindingAddress = virtualCluster.getBindAddress();
 
-        var creations = upstreamNodes.entrySet().stream().map(e -> new VirtualClusterBrokerBinding(virtualCluster, e.getValue(), e.getKey()))
-                .collect(Collectors.toCollection(ConcurrentHashMap::newKeySet));
+        var discoveryBrokerIds = Optional.ofNullable(virtualCluster.discoveryAddressMap()).map(Map::keySet).orElse(Set.of());
+        var allBrokerIds = Stream.concat(discoveryBrokerIds.stream(), upstreamNodes.keySet().stream()).collect(Collectors.toUnmodifiableSet());
+
+        var creations = constructPossibleBindingsToCreate(virtualCluster, upstreamNodes);
 
         // first assemble the stream of de-registrations (and by side effect: update creations)
         var deregs = allOfStage(listeningChannels.values().stream()
@@ -334,7 +355,7 @@ public class EndpointRegistry implements EndpointReconciler, VirtualClusterBindi
                                     .filter(VirtualClusterBrokerBinding.class::isInstance)
                                     .map(VirtualClusterBrokerBinding.class::cast)
                                     .peek(creations::remove) // side effect
-                                    .filter(vcbb -> !upstreamNodes.containsKey(vcbb.nodeId()))
+                                    .filter(vcbb -> !allBrokerIds.contains(vcbb.nodeId()))
                                     .map(vcbb -> deregisterBinding(virtualCluster, vcbb::equals)));
                         }))));
 
@@ -358,6 +379,24 @@ public class EndpointRegistry implements EndpointReconciler, VirtualClusterBindi
                 });
     }
 
+    private Set<VirtualClusterBrokerBinding> constructPossibleBindingsToCreate(VirtualCluster virtualCluster,
+                                                                               Map<Integer, HostPort> upstreamNodes) {
+        var upstreamBootstrap = virtualCluster.targetCluster().bootstrapServersList().get(0);
+        var discoveryBrokerIds = Optional.ofNullable(virtualCluster.discoveryAddressMap()).orElse(Map.of());
+        // create possible set of bindings to create
+        var creations = upstreamNodes.entrySet()
+                .stream()
+                .map(e -> new VirtualClusterBrokerBinding(virtualCluster, e.getValue(), e.getKey(), false))
+                .collect(Collectors.toCollection(ConcurrentHashMap::newKeySet));
+        // add bindings corresponding to any pre-bindings. There are marked as restricted and point to bootstrap.
+        creations.addAll(discoveryBrokerIds
+                .keySet()
+                .stream()
+                .filter(Predicate.not(upstreamNodes::containsKey))
+                .map(nodeId -> new VirtualClusterBrokerBinding(virtualCluster, upstreamBootstrap, nodeId, true)).toList());
+        return creations;
+    }
+
     /* test */ boolean isRegistered(VirtualCluster virtualCluster) {
         return registeredVirtualClusters.containsKey(virtualCluster);
     }
@@ -470,25 +509,59 @@ public class EndpointRegistry implements EndpointReconciler, VirtualClusterBindi
     public CompletionStage<VirtualClusterBinding> resolve(Endpoint endpoint, String sniHostname) {
         var lcr = this.listeningChannels.get(endpoint);
         if (lcr == null || lcr.unbindingStage().get() != null) {
-            return CompletableFuture.failedStage(buildEndpointResolutionException("Failed to find channel matching ", endpoint, sniHostname));
+            return CompletableFuture.failedStage(buildEndpointResolutionException("Failed to find channel matching", endpoint, sniHostname));
         }
+
         return lcr.bindingStage().thenApply(acceptorChannel -> {
             var bindings = acceptorChannel.attr(CHANNEL_BINDINGS);
             if (bindings == null || bindings.get() == null) {
-                throw buildEndpointResolutionException("No channel bindings found for ", endpoint, sniHostname);
+                throw buildEndpointResolutionException("No channel bindings found for", endpoint, sniHostname);
             }
             // We first look for a binding matching by SNI name, then fallback to a null match.
             var binding = bindings.get().getOrDefault(RoutingKey.createBindingKey(sniHostname), bindings.get().get(RoutingKey.NULL_ROUTING_KEY));
             if (binding == null) {
-                throw buildEndpointResolutionException("No channel bindings found for ", endpoint, sniHostname);
+                // If there is an SNI name that matches against the virtual cluster broker address pattern, we generate
+                // a restricted broker binding that points at the virtual cluster's bootstrap.
+                if (sniHostname != null) {
+                    var allBindingsForPort = bindings.get().values();
+                    var brokerAddress = new HostPort(sniHostname, endpoint.port());
+                    var allBootstrapBindings = getAllBootstrapBindings(allBindingsForPort);
+                    var bootstrapToBrokerId = allBootstrapBindings.stream()
+                            .collect(HashMap<VirtualClusterBootstrapBinding, Integer>::new, (m, b) -> {
+                                var nodeId = b.virtualCluster().getBrokerIdFromBrokerAddress(brokerAddress);
+                                if (nodeId != null) {
+                                    m.put(b, nodeId);
+                                }
+                            }, HashMap::putAll);
+                    var size = bootstrapToBrokerId.size();
+                    if (size > 1) {
+                        throw new EndpointResolutionException("Failed to generate an unbound broker binding from SNI " +
+                                "as it matches the broker address pattern of more than one virtual cluster",
+                                buildEndpointResolutionException("No channel bindings found for", endpoint, sniHostname));
+                    }
+                    else if (size == 1) {
+                        var e = bootstrapToBrokerId.entrySet().iterator().next();
+                        var bootstrapBinding = e.getKey();
+                        var nodeId = e.getValue();
+                        return new VirtualClusterBrokerBinding(bootstrapBinding.virtualCluster(), bootstrapBinding.upstreamTarget(), nodeId, true);
+                    }
+                }
+                throw buildEndpointResolutionException("No channel bindings found for", endpoint, sniHostname);
             }
             return binding;
         });
     }
 
+    private List<VirtualClusterBootstrapBinding> getAllBootstrapBindings(Collection<VirtualClusterBinding> allBindingsForPort) {
+        return allBindingsForPort.stream()
+                .filter(VirtualClusterBootstrapBinding.class::isInstance)
+                .map(VirtualClusterBootstrapBinding.class::cast)
+                .toList();
+    }
+
     private EndpointResolutionException buildEndpointResolutionException(String prefix, Endpoint endpoint, String sniHostname) {
         return new EndpointResolutionException(
-                ("%s binding address: %s, port %d, sniHostname: %s, tls %s").formatted(prefix,
+                ("%s binding address: %s, port: %d, sniHostname: %s, tls: %b").formatted(prefix,
                         endpoint.bindingAddress().orElse("<any>"),
                         endpoint.port(),
                         sniHostname == null ? "<none>" : sniHostname,
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/VirtualClusterBinding.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/VirtualClusterBinding.java
index 077a3e78..5a64beb4 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/VirtualClusterBinding.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/VirtualClusterBinding.java
@@ -26,4 +26,13 @@ public interface VirtualClusterBinding {
      * @return upstream target.
      */
     HostPort upstreamTarget();
+
+    /**
+     * If set true, the upstream target must only be used for metadata discovery.
+     *
+     * @return true if the target must be restricted to metadata discovery.
+     */
+    default boolean restrictUpstreamToMetadataDiscovery() {
+        return false;
+    }
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/VirtualClusterBootstrapBinding.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/VirtualClusterBootstrapBinding.java
index 7b909213..b3eb5bcf 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/VirtualClusterBootstrapBinding.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/VirtualClusterBootstrapBinding.java
@@ -13,6 +13,9 @@ import io.kroxylicious.proxy.service.HostPort;
 
 /**
  * A binding to a virtual cluster bootstrap.
+ *
+ * @param virtualCluster the virtual cluster
+ * @param upstreamTarget the upstream bootstrap target
  */
 public record VirtualClusterBootstrapBinding(VirtualCluster virtualCluster, HostPort upstreamTarget) implements VirtualClusterBinding {
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/VirtualClusterBrokerBinding.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/VirtualClusterBrokerBinding.java
index 04f5423f..50dd4ca7 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/VirtualClusterBrokerBinding.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/net/VirtualClusterBrokerBinding.java
@@ -13,8 +13,14 @@ import io.kroxylicious.proxy.service.HostPort;
 
 /**
  * A broker specific virtual cluster binding.
+ *
+ * @param virtualCluster                       the virtual cluster
+ * @param upstreamTarget                       the upstream target of this binding
+ * @param nodeId                               kafka nodeId of the target broker
+ * @param restrictUpstreamToMetadataDiscovery  true if the upstreamTarget corresponds to a broker, false if it points at a bootstrap.
  */
-public record VirtualClusterBrokerBinding(VirtualCluster virtualCluster, HostPort upstreamTarget, int nodeId) implements VirtualClusterBinding {
+public record VirtualClusterBrokerBinding(VirtualCluster virtualCluster, HostPort upstreamTarget, int nodeId, boolean restrictUpstreamToMetadataDiscovery)
+        implements VirtualClusterBinding {
     public VirtualClusterBrokerBinding {
         Objects.requireNonNull(virtualCluster, "virtualCluster must not be null");
         Objects.requireNonNull(upstreamTarget, "upstreamTarget must not be null");
@@ -25,10 +31,10 @@ public record VirtualClusterBrokerBinding(VirtualCluster virtualCluster, HostPor
         return "VirtualClusterBrokerBinding[" +
                 "virtualCluster=" + this.virtualCluster() + ", " +
                 "upstreamTarget=" + this.upstreamTarget() + ", " +
+                "restrictUpstreamToMetadataDiscovery=" + this.restrictUpstreamToMetadataDiscovery() + ", " +
                 "nodeId=" + nodeId + ']';
     }
 
-
     public boolean refersToSameVirtualClusterAndNode(VirtualClusterBrokerBinding other) {
         return other != null && other.nodeId == this.nodeId && Objects.equals(other.virtualCluster, this.virtualCluster);
     }
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 9b07665c..fdfecdfe 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/model/VirtualCluster.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/model/VirtualCluster.java
@@ -6,6 +6,7 @@
 package io.kroxylicious.proxy.model;
 
 import java.io.UncheckedIOException;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -144,6 +145,16 @@ public class VirtualCluster implements ClusterNetworkAddressConfigProvider {
         return clusterNetworkAddressConfigProvider.getSharedPorts();
     }
 
+    @Override
+    public Map<Integer, HostPort> discoveryAddressMap() {
+        return clusterNetworkAddressConfigProvider.discoveryAddressMap();
+    }
+
+    @Override
+    public Integer getBrokerIdFromBrokerAddress(HostPort brokerAddress) {
+        return clusterNetworkAddressConfigProvider.getBrokerIdFromBrokerAddress(brokerAddress);
+    }
+
     public Optional<SslContext> getDownstreamSslContext() {
         return downstreamSslContext;
     }
@@ -151,4 +162,5 @@ public class VirtualCluster implements ClusterNetworkAddressConfigProvider {
     public Optional<SslContext> getUpstreamSslContext() {
         return upstreamSslContext;
     }
+
 }
diff --git a/pom.xml b/pom.xml
index 560f14eb..dbf25086 100644
--- a/pom.xml
+++ b/pom.xml
@@ -119,11 +119,6 @@
                 <artifactId>kroxylicious-api</artifactId>
                 <version>${kroxyliciousApi.version}</version>
             </dependency>
-            <dependency>
-                <groupId>io.kroxylicious</groupId>
-                <artifactId>kroxylicious-filter-api</artifactId>
-                <version>${kroxyliciousApi.version}</version>
-            </dependency>
             <dependency>
                 <groupId>io.kroxylicious</groupId>
                 <artifactId>kroxylicious-multitenant</artifactId>
@@ -272,8 +267,7 @@
     </dependencyManagement>
 
     <modules>
-        <module>api/kroxylicious-api</module>
-        <module>api/kroxylicious-filter-api</module>
+        <module>kroxylicious-api</module>
         <module>krpc-code-gen</module>
         <module>kroxylicious-test-tools</module>
         <module>kroxylicious-unit-test</module>
diff --git a/scripts/release-api.sh b/scripts/release-api.sh
index 1565d005..b341cc8a 100755
--- a/scripts/release-api.sh
+++ b/scripts/release-api.sh
@@ -8,7 +8,7 @@
 set -o nounset -e
 
 RELEASE_API_VERSION=${1}
-API_MODULES=':kroxylicious-api,:kroxylicious-filter-api'
+API_MODULES=':kroxylicious-api'
 
 #if [[ -z "${GPG_KEY}" ]]; then
 #    echo "GPG_KEY not set unable to sign the release. Please export GPG_KEY" 1>&2
@@ -21,8 +21,7 @@ if [[ -z ${RELEASE_API_VERSION} ]]; then
 fi
 
 echo "Setting API version to ${RELEASE_API_VERSION}"
-mvn -q versions:set -DnewVersion="${RELEASE_API_VERSION}" -DprocessAllModules=true -pl ":kroxylicious-api" -DgenerateBackupPoms=false
-mvn -q versions:set -DnewVersion="${RELEASE_API_VERSION}" -DprocessAllModules=true -pl ":kroxylicious-filter-api" -DgenerateBackupPoms=false
+mvn -q versions:set -DnewVersion="${RELEASE_API_VERSION}" -DprocessAllModules=true -pl ${API_MODULES} -DgenerateBackupPoms=false
 mvn -q clean install -Pquick -pl ${API_MODULES}  #quick sanity check to ensure the API modules still build
 mvn -q versions:set-property -Dproperty=kroxyliciousApi.version -DnewVersion="${RELEASE_API_VERSION}" -DgenerateBackupPoms=false
 


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 f4e077aa..1a935be8 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/ExpositionIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/ExpositionIT.java
@@ -8,6 +8,8 @@ package io.kroxylicious.proxy;
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -15,6 +17,7 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.Admin;
@@ -24,11 +27,19 @@ import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.security.plain.PlainLoginModule;
+import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import io.kroxylicious.net.IntegrationTestInetAddressResolverProvider;
 import io.kroxylicious.proxy.config.ClusterNetworkAddressConfigProviderDefinitionBuilder;
@@ -40,6 +51,7 @@ import io.kroxylicious.testing.kafka.api.KafkaCluster;
 import io.kroxylicious.testing.kafka.clients.CloseableAdmin;
 import io.kroxylicious.testing.kafka.common.BrokerCluster;
 import io.kroxylicious.testing.kafka.common.KeytoolCertificateGenerator;
+import io.kroxylicious.testing.kafka.common.SaslPlainAuth;
 import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
 
 import static io.kroxylicious.test.tester.KroxyliciousTesters.kroxyliciousTester;
@@ -50,57 +62,38 @@ import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
 /**
  * Integration tests that focus on the ability to present virtual clusters, with various numbers of brokers)
  * to the kafka clients
- *
+ * <br/>
  * TODO corner case test - verify kroxy's ability to recover for a temporary port already bound condition.
  */
 @ExtendWith(KafkaClusterExtension.class)
 public class ExpositionIT {
+    private static final Logger LOGGER = LoggerFactory.getLogger(ExpositionIT.class);
 
     private static final String TOPIC = "my-test-topic";
     public static final HostPort PROXY_ADDRESS = HostPort.parse("localhost:9192");
 
-    @TempDir
-    private Path certsDirectory;
-
-    private record ClientTrust(Path trustStore, String password) {
-    }
-
-    @Test
-    public void exposesSingleUpstreamClusterOverTls(KafkaCluster cluster) throws Exception {
-
-        var brokerCertificateGenerator = new KeytoolCertificateGenerator();
-        brokerCertificateGenerator.generateSelfSignedCertificateEntry("test@redhat.com", "localhost", "KI", "RedHat", null, null, "US");
-        var clientTrustStore = certsDirectory.resolve("kafka.truststore.jks");
-        brokerCertificateGenerator.generateTrustStore(brokerCertificateGenerator.getCertFilePath(), "client",
-                clientTrustStore.toAbsolutePath().toString());
+    private static final String SNI_BASE_ADDRESS = IntegrationTestInetAddressResolverProvider.generateFullyQualifiedDomainName("sni");
 
-        String bootstrapServers = cluster.getBootstrapServers();
+    public static final HostPort SNI_BOOTSTRAP = new HostPort("bootstrap." + SNI_BASE_ADDRESS, 9192);
+    public static final String SNI_BROKER_ADDRESS_PATTERN = "broker-$(nodeId)." + SNI_BASE_ADDRESS;
+    public static final String SASL_USER = "user";
+    public static final String SASL_PASSWORD = "password";
 
+    @TempDir
+    private static Path certsDirectory;
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("virtualClusterConfigurations")
+    public 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", new VirtualClusterBuilder()
-                        .withNewTargetCluster()
-                        .withBootstrapServers(bootstrapServers)
-                        .endTargetCluster()
-                        .withClusterNetworkAddressConfigProvider(
-                                new ClusterNetworkAddressConfigProviderDefinitionBuilder("PortPerBroker").withConfig("bootstrapAddress", PROXY_ADDRESS)
-                                        .build())
-                        .build());
-
-        var demo = builder.getVirtualClusters().get("demo");
-        demo = new VirtualClusterBuilder(demo)
-                .withNewTls()
-                .withNewKeyStoreKey()
-                .withStoreFile(brokerCertificateGenerator.getKeyStoreLocation())
-                .withNewInlinePasswordStoreProvider(brokerCertificateGenerator.getPassword())
-                .endKeyStoreKey()
-                .endTls()
-                .build();
-        builder.addToVirtualClusters("demo", demo);
+                .addToVirtualClusters("demo", virtualClusterBuilder.build());
 
         try (var tester = kroxyliciousTester(builder);
-                var admin = tester.admin("demo", Map.of(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name,
-                        SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, clientTrustStore.toAbsolutePath().toString(),
-                        SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, brokerCertificateGenerator.getPassword()))) {
+                var admin = tester.admin("demo", clientSecurityProtocolConfig)) {
             // do some work to ensure connection is opened
             createTopic(admin, TOPIC, 1);
 
@@ -147,7 +140,7 @@ public class ExpositionIT {
 
     @Test
     public void exposesTwoSeparateUpstreamClustersUsingSniRouting(KafkaCluster cluster) throws Exception {
-        var clientTrust = new ArrayList<ClientTrust>();
+        var keystoreTrustStoreList = new ArrayList<KeystoreTrustStorePair>();
         var virtualClusterCommonNamePattern = IntegrationTestInetAddressResolverProvider.generateFullyQualifiedDomainName(".virtualcluster%d");
         var virtualClusterBootstrapPattern = "bootstrap" + virtualClusterCommonNamePattern;
         var virtualClusterBrokerAddressPattern = "broker-$(nodeId)" + virtualClusterCommonNamePattern;
@@ -162,14 +155,9 @@ public class ExpositionIT {
 
         int numberOfVirtualClusters = 2;
         for (int i = 0; i < numberOfVirtualClusters; i++) {
-
             var virtualClusterFQDN = virtualClusterBootstrapPattern.formatted(i);
-            var brokerCertificateGenerator = new KeytoolCertificateGenerator();
-            brokerCertificateGenerator.generateSelfSignedCertificateEntry("test@redhat.com", "*" + virtualClusterCommonNamePattern.formatted(i), "KI", "RedHat", null,
-                    null, "US");
-            var clientTrustStore = certsDirectory.resolve("demo-%d.truststore.jks".formatted(i));
-            brokerCertificateGenerator.generateTrustStore(brokerCertificateGenerator.getCertFilePath(), "client", clientTrustStore.toAbsolutePath().toString());
-            clientTrust.add(new ClientTrust(clientTrustStore, brokerCertificateGenerator.getPassword()));
+            var keystoreTrustStorePair = buildKeystoreTrustStorePair("*" + virtualClusterCommonNamePattern.formatted(i));
+            keystoreTrustStoreList.add(keystoreTrustStorePair);
 
             var virtualCluster = new VirtualClusterBuilder(base)
                     .withClusterNetworkAddressConfigProvider(
@@ -178,23 +166,22 @@ public class ExpositionIT {
                                     .build())
                     .withNewTls()
                     .withNewKeyStoreKey()
-                    .withStoreFile(brokerCertificateGenerator.getKeyStoreLocation())
-                    .withNewInlinePasswordStoreProvider(brokerCertificateGenerator.getPassword())
+                    .withStoreFile(keystoreTrustStorePair.brokerKeyStore())
+                    .withNewInlinePasswordStoreProvider(keystoreTrustStorePair.password())
                     .endKeyStoreKey()
                     .endTls()
                     .withLogNetwork(true)
                     .withLogFrames(true)
                     .build();
             builder.addToVirtualClusters("cluster" + i, virtualCluster);
-
         }
 
         try (var tester = kroxyliciousTester(builder)) {
             for (int i = 0; i < numberOfVirtualClusters; i++) {
-                var trust = clientTrust.get(i);
+                var trust = keystoreTrustStoreList.get(i);
                 try (var admin = tester.admin("cluster" + i, Map.of(
                         CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name,
-                        SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, trust.trustStore().toAbsolutePath().toString(),
+                        SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, trust.clientTrustStore(),
                         SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, trust.password()))) {
                     // do some work to ensure virtual cluster is operational
                     createTopic(admin, TOPIC + i, 1);
@@ -205,7 +192,6 @@ public class ExpositionIT {
 
     @Test
     public void exposesClusterOfTwoBrokers(@BrokerCluster(numBrokers = 2) KafkaCluster cluster) throws Exception {
-
         var builder = new ConfigurationBuilder()
                 .addToVirtualClusters("demo", new VirtualClusterBuilder()
                         .withNewTargetCluster()
@@ -231,8 +217,194 @@ public class ExpositionIT {
         }
     }
 
+    private static Stream<Arguments> virtualClusterConfigurations() throws Exception {
+        var portPerBrokerKeystoreTrustStorePair = buildKeystoreTrustStorePair("localhost");
+        var sniKeystoreTrustStorePair = buildKeystoreTrustStorePair("*." + SNI_BASE_ADDRESS);
+
+        return Stream.of(
+                Arguments.of("PortPerBroker",
+                        new VirtualClusterBuilder()
+                                .withNewTls()
+                                .withNewKeyStoreKey()
+                                .withStoreFile(portPerBrokerKeystoreTrustStorePair.brokerKeyStore())
+                                .withNewInlinePasswordStoreProvider(portPerBrokerKeystoreTrustStorePair.password())
+                                .endKeyStoreKey()
+                                .endTls()
+                                .withNewTargetCluster()
+                                .endTargetCluster()
+                                .withClusterNetworkAddressConfigProvider(
+                                        new ClusterNetworkAddressConfigProviderDefinitionBuilder("PortPerBroker")
+                                                .withConfig("bootstrapAddress", PROXY_ADDRESS)
+                                                .build()),
+                        Map.of(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name,
+                                SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, portPerBrokerKeystoreTrustStorePair.clientTrustStore(),
+                                SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, portPerBrokerKeystoreTrustStorePair.password())),
+                Arguments.of("SniRouting",
+                        new VirtualClusterBuilder()
+                                .withNewTls()
+                                .withNewKeyStoreKey()
+                                .withStoreFile(sniKeystoreTrustStorePair.brokerKeyStore())
+                                .withNewInlinePasswordStoreProvider(sniKeystoreTrustStorePair.password())
+                                .endKeyStoreKey()
+                                .endTls()
+                                .withNewTargetCluster()
+                                .endTargetCluster()
+                                .withClusterNetworkAddressConfigProvider(
+                                        new ClusterNetworkAddressConfigProviderDefinitionBuilder("SniRouting")
+                                                .withConfig("bootstrapAddress", SNI_BOOTSTRAP)
+                                                .withConfig("brokerAddressPattern", SNI_BROKER_ADDRESS_PATTERN)
+                                                .build()),
+                        Map.of(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name,
+                                SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, sniKeystoreTrustStorePair.clientTrustStore(),
+                                SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, sniKeystoreTrustStorePair.password())));
+    }
+
+    /**
+     * @see #connectToExposedBrokerEndpointsDirectlyAfterKroxyliciousRestart(VirtualClusterBuilder, Map, KafkaCluster)
+     * @param name test name
+     * @param virtualClusterBuilder the virtual cluster builder
+     * @param clientSecurityProtocolConfig addition client configuration
+     * @param cluster kafka cluster
+     */
+    @ParameterizedTest(name = "{0}")
+    @MethodSource(value = "virtualClusterConfigurations")
+    public void connectToExposedBrokerEndpointsDirectlyAfterKroxyliciousRestart(String name,
+                                                                                VirtualClusterBuilder virtualClusterBuilder,
+                                                                                Map<String, Object> clientSecurityProtocolConfig,
+                                                                                @BrokerCluster(numBrokers = 2) KafkaCluster cluster) {
+        connectToExposedBrokerEndpointsDirectlyAfterKroxyliciousRestart(virtualClusterBuilder, clientSecurityProtocolConfig, cluster);
+    }
+
+    /**
+     * @see #connectToExposedBrokerEndpointsDirectlyAfterKroxyliciousRestart(VirtualClusterBuilder, Map, KafkaCluster)
+     * @param name test name
+     * @param virtualClusterBuilder the virtual cluster builder
+     * @param clientSecurityProtocolConfig addition client configuration
+     * @param cluster kafka cluster
+     */
+    @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) {
+
+        var securityProtocol = virtualClusterBuilder.hasTls() ? SecurityProtocol.SASL_SSL : SecurityProtocol.SASL_PLAINTEXT;
+        clientSecurityProtocolConfig = new HashMap<>(clientSecurityProtocolConfig);
+        clientSecurityProtocolConfig.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol.name);
+        clientSecurityProtocolConfig.put(SaslConfigs.SASL_JAAS_CONFIG,
+                String.format("""
+                        %s required username="%s" password="%s";""",
+                        PlainLoginModule.class.getName(), SASL_USER, SASL_PASSWORD));
+        clientSecurityProtocolConfig.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
+
+        connectToExposedBrokerEndpointsDirectlyAfterKroxyliciousRestart(virtualClusterBuilder, clientSecurityProtocolConfig, cluster);
+    }
+
+    /**
+     * This test ensures that Kroxylicious, on startup, exposes all the brokers of the target cluster
+     * without requiring that a client connects to bootstrap first.  This is important for resilience: even though
+     * Kafka clients are configured with a bootstrap address there is no guarantee that the client will reconsult
+     * bootstrap when it is trying to re-establish lost connections to brokers it already knows about.
+     *
+     * @param virtualClusterBuilder the virtual cluster builder
+     * @param clientSecurityProtocolConfig addition client configuration
+     * @param cluster kafka cluster
+     */
+    private void connectToExposedBrokerEndpointsDirectlyAfterKroxyliciousRestart(VirtualClusterBuilder virtualClusterBuilder,
+                                                                                 Map<String, Object> clientSecurityProtocolConfig,
+                                                                                 KafkaCluster cluster) {
+        virtualClusterBuilder.editOrNewTargetCluster().withBootstrapServers(cluster.getBootstrapServers()).endTargetCluster();
+
+        var builder = new ConfigurationBuilder()
+                .addToVirtualClusters("demo", virtualClusterBuilder.build());
+
+        // First, learn the broker endpoints.
+
+        Collection<Node> originalNodes;
+        try (var tester = kroxyliciousTester(builder);
+                var admin = tester.admin(clientSecurityProtocolConfig)) {
+            originalNodes = await().atMost(Duration.ofSeconds(5)).until(() -> admin.describeCluster().nodes().get(),
+                    n -> n.size() == cluster.getNumOfBrokers());
+        }
+
+        LOGGER.debug("Bootstrap discovered broker nodes {}", originalNodes);
+
+        // Now, iterate across the learnt broker endpoints, connecting the client to each in turn verifying that the
+        // client discovers the remainder of the cluster. Note the Kroxylicious restart on each iteration, this
+        // ensures it has zero-state.
+        originalNodes.forEach(node -> {
+            LOGGER.debug("Testing connection to {}", node);
+            var brokerAddress = toAddress(node);
+            var brokerConfig = new HashMap<String, Object>(Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerAddress));
+            brokerConfig.putAll(clientSecurityProtocolConfig);
+
+            try (var tester = kroxyliciousTester(builder);
+                    var admin = tester.admin(brokerConfig)) {
+                var rediscoveredNodes = await().atMost(Duration.ofSeconds(5)).until(() -> admin.describeCluster().nodes().get(),
+                        n -> n.size() == cluster.getNumOfBrokers());
+                assertThat(rediscoveredNodes).containsExactlyElementsOf(originalNodes);
+            }
+        });
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource(value = "virtualClusterConfigurations")
+    public 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());
+
+        final HostPort discoveryBrokerAddressToProbe;
+        var provider = virtualClusterBuilder.getClusterNetworkAddressConfigProvider();
+        if (provider.type().equals("SniRouting")) {
+            discoveryBrokerAddressToProbe = new HostPort(SNI_BROKER_ADDRESS_PATTERN.replace("$(nodeId)", Integer.toString(cluster.getNumOfBrokers())),
+                    SNI_BOOTSTRAP.port());
+        }
+        else {
+            discoveryBrokerAddressToProbe = new HostPort(PROXY_ADDRESS.host(), PROXY_ADDRESS.port() + cluster.getNumOfBrokers() + 1);
+        }
+
+        // precondition: verify that discoveryBrokerAddressToProbe isn't actually a broker address
+        Collection<Node> originalNodes;
+        try (var tester = kroxyliciousTester(builder)) {
+            try (var admin = tester.admin(clientSecurityProtocolConfig)) {
+                originalNodes = await().atMost(Duration.ofSeconds(5)).until(() -> admin.describeCluster().nodes().get(),
+                        n -> n.size() == cluster.getNumOfBrokers());
+                var hostPorts = originalNodes.stream().map(n -> new HostPort(n.host(), n.port())).collect(Collectors.toSet());
+                assertThat(hostPorts)
+                        .describedAs("test precondition broken - fail to properly deduce a pre-bound broker address. deduced %s, cluster's nodes %s",
+                                discoveryBrokerAddressToProbe,
+                                originalNodes)
+                        .doesNotContain(discoveryBrokerAddressToProbe);
+            }
+        }
+
+        try (var tester = kroxyliciousTester(builder)) {
+            var brokerConfig = new HashMap<String, Object>(Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, discoveryBrokerAddressToProbe.toString()));
+            brokerConfig.putAll(clientSecurityProtocolConfig);
+
+            try (var admin = tester.admin(brokerConfig)) {
+                var rediscoveredNodes = await().atMost(Duration.ofSeconds(5)).until(() -> admin.describeCluster().nodes().get(),
+                        n -> n.size() == cluster.getNumOfBrokers());
+                assertThat(rediscoveredNodes).containsExactlyElementsOf(originalNodes);
+            }
+
+            // Now dial in a second time on the same pre-bound address, this confirms that kroxylicious continues to listen, even though
+            // it has reconciled.
+            try (var admin = tester.admin(brokerConfig)) {
+                var rediscoveredNodes = await().atMost(Duration.ofSeconds(5)).until(() -> admin.describeCluster().nodes().get(),
+                        n -> n.size() == cluster.getNumOfBrokers());
+                assertThat(rediscoveredNodes).containsExactlyElementsOf(originalNodes);
+            }
+        }
+    }
+
     @Test
-    public void exposedClusterAddsBroker(@BrokerCluster() KafkaCluster cluster) throws Exception {
+    public void targetClusterDynamicallyAddsBroker(@BrokerCluster KafkaCluster cluster) throws Exception {
         var builder = new ConfigurationBuilder()
                 .addToVirtualClusters("demo", new VirtualClusterBuilder()
                         .withNewTargetCluster()
@@ -264,7 +436,7 @@ public class ExpositionIT {
     }
 
     @Test
-    public void exposedClusterRemovesBroker(@BrokerCluster(numBrokers = 2) KafkaCluster cluster) throws Exception {
+    public void targetClusterDynamicallyRemovesBroker(@BrokerCluster(numBrokers = 2) KafkaCluster cluster) throws Exception {
         var builder = new ConfigurationBuilder()
                 .addToVirtualClusters("demo", new VirtualClusterBuilder()
                         .withNewTargetCluster()
@@ -351,4 +523,20 @@ public class ExpositionIT {
     private static String toAddress(Node n) {
         return n.host() + ":" + n.port();
     }
+
+    private record KeystoreTrustStorePair(String brokerKeyStore, String clientTrustStore, String password) {
+    }
+
+    @NotNull
+    private static ExpositionIT.KeystoreTrustStorePair buildKeystoreTrustStorePair(String domain) throws Exception {
+        var brokerCertificateGenerator = new KeytoolCertificateGenerator();
+        brokerCertificateGenerator.generateSelfSignedCertificateEntry("test@redhat.com", domain, "KI", "kroxylicious.io", null, null, "US");
+        Path resolve = certsDirectory.resolve(UUID.randomUUID().toString());
+        var unused = resolve.toFile().mkdirs();
+        var clientTrustStore = resolve.resolve("kafka.truststore.jks");
+        brokerCertificateGenerator.generateTrustStore(brokerCertificateGenerator.getCertFilePath(), "client",
+                clientTrustStore.toAbsolutePath().toString());
+        return new KeystoreTrustStorePair(brokerCertificateGenerator.getKeyStoreLocation(), clientTrustStore.toAbsolutePath().toString(),
+                brokerCertificateGenerator.getPassword());
+    }
 }
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/ResilienceIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/ResilienceIT.java
new file mode 100644
index 00000000..8ea16f71
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/ResilienceIT.java
@@ -0,0 +1,150 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.proxy;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.kroxylicious.proxy.config.ConfigurationBuilder;
+import io.kroxylicious.testing.kafka.api.KafkaCluster;
+import io.kroxylicious.testing.kafka.common.BrokerCluster;
+import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
+
+import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.proxy;
+import static io.kroxylicious.test.tester.KroxyliciousTesters.kroxyliciousTester;
+import static org.apache.kafka.clients.producer.ProducerConfig.CLIENT_ID_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.RETRY_BACKOFF_MS_CONFIG;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests with the aim of demonstrating that system survives a Kroxylicious restart.
+ */
+@ExtendWith(KafkaClusterExtension.class)
+public class ResilienceIT {
+    private static final Logger LOGGER = LoggerFactory.getLogger(ResilienceIT.class);
+
+    static @BrokerCluster(numBrokers = 3) KafkaCluster cluster;
+
+    @Test
+    public void kafkaProducerShouldTolerateKroxyliciousRestarting(Admin admin) throws Exception {
+        String randomTopic = UUID.randomUUID().toString();
+        admin.createTopics(List.of(new NewTopic(randomTopic, 1, (short) 1))).all().get();
+        testProducerCanSurviveARestart(proxy(cluster), randomTopic);
+    }
+
+    @Test
+    public void kafkaConsumerShouldTolerateKroxyliciousRestarting(Admin admin) throws Exception {
+        String randomTopic = UUID.randomUUID().toString();
+        admin.createTopics(List.of(new NewTopic(randomTopic, 1, (short) 1))).all().get();
+        testConsumerCanSurviveKroxyliciousRestart(proxy(cluster), randomTopic);
+    }
+
+    private static void testConsumerCanSurviveKroxyliciousRestart(ConfigurationBuilder builder, String topic)
+            throws Exception {
+        var producerConfig = new HashMap<String, Object>(Map.of(CLIENT_ID_CONFIG, "producer",
+                DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000));
+        var consumerConfig = new HashMap<String, Object>(Map.of(CLIENT_ID_CONFIG, "consumer",
+                ConsumerConfig.GROUP_ID_CONFIG, "mygroup",
+                ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"));
+
+        Consumer<String, String> consumer;
+        try (var tester = kroxyliciousTester(builder);
+                var producer = tester.producer(producerConfig)) {
+            consumer = tester.consumer(consumerConfig);
+            producer.send(new ProducerRecord<>(topic, "my-key", "Hello, world!")).get(10, TimeUnit.SECONDS);
+            consumer.subscribe(Set.of(topic));
+            var records = consumer.poll(Duration.ofSeconds(10));
+            assertThat(records).hasSize(1);
+            assertThat(records.iterator()).toIterable().map(ConsumerRecord::value).containsExactly("Hello, world!");
+
+            assertEquals(1, records.count());
+            assertEquals("Hello, world!", records.iterator().next().value());
+            producer.send(new ProducerRecord<>(topic, "my-key", "Hello, again!")).get(10, TimeUnit.SECONDS);
+        }
+
+        LOGGER.debug("Restarting proxy");
+
+        try (var ignored = kroxyliciousTester(builder)) {
+            var records = consumer.poll(Duration.ofSeconds(10));
+            assertThat(records).hasSize(1);
+            assertThat(records.iterator()).toIterable().map(ConsumerRecord::value).containsExactly("Hello, again!");
+            consumer.close();
+        }
+    }
+
+    private void testProducerCanSurviveARestart(ConfigurationBuilder builder, String topic) throws Exception {
+
+        var producerConfig = new HashMap<String, Object>(Map.of(CLIENT_ID_CONFIG, "producer",
+                DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000,
+                RECONNECT_BACKOFF_MS_CONFIG, 5,
+                RECONNECT_BACKOFF_MAX_MS_CONFIG, 100,
+                RETRY_BACKOFF_MS_CONFIG, 0));
+        var consumerConfig = new HashMap<String, Object>(Map.of(CLIENT_ID_CONFIG, "consumer",
+                ConsumerConfig.GROUP_ID_CONFIG, "mygroup",
+                ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"));
+
+        Producer<String, String> producer = null;
+        Consumer<String, String> consumer = null;
+
+        try {
+            try (var tester = kroxyliciousTester(builder)) {
+                producer = tester.producer(producerConfig);
+                consumer = tester.consumer(consumerConfig);
+                consumer.subscribe(Set.of(topic));
+                var response = producer.send(new ProducerRecord<>(topic, "my-key", "Hello, world!")).get(10, TimeUnit.SECONDS);
+                LOGGER.warn("response {}", response);
+            }
+
+            LOGGER.debug("Restarting proxy");
+
+            try (var ignored = kroxyliciousTester(builder)) {
+                // re-use the existing producer and consumer (made through Kroxylicious's first incarnation). This provides us the assurance
+                // that they were able to reconnect successfully.
+                producer.send(new ProducerRecord<>(topic, "my-key", "Hello, again!")).get(10, TimeUnit.SECONDS);
+                producer.close();
+                var records = consumer.poll(Duration.ofSeconds(20));
+                consumer.close();
+                assertThat(records).hasSize(2);
+                assertThat(records.iterator()).toIterable().map(ConsumerRecord::value).containsExactly("Hello, world!", "Hello, again!");
+            }
+        }
+        finally {
+            try {
+                if (producer != null) {
+                    producer.close();
+                }
+            }
+            finally {
+                if (consumer != null) {
+                    consumer.close();
+                }
+            }
+
+        }
+    }
+
+}
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..f3c25c88 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
@@ -6,11 +6,17 @@
 
 package io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider;
 
+import java.util.stream.Stream;
+
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.converter.ConvertWith;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.NullAndEmptySource;
 import org.junit.jupiter.params.provider.ValueSource;
 
+import io.kroxylicious.proxy.HostPortConverter;
 import io.kroxylicious.proxy.service.HostPort;
 
 import static io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.SniRoutingClusterNetworkAddressConfigProvider.SniRoutingClusterNetworkAddressConfigProviderConfig;
@@ -50,10 +56,32 @@ class SniRoutingClusterNetworkAddressConfigProviderTest {
         assertThat(provider.getBrokerAddress(0)).isEqualTo(HostPort.parse("broker-0.kafka:1234"));
     }
 
+    public static Stream<Arguments> getBrokerIdFromBrokerAddress() {
+        return Stream.of(
+                Arguments.of("broker 0", "broker-0.kafka:1234", 0),
+                Arguments.of("broker 99", "broker-99.kafka:1234", 99),
+                Arguments.of("RFC 4343 case insensitive", "BROKER-0.KAFKA:1234", 0),
+                Arguments.of("port mismatch", "broker-0.kafka:1235", null),
+                Arguments.of("host mismatch", "broker-0.another:1234", null),
+                Arguments.of("RE anchoring", "0.kafka:1234", null),
+                Arguments.of("RE anchoring", "start.broker-0.kafka.end:1234", null),
+                Arguments.of("RE metacharacters in brokerAddressPattern escaped", "broker-0xkafka:1234", null));
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource
+    void getBrokerIdFromBrokerAddress(String name, @ConvertWith(HostPortConverter.class) HostPort address, Integer expected) {
+        var provider = new SniRoutingClusterNetworkAddressConfigProvider(
+                new SniRoutingClusterNetworkAddressConfigProviderConfig(parse("boot.kafka:1234"),
+                        "broker-$(nodeId).kafka"));
+
+        assertThat(provider.getBrokerIdFromBrokerAddress(address)).isEqualTo(expected);
+    }
+
     @Test
     void badNodeId() {
         assertThrows(IllegalArgumentException.class, () -> new SniRoutingClusterNetworkAddressConfigProvider(
                 new SniRoutingClusterNetworkAddressConfigProviderConfig(parse("boot.kafka:1234"), "broker-$(nodeId).kafka"))
                 .getBrokerAddress(-1));
     }
-}
\ No newline at end of file
+}
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilterTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilterTest.java
index d37605ec..8e0451df 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilterTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilterTest.java
@@ -107,7 +107,8 @@ class BrokerAddressFilterTest {
         when(virtualCluster.getBrokerAddress(0)).thenReturn(HostPort.parse("downstream:19199"));
 
         var nodeMap = Map.of(0, HostPort.parse("upstream:9199"));
-        lenient().when(endpointReconciler.reconcile(Mockito.eq(virtualCluster), Mockito.eq(nodeMap))).thenReturn(CompletableFuture.completedStage(null));
+        lenient().when(endpointReconciler.reconcile(Mockito.eq(virtualCluster), Mockito.eq(nodeMap)))
+                .thenReturn(CompletableFuture.completedStage(null));
     }
 
     @ParameterizedTest(name = "{0}")
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearnerTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearnerTest.java
new file mode 100644
index 00000000..e30278c2
--- /dev/null
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearnerTest.java
@@ -0,0 +1,97 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.internal.filter;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
+
+import org.apache.kafka.common.message.ApiVersionsRequestData;
+import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.SaslAuthenticateRequestData;
+import org.apache.kafka.common.message.SaslHandshakeRequestData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.ApiVersionsRequest;
+import org.apache.kafka.common.requests.MetadataRequest;
+import org.apache.kafka.common.requests.ProduceRequest;
+import org.apache.kafka.common.requests.SaslAuthenticateRequest;
+import org.apache.kafka.common.requests.SaslHandshakeRequest;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+
+import static org.mockito.ArgumentMatchers.anyShort;
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class EagerMetadataLearnerTest {
+
+    @Mock
+    KrpcFilterContext context;
+    private EagerMetadataLearner learner;
+
+    @BeforeEach
+    void setUp() {
+        learner = new EagerMetadataLearner();
+    }
+
+    public static Stream<Arguments> preludeRequests() {
+        return Stream.of(
+                toArgs(new ApiVersionsRequest(new ApiVersionsRequestData(), ApiVersionsRequestData.HIGHEST_SUPPORTED_VERSION)),
+                toArgs(new SaslHandshakeRequest(new SaslHandshakeRequestData(), SaslHandshakeRequestData.HIGHEST_SUPPORTED_VERSION)),
+                toArgs(new SaslAuthenticateRequest(new SaslAuthenticateRequestData(), SaslHandshakeRequestData.HIGHEST_SUPPORTED_VERSION)));
+    }
+
+    @ParameterizedTest
+    @MethodSource("preludeRequests")
+    public void forwardsRequestsOfKafkaPrelude(ApiKeys apiKey, RequestHeaderData header, ApiMessage request) {
+        learner.onRequest(apiKey, header, request, context);
+        verify(context).forwardRequest(header, request);
+    }
+
+    public static Stream<Arguments> postPreludeRequests() {
+        return Stream.of(
+                toArgs(new ProduceRequest(new ProduceRequestData(), ProduceRequestData.HIGHEST_SUPPORTED_VERSION)),
+                toArgs(new MetadataRequest(new MetadataRequestData(), MetadataRequestData.HIGHEST_SUPPORTED_VERSION)),
+                toArgs(new MetadataRequest(new MetadataRequestData(), MetadataRequestData.LOWEST_SUPPORTED_VERSION)));
+    }
+
+    @ParameterizedTest
+    @MethodSource("postPreludeRequests")
+    public void spontaneouslyEmitsMetadataRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage request) {
+        var metadataResponse = new MetadataResponseData();
+        metadataResponse.brokers().add(new MetadataResponseData.MetadataResponseBroker().setNodeId(1).setHost("localhost").setPort(1234));
+
+        when(context.sendRequest(anyShort(), isA(MetadataRequestData.class))).thenReturn(CompletableFuture.completedStage(metadataResponse));
+        learner.onRequest(apiKey, header, request, context);
+
+        verify(context, apiKey.equals(ApiKeys.METADATA) ? times(1) : never()).forwardResponse(metadataResponse);
+        verify(context).closeConnection();
+    }
+
+    private static Arguments toArgs(AbstractRequest request) {
+        var header = new RequestHeaderData().setRequestApiKey(request.apiKey().id).setRequestApiVersion(request.version());
+        var apiKey = request.apiKey();
+        var request1 = request.data();
+        return Arguments.of(apiKey, header, request1);
+    }
+
+}
\ No newline at end of file
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/net/EndpointRegistryTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/net/EndpointRegistryTest.java
index 28c3a710..36bc2e1c 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/net/EndpointRegistryTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/net/EndpointRegistryTest.java
@@ -39,6 +39,7 @@ import io.kroxylicious.proxy.service.HostPort;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mock.Strictness.LENIENT;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.mock;
@@ -51,6 +52,10 @@ class EndpointRegistryTest {
     private static final HostPort DOWNSTREAM_BOOTSTRAP_DIFF_PORT = new HostPort(DOWNSTREAM_BOOTSTRAP.host(), DOWNSTREAM_BOOTSTRAP.port() + 1);
     private static final HostPort DOWNSTREAM_BROKER_0 = HostPort.parse("downstream-broker0:9193");
     private static final HostPort DOWNSTREAM_BROKER_1 = HostPort.parse("downstream-broker1:9194");
+    private static final String SNI_DOMAIN_SUFFIX = ".cluster.kafka.com";
+    private static final HostPort SNI_DOWNSTREAM_BOOTSTRAP = new HostPort("bootstrap" + SNI_DOMAIN_SUFFIX, 9192);
+    private static final HostPort SNI_DOWNSTREAM_BROKER_0 = new HostPort("broker-0" + SNI_DOMAIN_SUFFIX, SNI_DOWNSTREAM_BOOTSTRAP.port());
+    private static final HostPort SNI_DOWNSTREAM_BROKER_1 = new HostPort("broker-1" + SNI_DOMAIN_SUFFIX, SNI_DOWNSTREAM_BOOTSTRAP.port());
     private static final HostPort UPSTREAM_BOOTSTRAP = HostPort.parse("upstream-bootstrap:19192");
     private static final HostPort UPSTREAM_BROKER_0 = HostPort.parse("upstream-broker0:19193");
     private static final HostPort UPSTREAM_BROKER_1 = HostPort.parse("upstream-broker1:19194");
@@ -73,6 +78,23 @@ class EndpointRegistryTest {
         assertThat(endpointRegistry.listeningChannelCount()).isEqualTo(1);
     }
 
+    @Test
+    public void registerVirtualClusterWithDiscoveryAddresses() throws Exception {
+        configureVirtualClusterMock(virtualCluster1, DOWNSTREAM_BOOTSTRAP, UPSTREAM_BOOTSTRAP, false, false,
+                Map.of(0, DOWNSTREAM_BROKER_0, 1, DOWNSTREAM_BROKER_1));
+
+        var rf = endpointRegistry.registerVirtualCluster(virtualCluster1).toCompletableFuture();
+        verifyAndProcessNetworkEventQueue(createTestNetworkBindRequest(DOWNSTREAM_BOOTSTRAP.port(), false),
+                createTestNetworkBindRequest(DOWNSTREAM_BROKER_0.port(), false),
+                createTestNetworkBindRequest(DOWNSTREAM_BROKER_1.port(), false));
+        assertThat(rf).isCompleted();
+
+        verifyVirtualClusterRegisterFuture(DOWNSTREAM_BOOTSTRAP.port(), false, rf);
+
+        assertThat(endpointRegistry.isRegistered(virtualCluster1)).isTrue();
+        assertThat(endpointRegistry.listeningChannelCount()).isEqualTo(3);
+    }
+
     @Test
     public void registerVirtualClusterTls() throws Exception {
         configureVirtualClusterMock(virtualCluster1, DOWNSTREAM_BOOTSTRAP, UPSTREAM_BOOTSTRAP, true);
@@ -276,7 +298,7 @@ class EndpointRegistryTest {
     public void resolveBootstrap(@ConvertWith(HostPortConverter.class) HostPort downstreamBootstrap, @ConvertWith(HostPortConverter.class) HostPort upstreamBootstrap,
                                  boolean tls, boolean sni)
             throws Exception {
-        configureVirtualClusterMock(virtualCluster1, HostPort.parse(downstreamBootstrap.toString()), HostPort.parse(upstreamBootstrap.toString()), tls, sni);
+        configureVirtualClusterMock(virtualCluster1, HostPort.parse(downstreamBootstrap.toString()), HostPort.parse(upstreamBootstrap.toString()), tls, sni, null);
 
         var f = endpointRegistry.registerVirtualCluster(virtualCluster1).toCompletableFuture();
         verifyAndProcessNetworkEventQueue(createTestNetworkBindRequest(downstreamBootstrap.port(), tls));
@@ -292,8 +314,7 @@ class EndpointRegistryTest {
     public void resolveBootstrapResolutionFailures(String name,
                                                    @ConvertWith(HostPortConverter.class) HostPort downstreamBootstrap,
                                                    @ConvertWith(HostPortConverter.class) HostPort upstreamBootstrap,
-                                                   @ConvertWith(HostPortConverter.class) HostPort resolveAddress)
-            throws Exception {
+                                                   @ConvertWith(HostPortConverter.class) HostPort resolveAddress) {
         configureVirtualClusterMock(virtualCluster1, HostPort.parse(downstreamBootstrap.toString()), HostPort.parse(upstreamBootstrap.toString()), true);
 
         var f = endpointRegistry.registerVirtualCluster(virtualCluster1).toCompletableFuture();
@@ -322,6 +343,61 @@ class EndpointRegistryTest {
         assertThat(binding).isEqualTo(new VirtualClusterBootstrapBinding(virtualCluster1, upstreamBootstrap));
     }
 
+    @Test
+    public void resolveUsingSniMatch() throws Exception {
+        configureVirtualClusterMock(virtualCluster1, SNI_DOWNSTREAM_BOOTSTRAP, UPSTREAM_BOOTSTRAP, true, true, null);
+
+        var regf = endpointRegistry.registerVirtualCluster(virtualCluster1).toCompletableFuture();
+        verifyAndProcessNetworkEventQueue(createTestNetworkBindRequest(SNI_DOWNSTREAM_BOOTSTRAP.port(), true));
+        assertThat(regf).isCompleted();
+
+        when(virtualCluster1.getBrokerIdFromBrokerAddress(SNI_DOWNSTREAM_BROKER_0)).thenReturn(0);
+        var sniFallbackBinding = endpointRegistry.resolve(Endpoint.createEndpoint(SNI_DOWNSTREAM_BROKER_0.port(), true), SNI_DOWNSTREAM_BROKER_0.host())
+                .toCompletableFuture().get();
+        assertThat(sniFallbackBinding).isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, UPSTREAM_BOOTSTRAP, 0, true));
+    }
+
+    @Test
+    public void resolveUsingSniFailsDueToMismatch() {
+        configureVirtualClusterMock(virtualCluster1, SNI_DOWNSTREAM_BOOTSTRAP, UPSTREAM_BOOTSTRAP, true, true, null);
+
+        var regf = endpointRegistry.registerVirtualCluster(virtualCluster1).toCompletableFuture();
+        verifyAndProcessNetworkEventQueue(createTestNetworkBindRequest(SNI_DOWNSTREAM_BOOTSTRAP.port(), true));
+        assertThat(regf).isCompleted();
+
+        when(virtualCluster1.getBrokerIdFromBrokerAddress(SNI_DOWNSTREAM_BROKER_0)).thenReturn(null);
+        var executionException = assertThrows(ExecutionException.class,
+                () -> endpointRegistry.resolve(Endpoint.createEndpoint(SNI_DOWNSTREAM_BROKER_0.port(), true), SNI_DOWNSTREAM_BROKER_0.host()).toCompletableFuture()
+                        .get());
+        assertThat(executionException).hasCauseInstanceOf(EndpointResolutionException.class);
+    }
+
+    @Test
+    public void resolveIgnoresRestrictedSniMatchAfterReconcile() throws Exception {
+        configureVirtualClusterMock(virtualCluster1, SNI_DOWNSTREAM_BOOTSTRAP, UPSTREAM_BOOTSTRAP, true, true, null);
+
+        var regf = endpointRegistry.registerVirtualCluster(virtualCluster1).toCompletableFuture();
+        verifyAndProcessNetworkEventQueue(createTestNetworkBindRequest(SNI_DOWNSTREAM_BOOTSTRAP.port(), true));
+        assertThat(regf).isCompleted();
+
+        when(virtualCluster1.getBrokerAddress(0)).thenReturn(SNI_DOWNSTREAM_BROKER_0);
+        var recf = endpointRegistry.reconcile(virtualCluster1, Map.of(0, UPSTREAM_BROKER_0)).toCompletableFuture();
+        verifyAndProcessNetworkEventQueue();
+        assertThat(recf).isCompleted();
+
+        when(virtualCluster1.getBrokerIdFromBrokerAddress(SNI_DOWNSTREAM_BROKER_0)).thenReturn(0);
+        var brokerBinding = endpointRegistry.resolve(Endpoint.createEndpoint(SNI_DOWNSTREAM_BROKER_0.port(), true), SNI_DOWNSTREAM_BROKER_0.host()).toCompletableFuture()
+                .get();
+        assertThat(brokerBinding).isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, UPSTREAM_BROKER_0, 0, false));
+
+        // Now try to resolve another broker address that matches the pattern, but isn't resolved.
+        when(virtualCluster1.getBrokerIdFromBrokerAddress(SNI_DOWNSTREAM_BROKER_1)).thenReturn(1);
+        var sniMatchBinding = endpointRegistry.resolve(Endpoint.createEndpoint(SNI_DOWNSTREAM_BROKER_1.port(), true), SNI_DOWNSTREAM_BROKER_1.host())
+                .toCompletableFuture()
+                .get();
+        assertThat(sniMatchBinding).isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, UPSTREAM_BOOTSTRAP, 1, true));
+    }
+
     @Test
     public void bindingAddressEndpointSeparation() throws Exception {
         var bindingAddress1 = Optional.of("127.0.0.1");
@@ -389,7 +465,21 @@ class EndpointRegistryTest {
         assertThat(endpointRegistry.listeningChannelCount()).isEqualTo(2);
 
         var binding = endpointRegistry.resolve(Endpoint.createEndpoint(DOWNSTREAM_BROKER_0.port(), false), null).toCompletableFuture().get();
-        assertThat(binding).isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, UPSTREAM_BROKER_0, 0));
+        assertThat(binding).isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, UPSTREAM_BROKER_0, 0, false));
+    }
+
+    @Test
+    public void resolveDiscoveryBrokerAddress() throws Exception {
+        configureVirtualClusterMock(virtualCluster1, DOWNSTREAM_BOOTSTRAP, UPSTREAM_BOOTSTRAP, false, false, Map.of(0, DOWNSTREAM_BROKER_0));
+
+        var regf = endpointRegistry.registerVirtualCluster(virtualCluster1).toCompletableFuture();
+        verifyAndProcessNetworkEventQueue(createTestNetworkBindRequest(DOWNSTREAM_BOOTSTRAP.port(), false),
+                createTestNetworkBindRequest(DOWNSTREAM_BROKER_0.port(), false));
+        assertThat(regf).isCompleted();
+        assertThat(endpointRegistry.listeningChannelCount()).isEqualTo(2);
+
+        var binding = endpointRegistry.resolve(Endpoint.createEndpoint(DOWNSTREAM_BROKER_0.port(), false), null).toCompletableFuture().get();
+        assertThat(binding).isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, UPSTREAM_BOOTSTRAP, 0, true));
     }
 
     @Test
@@ -438,7 +528,7 @@ class EndpointRegistryTest {
         assertThat(endpointRegistry.listeningChannelCount()).isEqualTo(2);
 
         var resolvedBindingBeforeChange = endpointRegistry.resolve(Endpoint.createEndpoint(DOWNSTREAM_BROKER_0.port(), false), null).toCompletableFuture().get();
-        assertThat(resolvedBindingBeforeChange).isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, UPSTREAM_BROKER_0, 0));
+        assertThat(resolvedBindingBeforeChange).isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, UPSTREAM_BROKER_0, 0, false));
 
         // Target cluster updates the address for broker 0
         var recf2 = endpointRegistry.reconcile(virtualCluster1, Map.of(0, upstreamBrokerUpdated0)).toCompletableFuture();
@@ -448,7 +538,45 @@ class EndpointRegistryTest {
         assertThat(endpointRegistry.listeningChannelCount()).isEqualTo(2);
 
         var resolvedBindingAfterChange = endpointRegistry.resolve(Endpoint.createEndpoint(DOWNSTREAM_BROKER_0.port(), false), null).toCompletableFuture().get();
-        assertThat(resolvedBindingAfterChange).isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, upstreamBrokerUpdated0, 0));
+        assertThat(resolvedBindingAfterChange).isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, upstreamBrokerUpdated0, 0, false));
+    }
+
+    @Test
+    public void reconcileReplacesDiscoveryAddress() throws Exception {
+        configureVirtualClusterMock(virtualCluster1, DOWNSTREAM_BOOTSTRAP, UPSTREAM_BOOTSTRAP, false, false,
+                Map.of(0, DOWNSTREAM_BROKER_0, 1, DOWNSTREAM_BROKER_1));
+
+        var regf = endpointRegistry.registerVirtualCluster(virtualCluster1).toCompletableFuture();
+        verifyAndProcessNetworkEventQueue(createTestNetworkBindRequest(DOWNSTREAM_BOOTSTRAP.port(), false),
+                createTestNetworkBindRequest(DOWNSTREAM_BROKER_0.port(), false),
+                createTestNetworkBindRequest(DOWNSTREAM_BROKER_1.port(), false));
+        assertThat(regf).isDone();
+
+        var resolveBroker0BeforeReconcile = endpointRegistry.resolve(Endpoint.createEndpoint(DOWNSTREAM_BROKER_0.port(), false), null).toCompletableFuture().get();
+        assertThat(resolveBroker0BeforeReconcile)
+                .describedAs("Resolving pre-bound broker 0 should yield the upstream address for bootstrap")
+                .isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, UPSTREAM_BOOTSTRAP, 0, true));
+
+        // Reconcile learns that upstream broker topology actually has only one broker
+        // DOWNSTREAM_BROKER_0 binding has to be updated
+        when(virtualCluster1.getBrokerAddress(0)).thenReturn(DOWNSTREAM_BROKER_0);
+        var reconcileFuture = endpointRegistry.reconcile(virtualCluster1, Map.of(0, UPSTREAM_BROKER_0)).toCompletableFuture();
+        verifyAndProcessNetworkEventQueue();
+        assertThat(reconcileFuture).isCompleted();
+        assertThat(endpointRegistry.listeningChannelCount()).isEqualTo(3);
+
+        // Now resolving broker 0 yields the upstream address for broker 0
+        var resolveBroker0AfterReconcile = endpointRegistry.resolve(Endpoint.createEndpoint(DOWNSTREAM_BROKER_0.port(), false), null).toCompletableFuture().get();
+        assertThat(resolveBroker0AfterReconcile)
+                .describedAs("Resolving reconciled broker 0 should yield the actual upstream address for broker0")
+                .isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, UPSTREAM_BROKER_0, 0, false));
+
+        // And resolving broker 1 still yields the bootstrap
+        var resolveBroker1AfterReconcile = endpointRegistry.resolve(Endpoint.createEndpoint(DOWNSTREAM_BROKER_1.port(), false), null).toCompletableFuture().get();
+        assertThat(resolveBroker1AfterReconcile)
+                .describedAs("Resolving reconciled broker 1 should still yield bootstrap")
+                .isEqualTo(new VirtualClusterBrokerBinding(virtualCluster1, UPSTREAM_BOOTSTRAP, 1, true));
+
     }
 
     @Test
@@ -599,14 +727,17 @@ class EndpointRegistryTest {
     }
 
     private void configureVirtualClusterMock(VirtualCluster cluster, HostPort downstreamBootstrap, HostPort upstreamBootstrap, boolean tls) {
-        configureVirtualClusterMock(cluster, downstreamBootstrap, upstreamBootstrap, tls, tls);
+        configureVirtualClusterMock(cluster, downstreamBootstrap, upstreamBootstrap, tls, tls, null);
     }
 
-    private void configureVirtualClusterMock(VirtualCluster cluster, HostPort downstreamBootstrap, HostPort upstreamBootstrap, boolean tls, boolean sni) {
+    private void configureVirtualClusterMock(VirtualCluster cluster, HostPort downstreamBootstrap, HostPort upstreamBootstrap, boolean tls, boolean sni,
+                                             Map<Integer, HostPort> discoveryAddressMap) {
         when(cluster.getClusterBootstrapAddress()).thenReturn(downstreamBootstrap);
         when(cluster.isUseTls()).thenReturn(tls);
         when(cluster.requiresTls()).thenReturn(sni);
+        when(cluster.discoveryAddressMap()).thenReturn(discoveryAddressMap);
         when(cluster.targetCluster()).thenReturn(new TargetCluster(upstreamBootstrap.toString(), null));
+        when(cluster.getBrokerIdFromBrokerAddress(any(HostPort.class))).thenReturn(null);
     }
 
     private void verifyVirtualClusterRegisterFuture(int expectedPort, boolean expectedTls, CompletableFuture<Endpoint> future) throws Exception {
diff --git a/performance-tests/pom.xml b/performance-tests/pom.xml
index d476484b..93e5c507 100644
--- a/performance-tests/pom.xml
+++ b/performance-tests/pom.xml
@@ -30,7 +30,7 @@
         <!-- project dependencies - runtime and compile -->
         <dependency>
             <groupId>io.kroxylicious</groupId>
-            <artifactId>kroxylicious-filter-api</artifactId>
+            <artifactId>kroxylicious-api</artifactId>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
diff --git a/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
index 45995764..d47ca7fd 100644
--- a/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
+++ b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
@@ -188,5 +188,10 @@ public class InvokerDispatchBenchmark {
         public void forwardResponse(ApiMessage response) {
 
         }
+
+        @Override
+        public void closeConnection() {
+
+        }
     }
 }