Repository: kroxylicious/kroxylicious
Pull request #406 (2023-06-23 09:34:45)

docs: update how it works section to describe
### Type of change


- Documentation

### Description

- virtual clusters
- target clusters
- how virtual clusters expose endpoints
- upstream/downstream TLS

### 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
- [ ] Make sure all Sonar-Lint warnings are addressed or are justifiably ignored.
- [ ] Update documentation
- [ ] Reference relevant issue(s) and close them after merging
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


k-wall: @SamBarker @robobario hello - could you review the shape of documentation I am proposing here.  I haven't had chance to proof read today, so please don't waste time doing that.  I'm really looking a sense of if I'm hitting the right level of detail and the structure makes sense.
SamBarker: @gracegrimwood & @showuon could you give this a read too? Its all still fresh for you both. 
gracegrimwood: LGTM 😃 The diagrams are definitely a helpful addition too
k-wall: The PR is properly ready for review. All comments welcome.  

Patch:
diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml
index cbca882a..cdf0baad 100644
--- a/.github/workflows/docker.yml
+++ b/.github/workflows/docker.yml
@@ -3,6 +3,8 @@ name: Docker Build
 on:
   workflow_dispatch:
   push:
+    branches:
+      - 'main'
     tags:
       - 'v*.*.*'
 
diff --git a/CHANGELOG.md b/CHANGELOG.md
index dc5c721d..d177567d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,8 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 
 ## 0.3.0
 
+* [#392](https://github.com/kroxylicious/kroxylicious/pull/392): Introduce CompositeFilters
+* [#401](https://github.com/kroxylicious/kroxylicious/pull/401): Fix netty buffer leak when doing a short-circuit response
 * [#409](https://github.com/kroxylicious/kroxylicious/pull/409): Bump netty.version from 4.1.93.Final to 4.1.94.Final #409 
 * [#374](https://github.com/kroxylicious/kroxylicious/issues/374) Upstream TLS support
 * [#375](https://github.com/kroxylicious/kroxylicious/issues/375) Support key material in PEM format (X.509 certificates and PKCS-8 private keys) 
@@ -18,3 +20,7 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 
 The default metrics port has changed from 9193 to 9190 to prevent port collisions
 
+Filter Authors can now implement CompositeFilter if they want a single configuration block to contribute multiple Filters
+to the Filter chain. This enables them to write smaller, more focused Filter implementations but deliver them as a whole
+behaviour with a single block of configuration in the Kroxylicious configuration yaml. This interface is mutually exclusive
+with RequestFilter, ResponseFilter or any specific message Filter interfaces.
diff --git a/Dockerfile b/Dockerfile
index ce85162a..7041d4d2 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -9,7 +9,7 @@ FROM registry.access.redhat.com/ubi9/openjdk-17:1.15 AS builder
 USER root
 WORKDIR /opt/kroxylicious
 COPY . .
-RUN ./mvnw clean verify -Pdist -Dquick -pl :kroxylicious -am
+RUN ./mvnw -B clean verify -Pdist -Dquick -pl :kroxylicious -am
 USER 185
 FROM registry.access.redhat.com/ubi9/ubi-minimal:9.2
 
diff --git a/api/kroxylicious-api/pom.xml b/api/kroxylicious-api/pom.xml
index 6d4174bb..049cf65d 100644
--- a/api/kroxylicious-api/pom.xml
+++ b/api/kroxylicious-api/pom.xml
@@ -26,19 +26,20 @@
     <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>
-        <!-- Test dependencies -->
+        <!-- third party dependencies - test -->
         <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter</artifactId>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.assertj</groupId>
-            <artifactId>assertj-core</artifactId>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
diff --git a/api/kroxylicious-filter-api/pom.xml b/api/kroxylicious-filter-api/pom.xml
index 3ecbf2dd..d662a23a 100644
--- a/api/kroxylicious-filter-api/pom.xml
+++ b/api/kroxylicious-filter-api/pom.xml
@@ -30,10 +30,16 @@
     <description>The public interfaces, classes, etc required to author custom filters and configure Kroxylicious to use them</description>
 
     <dependencies>
+        <!-- project dependencies - runtime and compile -->
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-api</artifactId>
-            <version>${kroxyliciousApi.version}</version>
+        </dependency>
+
+        <!-- third party dependencies - runtime and compile -->
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
         </dependency>
         <dependency>
             <groupId>io.micrometer</groupId>
@@ -47,9 +53,22 @@
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
         </dependency>
+
+        <!-- third party dependencies - test -->
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-annotations</artifactId>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <scope>test</scope>
         </dependency>
     </dependencies>
 
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/CompositeFilter.java b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/CompositeFilter.java
new file mode 100644
index 00000000..b47b60ce
--- /dev/null
+++ b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/CompositeFilter.java
@@ -0,0 +1,36 @@
+/*
+ * 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.filter;
+
+import java.util.List;
+
+/**
+ * Filters can implement this when they are composed of multiple discrete
+ * filters. If a Filter implements CompositeFilter it cannot implement
+ * {@link RequestFilter}, {@link ResponseFilter} or any of the specific
+ * message Filter interfaces like {@link ApiVersionsRequestFilter}.
+ *<p>
+ * This is useful if you want to create smaller component filters that are
+ * individually testable, but only require a single entry in the
+ * kroxylicious configuration to install your behaviour as a whole. An
+ * example would be if you want to use a {@link RequestFilter} to modify
+ * all clientId headers, but also want to modify some specific RPCs (like
+ * mutate produce messages with an {@link ProduceRequestFilter} implementation).
+ * The interfaces are incompatible so you must implement two different filter
+ * classes. The CompositeFilter enables you to install both filters with a single
+ * user configuration.
+ *</p>
+ */
+public interface CompositeFilter extends KrpcFilter {
+
+    /**
+     * Get composed KrpcFilters
+     * @return filters
+     */
+    List<KrpcFilter> getFilters();
+
+}
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterAndInvoker.java b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterAndInvoker.java
new file mode 100644
index 00000000..cfae1af5
--- /dev/null
+++ b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterAndInvoker.java
@@ -0,0 +1,32 @@
+/*
+ * 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.filter;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * A Filter and it's respective invoker
+ * @param filter filter
+ * @param invoker invoker
+ */
+public record FilterAndInvoker(KrpcFilter filter, FilterInvoker invoker) {
+    public FilterAndInvoker {
+        requireNonNull(filter, "filter cannot be null");
+        requireNonNull(invoker, "invoker cannot be null");
+    }
+
+    /**
+     * Builds a list of invokers for a filter
+     * @param filter filter
+     * @return a filter and its respective invoker
+     */
+    public static List<FilterAndInvoker> build(KrpcFilter filter) {
+        return FilterInvokers.from(filter);
+    }
+}
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvokers.java b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvokers.java
index c4c612a1..b914d4ca 100644
--- a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvokers.java
+++ b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/FilterInvokers.java
@@ -6,12 +6,17 @@
 
 package io.kroxylicious.proxy.filter;
 
+import java.util.List;
+import java.util.stream.Stream;
+
 /**
  * Factory for FilterInvokers. The intention is to keep the Invoker implementations
  * as private as we can, so that invocation is a framework concern.
  */
 public class FilterInvokers {
 
+    public static final int RECURSION_DEPTH_LIMIT = 2;
+
     private FilterInvokers() {
 
     }
@@ -23,42 +28,79 @@ public class FilterInvokers {
      *     <li>A KrpcFilter implementing {@link RequestFilter}</li>
      *     <li>A KrpcFilter implementing both {@link ResponseFilter} and {@link RequestFilter} </li>
      *     <li>A KrpcFilter implementing any number of Specific Message Filter interfaces</li>
+     *     <li>A KrpcFilter implementing {@link CompositeFilter}</li>
      * </ol>
-     * @throws IllegalArgumentException if specific Message Filter interfaces are mixed with {@link RequestFilter} or  {@link ResponseFilter}
+     * Examples of unsupported cases are:
+     * <ol>
+     *     <li>A KrpcFilter implementing {@link ResponseFilter} and any number of Specific Message Filter interfaces</li>
+     *     <li>A KrpcFilter implementing {@link CompositeFilter} and any number of Specific Message Filter interfaces</li>
+     *     <li>A KrpcFilter implementing {@link ResponseFilter} and {@link CompositeFilter}</li>
+     *     <li>A KrpcFilter implementing {@link RequestFilter} and {@link CompositeFilter}</li>
+     * </ol>
+     * @throws IllegalArgumentException if there is an invalid combination of Filter interfaces
      * @throws IllegalArgumentException if none of the supported interfaces are implemented
      * @param filter the Filter to create an invoker for
      * @return the invoker
      */
-    public static FilterInvoker from(KrpcFilter filter) {
+    static List<FilterAndInvoker> from(KrpcFilter filter) {
+        List<FilterAndInvoker> filterInvokers = invokersForFilter(filter, 0);
         // all invokers are wrapped in safe invoker so that clients can safely call onRequest/onResponse
         // even if the invoker isn't interested in that message.
-        return new SafeInvoker(invokerForFilter(filter));
+        return wrapAllInSafeInvoker(filterInvokers).toList();
     }
 
-    private static FilterInvoker invokerForFilter(KrpcFilter filter) {
+    private static List<FilterAndInvoker> invokersForFilter(KrpcFilter filter, int depth) {
+        boolean isCompositeFilter = filter instanceof CompositeFilter;
         boolean isResponseFilter = filter instanceof ResponseFilter;
         boolean isRequestFilter = filter instanceof RequestFilter;
         boolean isAnySpecificFilterInterface = SpecificFilterArrayInvoker.implementsAnySpecificFilterInterface(filter);
-        if (isAnySpecificFilterInterface && (isRequestFilter || isResponseFilter)) {
-            throw unsupportedFilterInstance(filter, "Cannot mix specific message filter interfaces and [RequestFilter|ResponseFilter] interfaces");
-        }
-        if (!isRequestFilter && !isResponseFilter && !isAnySpecificFilterInterface) {
-            throw unsupportedFilterInstance(filter, "KrpcFilter must implement ResponseFilter, RequestFilter or any combination of specific message Filter interfaces");
+        validateFilter(filter, isResponseFilter, isRequestFilter, isAnySpecificFilterInterface, isCompositeFilter);
+        if (isCompositeFilter) {
+            List<KrpcFilter> composedFilters = ((CompositeFilter) filter).getFilters();
+            if (depth >= RECURSION_DEPTH_LIMIT) {
+                throw new IllegalArgumentException("CompositeFilter's were nested too deeply, exceeded recursion depth limit of " + RECURSION_DEPTH_LIMIT);
+            }
+            return composedFilters.stream().flatMap(composedFilter -> invokersForFilter(composedFilter, depth + 1).stream()).toList();
         }
-        if (isResponseFilter && isRequestFilter) {
-            return new RequestResponseInvoker((RequestFilter) filter, (ResponseFilter) filter);
+        else if (isResponseFilter && isRequestFilter) {
+            return singleFilterAndInvoker(filter, new RequestResponseInvoker((RequestFilter) filter, (ResponseFilter) filter));
         }
         else if (isRequestFilter) {
-            return new RequestFilterInvoker((RequestFilter) filter);
+            return singleFilterAndInvoker(filter, new RequestFilterInvoker((RequestFilter) filter));
         }
         else if (isResponseFilter) {
-            return new ResponseFilterInvoker((ResponseFilter) filter);
+            return singleFilterAndInvoker(filter, new ResponseFilterInvoker((ResponseFilter) filter));
         }
         else {
-            return arrayInvoker(filter);
+            return singleFilterAndInvoker(filter, arrayInvoker(filter));
         }
     }
 
+    private static Stream<FilterAndInvoker> wrapAllInSafeInvoker(List<FilterAndInvoker> filterInvokers) {
+        return filterInvokers.stream().map(filterAndInvoker -> new FilterAndInvoker(filterAndInvoker.filter(), new SafeInvoker(filterAndInvoker.invoker())));
+    }
+
+    private static void validateFilter(KrpcFilter filter, boolean isResponseFilter, boolean isRequestFilter, boolean isAnySpecificFilterInterface,
+                                       boolean isCompositeFilter) {
+        if (isAnySpecificFilterInterface && isCompositeFilter) {
+            throw unsupportedFilterInstance(filter, "Cannot mix specific message filter interfaces and CompositeFilter interfaces");
+        }
+        if ((isRequestFilter || isResponseFilter) && isCompositeFilter) {
+            throw unsupportedFilterInstance(filter, "Cannot mix [RequestFilter|ResponseFilter] interfaces and CompositeFilter interfaces");
+        }
+        if (isAnySpecificFilterInterface && (isRequestFilter || isResponseFilter)) {
+            throw unsupportedFilterInstance(filter, "Cannot mix specific message filter interfaces and [RequestFilter|ResponseFilter] interfaces");
+        }
+        if (!isRequestFilter && !isResponseFilter && !isAnySpecificFilterInterface && !isCompositeFilter) {
+            throw unsupportedFilterInstance(filter,
+                    "KrpcFilter must implement ResponseFilter, RequestFilter, CompositeFilter or any combination of specific message Filter interfaces");
+        }
+    }
+
+    private static List<FilterAndInvoker> singleFilterAndInvoker(KrpcFilter filter, FilterInvoker invoker) {
+        return List.of(new FilterAndInvoker(filter, invoker));
+    }
+
     /**
      * Create an invoker for this filter that avoids instanceof when deciding
      * if the filter should be consulted/handle messages. Instead, it stores
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilter.java b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilter.java
index fd1d1466..7dc78aea 100644
--- a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilter.java
+++ b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/RequestFilter.java
@@ -12,7 +12,10 @@ import org.apache.kafka.common.protocol.ApiMessage;
 
 /**
  * A KrpcFilter implementation intended to simplify cases where we want to handle all or
- * most request types, for example to modify the request headers.
+ * most request types, for example to modify the request headers. If a Filter implements
+ * RequestFilter, it cannot also implement {@link CompositeFilter} or any of the specific
+ * message Filter interfaces like {@link ApiVersionsRequestFilter}. If a Filter implements
+ * RequestFilter, it may also implement {@link ResponseFilter}.
  */
 public interface RequestFilter extends KrpcFilter {
 
diff --git a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilter.java b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilter.java
index 784dbe86..e5ffaea9 100644
--- a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilter.java
+++ b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/ResponseFilter.java
@@ -12,7 +12,10 @@ import org.apache.kafka.common.protocol.ApiMessage;
 
 /**
  * A KrpcFilter implementation intended to simplify cases where we want to handle all or
- * most response types, for example to modify the response headers.
+ * most response types, for example to modify the response headers. If a Filter implements
+ * ResponseFilter, it cannot also implement {@link CompositeFilter} or any of the specific
+ * message Filter interfaces like {@link ApiVersionsRequestFilter}. If a Filter implements
+ * ResponseFilter, it may also implement {@link RequestFilter}.
  */
 public interface ResponseFilter extends KrpcFilter {
 
diff --git a/docs/available-filters.adoc b/docs/available-filters.adoc
index 905b089e..e15dadae 100644
--- a/docs/available-filters.adoc
+++ b/docs/available-filters.adoc
@@ -6,6 +6,9 @@ Only a few built in filters are provided.
 
 // TODO list them and describe them
 
+=== Schema-validation
+
+=== Multi-tenancy
 
 // == 3rd party filters
 //
diff --git a/docs/deploying.adoc b/docs/deploying.adoc
index 786d7c4b..c267558c 100644
--- a/docs/deploying.adoc
+++ b/docs/deploying.adoc
@@ -9,11 +9,257 @@ Put the filter jars in some directory
 
 
 
-== Configuring the proxy networking
+== Configuring virtual clusters
+
+As described earlier, the _Virtual Cluster_ is the downstream representation of a Kafka Cluster.  Kafka Client connect
+to the Virtual Cluster.
+
+You must define at least one Virtual Cluster.
+
+Let's look at how that is done by considering first a simple example.  After we will look at more advanced options
+including TLS.
+
+[source, yaml]
+----
+virtualClusters:
+  demo:                                         # <1>
+    targetCluster:
+      bootstrap_servers: myprivatecluster:9092  # <2>
+    clusterNetworkAddressConfigProvider:
+      type: PortPerBroker                       # <3>
+      config:
+        bootstrapAddress: mypublickroxylicious:9192    # <4>
+----
+<1> The name of the virtual cluster.
+<2> The bootstrap of the (physical) Kafka Cluster.  This is the Kafka Cluster being proxied.
+<3> The name of a cluster network address config provider. The built-in types are `PortPerBroker` and `SniRouting`.
+<4> The hostname and port of the bootstrap that will be used by the Kafka Clients.  The hostname must be resolved
+by the clients.
+
+This configuration declares a virtual cluster called `demo`.  The physical Kafka Cluster being proxied is the defined
+by the `targetCluster` element.  In this example, the `PortPerBroker` scheme is used by Kroxlicious to present the
+virtual cluster to the clients.  Under this schema, Kroxylicious will open a port for each broker of the target cluster
+with port numbers beginning at `9192` +1.  So, if the target cluster has three brokers, Kroxylicious will bind 9192 for
+bootstrap and 9193-9195 inclusive to allow the clients to connect to each broker.
+
+=== Cluster Network Address Config Providers
+
+The Cluster Network Address Config Provider controls how the virtual cluster is presented to the network. Two
+alternatives are supported: `PortPerBroker` and `SniRouting` which have different characteristics which make each
+suitable for different use-cases. They are described next.
+
+==== PortPerBroker scheme
+
+In the `PortPerBroker` scheme, Kroxylicious automatically opens a port for each virtual cluster's bootstrap and
+one port per broker of each target cluster.  So, if you have two virtual clusters, each targeting a Kafka Cluster
+of three brokers, Kroxylicious will bind eight ports in total.
+
+Kroxylicious monitors the broker topology of the target cluster at runtime. It will adjust the number of open ports
+dynamically.  If another broker is added to the Kafka Cluster, Kroxylicious will automatically open a port for it.
+Similarly, if a broker is removed from the Kafka Cluster, Kroxylicious will automatically close the port that was
+assigned to it.
+
+The `PortPerBroker` scheme can be used with either clear text or TLS downstream connections.
+
+[source, yaml]
+----
+clusterNetworkAddressConfigProvider:
+  type: PortPerBroker
+  config:
+    bootstrapAddress: mycluster.kafka.com:9192                                 # <1>
+    brokerAddressPattern: mybroker-$(nodeId).mycluster.kafka.com:$(portNumber) # <2>
+    brokerStartPort: 9193                                                      # <3>
+    numberOfBrokerPorts: 3                                                     # <4>
+    bindAddress: 192.168.0.1                                                   # <5>
+----
+<1> The hostname and port of the bootstrap that will be used by the Kafka Clients.
+<2> (Optional) The broker address pattern used to form the broker addresses.  If not defined, it defaults to the
+hostname part of the `bootstrapAddress` and the port number allocated to the broker.
+<3> (Optional) The starting number for broker port range. Defaults to the port of the `bootstrapAddress` plus 1.
+<4> (Optional) The _maximum_ number of brokers of ports that are permitted.  Defaults to 3.
+<5> (Optional) The bind address used when binding the ports. If undefined, all network interfaces will be bound.
+
+The `brokerAddressPattern` configuration parameters understands two replacement tokens.
+
+* this `$(portNumber)` is mandatory.  It is replaced by the port number computed for this broker.
+* the `$(nodeId)` token is optional.  If present, it will be replaced by the
+https://kafka.apache.org/documentation/#brokerconfigs_node.id[`node.id`] (or `broker.id`) assigned to the broker
+of the target cluster.
+
+For example if your configuration looks like the above and your cluster has three brokers, your Kafka Client will receive
+broker address information like this:
+
+0.  `mybroker-0.mycluster.kafka.com:9193`
+1.  `mybroker-1.mycluster.kafka.com:9194`
+2.  `mybroker-2.mycluster.kafka.com:9194`
+
+NOTE: It is a responsibility for the deployer of Kroxylicious to ensure that the bootstrap and generated broker
+DNS names are resolvable and routable by the Kafka Client.
+
+The `numberOfBrokerPorts` imposes a maximum on the number of brokers that a Kafka Cluster can have. Set this value
+to be as high as the maximum number of brokers that your operational rules allow for a Kafka Cluster.
+
+==== SniRouting scheme
+
+In the `SniRouting` scheme, Kroxylicious uses SNI information to route traffic to either the boostrap or individual
+brokers.   As this relies on SNI (Server Name Indication), the use of <<Downstream TLS>> is *required*.
+
+With this scheme, you have the choice to share a single port across all virtual clusters, or you can assign a different
+port to each.  Single port operation may have cost advantages when using load balancers of public clouds, as it allows
+a single cloud provider load balancer to be shared across all virtual clusters.
+
+[source, yaml]
+----
+clusterNetworkAddressConfigProvider:
+  type: SniRouting
+  config:
+    bootstrapAddress: mycluster.kafka.com:9192                    # <1>
+    brokerAddressPattern: mybroker-$(nodeId).mycluster.kafka.com  # <2>
+    bindAddress: 192.168.0.1                                      # <3>
+----
+<1> The hostname and port of the bootstrap that will be used by the Kafka Clients.
+<2> The broker address pattern used to form the broker addresses. The `$(nodeId)` token must be present.
+<3> (Optional) The bind address used when binding the port. If undefined, all network interfaces will be bound.
+
+The `brokerAddressPattern` configuration parameters requires that the `$(nodeId)` token is present within it.
+This is replaced by the https://kafka.apache.org/documentation/#brokerconfigs_node.id[`node.id`] (or `broker.id)
+assigned to the broker of the target cluster.  This allows Kroxylicious to generate separate routes for each broker.
+
+NOTE: It is a responsibility for the deployer of Kroxylicious to ensure that the bootstrap and generated broker
+DNS names are resolvable and routable by the Kafka Client.
+
+=== Transport Layer Security (TLS)
+
+In this section we look at how to enable TLS for either the downstream and/or upstream.   Note, there is no
+interdependency; it is supported to have TLS configured for the downstream and use clear text communications for the
+upstream, or vice-versa.
+
+NOTE: TLS is recommended for both upstream and downstream for production configurations.
+
+=== Downstream TLS
+
+Here's how to enable TLS for the downstream side. This means the Kafka Client will connect to the virtual cluster over
+TLS rather than clear text.  For this, you will need to obtain a TLS certificate for the virtual cluster from your
+Certificate Authority.
+
+NOTE: When requesting the certificate ensure that the certificate will match the names of the virtual cluster's
+bootstrap and broker addresses.  This may mean making use of wildcard certificates and/or Subject Alternative Names (SANs).
+
+Kroxylicious accepts key material in PKCS12 or JKS keystore format, or PEM formatted file(s).  The following configuration
+illustrates configuration with PKCS12 keystore.
+
+[source, yaml]
+----
+virtualClusters:
+  demo:
+    tls:
+        key:
+          storeFile: /opt/cert/server.p12               # <1>
+          storePassword:
+            passwordFile: /opt/cert/store.password      # <2>
+          keyPassword:
+            passwordFile: /opt/cert/key.password        # <3>
+          storeType: PKCS12                             # <4>
+    clusterNetworkAddressConfigProvider:
+      ...
+----
+<1> File system location of a keystore (or in the case of `PEM` format a text file containing the concatenation of the
+private key, certificate, and intermediates).
+<2> File system location of a file containing the key store's password.
+<3> (Optional) File system location of a file containing the key's password. If omitted the key store's password is
+used to decrypt the key too.
+<4> (Optional) Store type. Supported types are: `PKCS12`, `JKS` and `PEM`.  Defaults to Java default key store type
+which is usually `PKCS12`.
+
+Alternatively, if your key material is in separate PEM files (private key, and certificate/intermediates), the following
+configuration may be used:
+
+[source, yaml]
+----
+virtualClusters:
+  demo:
+    tls:
+        key:
+          privateKeyFile: /opt/cert/server.key          # <1>
+          certificateFile: /opt/cert/server.crt         # <2>
+          keyPassword:
+            passwordFile: /opt/cert/key.password        # <3>
+    clusterNetworkAddressConfigProvider:
+      ...
+----
+<1> File system location of the server private key.
+<2> File system location of the server certificate and intermediate(s).
+<3> (Optional) File system location of a file containing the key's password.
+
+NOTE: For the private-key, https://datatracker.ietf.org/doc/html/rfc5208[PKCS-8 keys] are supported by default.
+For https://datatracker.ietf.org/doc/html/rfc8017[PKCS-1 keys], https://www.bouncycastle.org/[Bouncycastle] libraries
+must be added to the Kroxylicious classpath. See https://github.com/netty/netty/issues/7323 for more details.
+
+=== Upstream TLS
+
+Here's how to enable TLS for the upstream side.  This means that Kroxylicious connects to the (physical) Kafka Cluster)
+over TLS.  For this, your Kafka Cluster must have already been configured to use TLS.
+
+By default, Kroxylicious inherits what it trusts from the platform it is running on and uses this to determine whether
+the Kafka Cluster is trusted or not.
+
+To support cases where trust must be overridden (such as use-cases involving the use of private CAs or self-signed
+certificates), Kroxylicious accepts override trust material in PKCS12 or JKS keystore format, or PEM formatted
+certificates.
+
+The following illustrates enabling TLS, inheriting platform trust:
+
+[source, yaml]
+----
+virtualClusters:
+  demo:
+    targetCluster:
+      bootstrap_servers: myprivatecluster:9092
+      tls: {}                                         <1>
+      #...
+----
+<1> Use an empty object to enable TLS inheriting trust from the platform.
+
+The following illustrates enabling TLS but with trust coming from a PKCS12 trust store instead of the platform:
+
+[source, yaml]
+----
+virtualClusters:
+  demo:
+    targetCluster:
+      bootstrap_servers: myprivatecluster:9092
+      tls:
+        trust:
+          storeFile: /opt/cert/trust.p12                # <1>
+          storePassword:
+            passwordFile: /opt/cert/store.password      # <2>
+          storeType: PKCS12                             # <3>
+      #...
+----
+<1> File system location of a truststore (or in the case of `PEM` format a text file containing the certificates).
+<2> File system location of a file containing the trust store's password.
+<3> (Optional) Trust store type. Supported types are: `PKCS12`, `JKS` and `PEM`.  Defaults to Java default key store type (PKCS12).
+
+It is also possible to disable trust so that Kroxylicious will connect to any Kafka Cluster regardless of its certificate
+validity.
+
+WARNING: This option is not recommended for production use.
+
+[source, yaml]
+----
+virtualClusters:
+  demo:
+    targetCluster:
+      bootstrap_servers: myprivatecluster:9092
+      tls:
+        trust:
+          insecure: true                                # <1>
+      #...
+----
+<1> Enables insecure TLS.
 
 YAML
 Proxy level configuration
-SSL
 
 == Configuring proxy plugins
 Filter level configuration
\ No newline at end of file
diff --git a/docs/overview.adoc b/docs/overview.adoc
index 0d8185c0..e323f5b3 100644
--- a/docs/overview.adoc
+++ b/docs/overview.adoc
@@ -2,12 +2,12 @@
 
 == What is Kroxylicious Proxy?
 
-Kroxylicous Proxy provides a pluggable, protocol-aware ("Layer 7") proxy for https://kafka.apache.org[Apache Kafka(R)] brokers and clusters, together with an API for conveniently implementing custom logic within such a proxy.
+Kroxylicious Proxy provides a pluggable, protocol-aware ("Layer 7") proxy for https://kafka.apache.org[Apache Kafka(R)] brokers and clusters, together with an API for conveniently implementing custom logic within such a proxy.
 
 == Why?
 
 Proxies are a powerful and flexible architectural pattern.
-For Kafka they can be used to add functionality to Kafka clusters which is not available out-of-the-box with Apache Kafka.
+For Kafka, they can be used to add functionality to Kafka clusters which is not available out-of-the-box with Apache Kafka.
 In an ideal world, such functionality would be implemented directly in Apache Kafka.
 But there are numerous practical reasons that can prevent this, for example:
 
@@ -19,25 +19,187 @@ For example using Kroxylicious proxy it's easier to experiment with alternative
 
 == How it works
 
-// TODO include a diagram
+First let's define the concepts in the landscape surrounding Kroxylicious.
+
+. _Kafka Client_, or _Client_ refers to any client application using a Kafka Client library to talk to a *Kafka Cluster*.
+. _Kafka Cluster_ or _Cluster_ refers to a cluster comprising one or more Kafka Brokers.
+. _Downstream_ refers to the area between Kafka Client and Kroxylicious.
+. _Upstream_ refers to the area between Kroxylicious and a Kafka Cluster.
+
+.Kroxylious landscape
+[a2s, format="svg"]
+....
+.---------------------------------------------------------------------------.
+|                                  :                                        |
+|                                  :              +===============+         |
+|                                  :              :[Cluster]      :         |
+|                         .-----------------.     :  .----------. :         |
+|     .----------.        |[Kroxylicious]   |     :  |[Broker1] | :         |
+|     |[Client]  |        |                 +------->|          | :         |
+|     |          +------->|                 |     :  |          | :         |
+|     |          |        |                 |     :  .----------. :         |
+|     |          |        |                 |     :               :         |
+|     .----------.        |                 |     :  .----------. :         |
+|                         |                 |     :  |[Broker2] | :         |
+|                         |                 +------->|          | :         |
+|                         |                 |     :  |          | :         |
+|                         |                 |     :  .----------. :         |
+|     .----------.        |                 |     :               :         |
+|     |[Client]  |        |                 |     :  .----------. :         |
+|     |          |        |                 |     :  |[Broker3] | :         |
+|     |          +------->|                 +------->|          | :         |
+|     |          |        |                 |     :  |          | :         |
+|     .----------.        |                 |     :  .----------. :         |
+|                         .-----------------.     .===============+         |
+|                                  :                                        |
+|                                  :                                        |
+|     [downstream]                 :                  [upstream]            |
+.---------------------------------------------------------------------------.
+[0,0]: {"fill":"#99d","a2s:delref":1}
+....
+
+Now let's define some concepts used within Kroxylicious itself.
+
+=== Virtual Cluster
+
+The _Virtual Cluster_ is the downstream representation of a Kafka Cluster.  At the conceptual level, a Kafka Client
+connects to a Virtual Cluster.  Kroxylicious proxies all communications made to the Virtual Cluster through to a
+(physical) Kafka Cluster, passing it through the _Filter Chain_.
+
+So far, this explanation has elided the detail of Kafka Brokers.  Let's talk about that now.
+
+The Virtual Cluster automatically exposes a bootstrap endpoint for the Virtual Cluster.  This is what the Kafka Client
+must specify as the https://kafka.apache.org/documentation/#producerconfigs_bootstrap.servers[bootstrap.servers] property
+in the client configuration.
+
+In addition to the bootstrap endpoint, Kroxylicious automatically exposes broker endpoints.  There is one broker endpoint
+for each broker of the physical cluster.  When the Client connects to a broker endpoint, Kroxylicious proxies all
+communications to the corresponding broker of the (physical) Kafka Cluster.
+
+Kroxylicious automatically intercepts all the Kafka RPC responses that contain a broker address.  It rewrites the address
+so that it refers to the corresponding broker endpoint of the Virtual Cluster.  This means when the Kafka Client
+goes to connect to, say broker 0, it does so through the Virtual Cluster.
+
+=== Target Cluster
+
+The _Target Cluster_ is the definition of physical Kafka Cluster within the Kroxylicious itself.
 
-When a Kafka protocol client (typically a producer, consumer or admin client) connects to a proxy instance it is first handled by a pluggable _network filter_
-This handles things like:
+A _Virtual Cluster_ has exactly one _Target Cluster_.
 
-* SSL
-* Possibly initial client requests such as `ApiVersions`, `SaslHandshake` and `SaslAuthenticate`
+There can be a _one-to-one_ relationship between Virtual Clusters and Target Clusters.
+The other possibility is _many-to-one_, where many Virtual Clusters point to the same Target Cluster.  The
+many-to-one pattern is exploited by filters such as the xref:available-filters.adoc#_multi_tenancy[Multi-tenancy]
+filter.
 
-The network filter then
+.One-to-One relationship between Virtual Cluster and Target Cluster
+[a2s, format="svg"]
+....
+.-------------------------------------------------------------------------------------.
+|                                                                                     |
+|                 '-------------------------------------'    '===================='   |
+| '---------'     |[Kroxylicious]                       |    :[Cluster]           :   |
+| |[Client] |=+   |                                     |    :                    :   |
+| '---------' :   |    '----------------------------'   |    :      '---------'   :   |
+|             :   |    |[Virtual Cluster]           |===+====|=====>|[Broker] |   :   |
+|             +===|===>|                            |   |    :      |         |   :   |
+|             :   |    |                            |   |    :      '---------'   :   |
+|             +===|===>|                            |   |    :                    :   |
+|                 |    |                            |   |    :      '---------'   :   |
+|                 |    |                            |===+====:==+==>|[Broker] |   :   |
+|                 |    '----------------------------'   |    :      |         |   :   |
+|                 |                                     |    :      '---------'   :   |
+|                 '-------------------------------------'    '===================='   |
+.-------------------------------------------------------------------------------------.
+[0,0]: {"fill":"#99d","a2s:delref":1}
+....
 
-. determines which broker to connect the client to
-. instantiates a chain of protocol filters for processing all further messages.
-. removes itself from the further processing
+.Many-to-one between Virtual Cluster and Target Cluster
+[a2s, format="svg"]
+....
+.-------------------------------------------------------------------------------------.
+|                                                                                     |
+|                 '-------------------------------------'    '===================='   |
+| '---------'     |[Kroxylicious]                       |    :[Cluster]           :   |
+| |[Client] |=+   |                                     |    :                    :   |
+| '---------' :   |    '----------------------------'   |    :      '---------'   :   |
+|             :   |    |[Virtual Cluster A]         |===+=+==|==+==>|[Broker] |   :   |
+|             +===|===>|                            |   | :  :      |         |   :   |
+|             :   |    |                            |   | :  :      '---------'   :   |
+|             +===|===>|                            |   | :  :                    :   |
+|                 |    |                            |   | :  :      '---------'   :   |
+|                 |    |                            |===+====:==+==>|[Broker] |   :   |
+|                 |    '----------------------------'   | :  :  :   |         |   :   |
+| '---------'     |                                     | :  :  :   '---------'   :   |
+| |[Client] |=+   |    '----------------------------'   | :  :  :                 :   |
+| '---------' :   |    |[Virtual Cluster B]         |===+=+  '===================='   |
+|             +===|===>|                            |   |       :                     |
+|             :   |    |                            |   |       :                     |
+|             +===|===>|                            |   |       :                     |
+|                 |    |                            |   |       :                     |
+|                 |    |                            |===+=======+                     |
+|                 |    '----------------------------'   |                             |
+|                 '-------------------------------------'                             |
+.-------------------------------------------------------------------------------------.
+[0,0]: {"fill":"#99d","a2s:delref":1}
+....
+
+A one-to-many pattern, where one Virtual Cluster points to many Target Clusters (providing amalgamation),
+is not a supported use-case.
+
+
+
+=== Filter Chain
+
+A _Filter Chain_ consists of an *ordered list* of pluggable _protocol filters_.
 
-A _filter chain_ consists of one or more pluggable _protocol filters_.
 A  _protocol filter_ implements some logic for intercepting, inspecting and/or manipulating Kafka protocol messages.
-Kafka protocol requests (such as `Produce` requests) pass sequentially through each of the protocol filters in the chain before being forwarded to the broker.
+Kafka protocol requests (such as `Produce` requests) pass sequentially through each of the protocol filters in the
+chain, beginning with the 1st filter in the chain and then following with the subsequent filters, before being
+forwarded to the broker.
+
+When the broker returns a response (such as a `Produce` response) the protocol filters in the chain are invoked in the
+reverse order (that is, beginning with the nth filter in the chain, then the n-1th and so on) with each having the
+opportunity to inspect and/or manipulating the response. Eventually a response is returned to the client.
 
-When the broker returns a response (such as a `Produce` response) the protocol filters in the chain are invoked in the reverse order, each having the opportunity to inspecting and/or manipulating the response. Eventually a response is returned to the client.
+The description above describes only the basic capabilities of the protocol filter. Richer features of filters
+are described later.
+
+// TODO document additional filter behaviours https://github.com/kroxylicious/kroxylicious/issues/420
+
+.Illustration of a request and response being manipulated by filters in a chain
+[a2s, format="svg"]
+....
+.----------------------------------------------------------------------------------------------------------------------.
+|                                                                                                                      |
+|                       '---------------------------------------------------------------'                              |
+|                       |[Kroxylicious]                                                 |                              |
+|                       |                                                               |                              |
+|                       |   '----------------------------------------------------'      |      '--------------------'  |
+|                       |   |[Virtual Cluster]                                   |      |      |[Cluster]           |  |
+|  '-------------'      |   |   '----------'     '----------'     '----------'   |      |      |    '------------'  |  |
+|  |[Client]     |      |   |   |[Filter1] |     |[Filter2] |     |[Filter3] |   |      |      |    |[Broker]    |  |  |
+|  |             |======|===|==>|          |====>|          |====>|          |===|======|======|===>|            |  |  |
+|  |             |  A   |   |   | F(A)-->B |  B  | F(B)-->C |  C  | F(C)-->D |   |      |      | D  |            |  |  |
+|  |             |      |   |   |          |     |          |     |          |   |      |      |    |            |  |  |
+|  |             |<=====|===|===|          |<====|          |<====|          |<==|======|======|====|            |  |  |
+|  |             |  W   |   |   | f(X)-->W |  X  | f(Y)-->X |  Y  | f(Z)-->Y |   |      |      | Z  |            |  |  |
+|  '-------------'      |   |   '----------'     '----------'     '----------'   |      |      |    '------------'  |  |
+|                       |   |                                                    |      |      '--------------------'  |
+|                       |   '----------------------------------------------------'      |                              |
+|                       |                                                               |                              |
+|                       '---------------------------------------------------------------'                              |
+|                                                                                                                      |
+.----------------------------------------------------------------------------------------------------------------------.
+[0,0]: {"fill":"#99d","a2s:delref":1}
+....
+
+As mentioned above, Kroxylicious takes the responsibility to rewrite the Kafka RPC responses that carry broker address
+information so that they reflect the broker addresses exposed by the Virtual Cluster. These are the
+https://kafka.apache.org/protocol.html#The_Messages_Metadata[`Metadata`],
+https://kafka.apache.org/protocol.html#The_Messages_DescribeCluster[`DescribeCluster`] and
+https://kafka.apache.org/protocol.html#The_Messages_FindCoordinator[`FindCoordinator`] responses. This processing is
+entirely transparent to the work of the protocol filters.  _Filter authors_ are free to write their own filters that
+intercept these responses too.
 
 === Filter composition
 
@@ -89,17 +251,9 @@ _Transparent_ multi-tenancy means each tenant has the illusion of having their o
 
 We can further classify deployment topologies in how many proxy instances are used. For example:
 
-Proxy-per-broker::
-This is probably the simplest way to deploy a proxy.
-+
-If you control the Kafka cluster you can use the https://kafka.apache.org/documentation.html#brokerconfigs_advertised.listeners[`advertised.listeners`] broker config to ensure all access to the cluster happens via the proxy.
-When the Kafka cluster is running on Kubernetes proxy-per-broker can be achived using the side-car pattern.
-+
-If you do not control the Kafka cluster you can use a protocol filter to rewrite  `Metadata` (and a few other responses) sent to clients so that they only discover proxy-fronted brokers, and therefore only connect via proxies.
-
 Single proxy instance::
 
-Proxy cluster::
+Proxy pool::
 //-
 
 == More about filters
diff --git a/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml b/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml
index 23ad9fb8..eaeee8cd 100644
--- a/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml
+++ b/kroxylicious-additional-filters/kroxylicious-multitenant/pom.xml
@@ -26,20 +26,27 @@
     </description>
 
     <dependencies>
+        <!-- project dependencies - runtime and compile -->
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-api</artifactId>
-            <version>${kroxyliciousApi.version}</version>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-filter-api</artifactId>
-            <version>${kroxyliciousApi.version}</version>
         </dependency>
-        <!-- Test dependencies -->
+
+        <!-- project dependencies - test -->
         <dependency>
-            <groupId>com.google.guava</groupId>
-            <artifactId>guava</artifactId>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-unit-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <!-- third party dependencies - test -->
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
@@ -48,8 +55,8 @@
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-databind</artifactId>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
@@ -58,18 +65,18 @@
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter</artifactId>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter-params</artifactId>
+            <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.assertj</groupId>
-            <artifactId>assertj-core</artifactId>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
@@ -82,12 +89,6 @@
             <artifactId>zjsonpatch</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>io.kroxylicious</groupId>
-            <artifactId>kroxylicious-unit-test</artifactId>
-            <version>0.3.0-SNAPSHOT</version>
-            <scope>test</scope>
-        </dependency>
     </dependencies>
     <build>
         <plugins>
diff --git a/kroxylicious-additional-filters/kroxylicious-schema-validation/pom.xml b/kroxylicious-additional-filters/kroxylicious-schema-validation/pom.xml
index 5bd5219a..38553cc5 100644
--- a/kroxylicious-additional-filters/kroxylicious-schema-validation/pom.xml
+++ b/kroxylicious-additional-filters/kroxylicious-schema-validation/pom.xml
@@ -26,29 +26,36 @@
     </description>
 
     <dependencies>
+        <!-- project dependencies - runtime and compile -->
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-api</artifactId>
-            <version>${kroxyliciousApi.version}</version>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-filter-api</artifactId>
-            <version>${kroxyliciousApi.version}</version>
         </dependency>
+
+        <!-- third party dependencies - runtime and compile -->
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
         </dependency>
-        <!-- Test dependencies -->
+
+        <!-- third party dependencies - test -->
         <dependency>
-            <groupId>com.google.guava</groupId>
-            <artifactId>guava</artifactId>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>com.fasterxml.jackson.dataformat</groupId>
-            <artifactId>jackson-dataformat-yaml</artifactId>
+            <groupId>com.flipkart.zjsonpatch</groupId>
+            <artifactId>zjsonpatch</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
@@ -57,18 +64,18 @@
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter</artifactId>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter-params</artifactId>
+            <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.assertj</groupId>
-            <artifactId>assertj-core</artifactId>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
@@ -76,10 +83,5 @@
             <artifactId>mockito-core</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>com.flipkart.zjsonpatch</groupId>
-            <artifactId>zjsonpatch</artifactId>
-            <scope>test</scope>
-        </dependency>
     </dependencies>
 </project>
diff --git a/kroxylicious-sample/pom.xml b/kroxylicious-sample/pom.xml
index 34dab58f..f2c797d0 100644
--- a/kroxylicious-sample/pom.xml
+++ b/kroxylicious-sample/pom.xml
@@ -25,38 +25,42 @@
     <name>Kroxylicious Sample</name>
 
     <dependencies>
+        <!-- project dependencies - runtime and compile -->
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-api</artifactId>
-            <version>${kroxyliciousApi.version}</version>
             <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-filter-api</artifactId>
-            <version>${kroxyliciousApi.version}</version>
             <scope>provided</scope>
         </dependency>
-        <dependency>
-            <groupId>io.kroxylicious.testing</groupId>
-            <artifactId>testing-junit5-extension</artifactId>
-            <scope>test</scope>
-        </dependency>
+
+        <!-- project dependencies - test -->
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-test-tools</artifactId>
             <scope>test</scope>
         </dependency>
+
         <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter-api</artifactId>
+            <groupId>io.kroxylicious.testing</groupId>
+            <artifactId>testing-junit5-extension</artifactId>
             <scope>test</scope>
         </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-api</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>
diff --git a/kroxylicious/pom.xml b/kroxylicious/pom.xml
index 667e3816..948a1857 100644
--- a/kroxylicious/pom.xml
+++ b/kroxylicious/pom.xml
@@ -23,48 +23,94 @@
 
     <properties>
         <libs.dir>libs</libs.dir>
-        <log4j.version>2.20.0</log4j.version>
     </properties>
 
     <name>Kroxylicious Proxy</name>
 
     <dependencies>
+        <!-- project dependencies - runtime and compile -->
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-api</artifactId>
-            <version>${kroxyliciousApi.version}</version>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-filter-api</artifactId>
-            <version>${kroxyliciousApi.version}</version>
         </dependency>
-        <!-- We depend on the specific Netty dependencies not netty-all to reduce the size of fatjars -->
+
+        <!-- project dependencies - test -->
         <dependency>
-            <groupId>io.netty</groupId>
-            <artifactId>netty-common</artifactId>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-unit-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.kroxylicious.testing</groupId>
+            <artifactId>testing-junit5-extension</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.kroxylicious.testing</groupId>
+            <artifactId>testing-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+
+        <!-- third party dependencies - runtime and compile -->
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.module</groupId>
+            <artifactId>jackson-module-parameter-names</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jdk8</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.google.code.findbugs</groupId>
+            <artifactId>annotations</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>info.picocli</groupId>
+            <artifactId>picocli</artifactId>
         </dependency>
         <dependency>
             <groupId>io.micrometer</groupId>
             <artifactId>micrometer-registry-prometheus</artifactId>
         </dependency>
+        <!-- We depend on the specific Netty dependencies not netty-all to reduce the size of fatjars -->
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-buffer</artifactId>
         </dependency>
         <dependency>
             <groupId>io.netty</groupId>
-            <artifactId>netty-codec-http</artifactId>
-            <version>${netty.version}</version>
+            <artifactId>netty-common</artifactId>
         </dependency>
         <dependency>
             <groupId>io.netty</groupId>
-            <artifactId>netty-transport</artifactId>
+            <artifactId>netty-codec-haproxy</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-codec-http</artifactId>
         </dependency>
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-handler</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-transport</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-transport-native-epoll</artifactId>
@@ -83,20 +129,14 @@
             <groupId>io.netty</groupId>
             <artifactId>netty-transport-classes-kqueue</artifactId>
         </dependency>
-        <dependency>
-            <groupId>io.netty.incubator</groupId>
-            <artifactId>netty-incubator-transport-native-io_uring</artifactId>
-            <version>${netty.io_uring.version}</version>
-            <classifier>${netty.io_uring.classifier}</classifier>
-        </dependency>
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-transport-native-unix-common</artifactId>
         </dependency>
         <dependency>
-            <groupId>io.netty</groupId>
-            <artifactId>netty-codec-haproxy</artifactId>
-            <version>${netty.version}</version>
+            <groupId>io.netty.incubator</groupId>
+            <artifactId>netty-incubator-transport-native-io_uring</artifactId>
+            <classifier>${netty.io_uring.classifier}</classifier>
         </dependency>
         <dependency>
             <groupId>org.apache.kafka</groupId>
@@ -106,40 +146,8 @@
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
         </dependency>
-        <dependency>
-            <groupId>info.picocli</groupId>
-            <artifactId>picocli</artifactId>
-            <version>${picocli.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-databind</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>com.fasterxml.jackson.dataformat</groupId>
-            <artifactId>jackson-dataformat-yaml</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>com.fasterxml.jackson.module</groupId>
-            <artifactId>jackson-module-parameter-names</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>com.fasterxml.jackson.datatype</groupId>
-            <artifactId>jackson-datatype-jdk8</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>com.google.code.findbugs</groupId>
-            <artifactId>annotations</artifactId>
-        </dependency>
 
-        <!-- Test dependencies -->
-        <dependency>
-            <groupId>io.kroxylicious</groupId>
-            <artifactId>kroxylicious-unit-test</artifactId>
-            <version>0.3.0-SNAPSHOT</version>
-            <scope>test</scope>
-        </dependency>
+        <!-- third party dependencies - test -->
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java
index bcfcfde4..874815e5 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java
@@ -8,7 +8,7 @@ package io.kroxylicious.proxy.bootstrap;
 import java.util.List;
 
 import io.kroxylicious.proxy.config.Configuration;
-import io.kroxylicious.proxy.filter.KrpcFilter;
+import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.internal.filter.FilterContributorManager;
 
 /**
@@ -29,12 +29,13 @@ public class FilterChainFactory {
      *
      * @return the new chain.
      */
-    public List<KrpcFilter> createFilters() {
+    public List<FilterAndInvoker> createFilters() {
         FilterContributorManager filterContributorManager = FilterContributorManager.getInstance();
 
         return config.filters()
                 .stream()
                 .map(f -> filterContributorManager.getFilter(f.type(), f.config()))
+                .flatMap(filter -> FilterAndInvoker.build(filter).stream())
                 .toList();
     }
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/NetFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/NetFilter.java
index 41eb2091..360b8b06 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/NetFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/filter/NetFilter.java
@@ -80,7 +80,7 @@ public interface NetFilter {
          * @param target upstream broker target
          * @param filters The filters
          */
-        void initiateConnect(HostPort target, List<KrpcFilter> filters);
+        void initiateConnect(HostPort target, List<FilterAndInvoker> filters);
 
         // TODO add API for delayed responses
     }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/DecodedFrame.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/DecodedFrame.java
index fc4e8465..6b6406d4 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/DecodedFrame.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/DecodedFrame.java
@@ -135,4 +135,9 @@ public abstract class DecodedFrame<H extends ApiMessage, B extends ApiMessage>
     protected void deallocate() {
         buffers.forEach(ByteBuf::release);
     }
+
+    public void transferBuffersTo(DecodedFrame<?, ?> frame) {
+        frame.buffers.addAll(this.buffers);
+        this.buffers.clear();
+    }
 }
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 434fd9ff..dd496a2e 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
@@ -223,6 +223,7 @@ class DefaultFilterContext implements KrpcFilterContext {
                     "Attempt to respond with ApiMessage of type " + ApiKeys.forId(response.apiKey()) + " but request is of type " + decodedFrame.apiKey());
         }
         DecodedResponseFrame<?> responseFrame = new DecodedResponseFrame<>(decodedFrame.apiVersion(), decodedFrame.correlationId(), header, response);
+        decodedFrame.transferBuffersTo(responseFrame);
         if (LOGGER.isDebugEnabled()) {
             LOGGER.debug("{}: Forwarding response: {}", channelDescriptor(), decodedFrame);
         }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
index 82b2ea3c..8c246fec 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
@@ -17,8 +17,8 @@ import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 
+import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.filter.FilterInvoker;
-import io.kroxylicious.proxy.filter.FilterInvokers;
 import io.kroxylicious.proxy.filter.KrpcFilter;
 import io.kroxylicious.proxy.frame.DecodedRequestFrame;
 import io.kroxylicious.proxy.frame.DecodedResponseFrame;
@@ -39,9 +39,9 @@ public class FilterHandler
     private final String sniHostname;
     private final FilterInvoker invoker;
 
-    public FilterHandler(KrpcFilter filter, long timeoutMs, String sniHostname) {
-        this.filter = Objects.requireNonNull(filter);
-        this.invoker = FilterInvokers.from(filter);
+    public FilterHandler(FilterAndInvoker filterAndInvoker, long timeoutMs, String sniHostname) {
+        this.filter = Objects.requireNonNull(filterAndInvoker).filter();
+        this.invoker = filterAndInvoker.invoker();
         this.timeoutMs = Assertions.requireStrictlyPositive(timeoutMs, "timeout");
         this.sniHostname = sniHostname;
     }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
index 4bf6e6c0..7df75bae 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
@@ -34,7 +34,7 @@ import io.netty.handler.codec.haproxy.HAProxyMessage;
 import io.netty.handler.logging.LoggingHandler;
 import io.netty.handler.ssl.SniCompletionEvent;
 
-import io.kroxylicious.proxy.filter.KrpcFilter;
+import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.filter.NetFilter;
 import io.kroxylicious.proxy.frame.DecodedRequestFrame;
 import io.kroxylicious.proxy.frame.DecodedResponseFrame;
@@ -224,7 +224,7 @@ public class KafkaProxyFrontendHandler
     }
 
     @Override
-    public void initiateConnect(HostPort remote, List<KrpcFilter> filters) {
+    public void initiateConnect(HostPort remote, List<FilterAndInvoker> filters) {
         if (backendHandler != null) {
             throw new IllegalStateException();
         }
@@ -284,7 +284,7 @@ public class KafkaProxyFrontendHandler
         return b.connect(remoteHost, remotePort);
     }
 
-    private void addFiltersToPipeline(List<KrpcFilter> filters, ChannelPipeline pipeline) {
+    private void addFiltersToPipeline(List<FilterAndInvoker> filters, ChannelPipeline pipeline) {
         for (var filter : filters) {
             // TODO configurable timeout
             pipeline.addFirst(filter.toString(), new FilterHandler(filter, 20000, sniHostname));
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 0e806a2b..be380e46 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
@@ -27,6 +27,7 @@ import io.netty.util.concurrent.Future;
 
 import io.kroxylicious.proxy.bootstrap.FilterChainFactory;
 import io.kroxylicious.proxy.config.Configuration;
+import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.internal.codec.KafkaRequestDecoder;
 import io.kroxylicious.proxy.internal.codec.KafkaResponseEncoder;
 import io.kroxylicious.proxy.internal.filter.BrokerAddressFilter;
@@ -171,7 +172,7 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
 
             var filters = new ArrayList<>(filterChainFactory.createFilters());
             // Add internal filters.
-            filters.add(new BrokerAddressFilter(virtualCluster, endpointReconciler));
+            filters.addAll(FilterAndInvoker.build(new BrokerAddressFilter(virtualCluster, endpointReconciler)));
 
             var target = binding.upstreamTarget();
             if (target == null) {
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/DecodePredicate.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/DecodePredicate.java
index 470427d7..5c448cd6 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/DecodePredicate.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/DecodePredicate.java
@@ -9,9 +9,8 @@ import java.util.List;
 
 import org.apache.kafka.common.protocol.ApiKeys;
 
+import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.filter.FilterInvoker;
-import io.kroxylicious.proxy.filter.FilterInvokers;
-import io.kroxylicious.proxy.filter.KrpcFilter;
 
 /**
  * Encapsulates decisions about whether requests and responses should be
@@ -22,9 +21,9 @@ import io.kroxylicious.proxy.filter.KrpcFilter;
  * who the authorized user or, or which back-end cluster they're connected to.
  */
 public interface DecodePredicate {
-    static DecodePredicate forFilters(List<KrpcFilter> filters) {
+    static DecodePredicate forFilters(List<FilterAndInvoker> filterAndInvokers) {
 
-        List<FilterInvoker> invokers = filters.stream().map(FilterInvokers::from).toList();
+        List<FilterInvoker> invokers = filterAndInvokers.stream().map(FilterAndInvoker::invoker).toList();
         return new DecodePredicate() {
             @Override
             public boolean shouldDecodeResponse(ApiKeys apiKey, short apiVersion) {
@@ -48,7 +47,7 @@ public interface DecodePredicate {
 
             @Override
             public String toString() {
-                return "DecodePredicate$forFilters{" + filters + "}";
+                return "DecodePredicate$forFilters{" + filterAndInvokers + "}";
             }
         };
     }
diff --git a/krpc-code-gen/pom.xml b/krpc-code-gen/pom.xml
index c3771a96..8ca5bbe9 100644
--- a/krpc-code-gen/pom.xml
+++ b/krpc-code-gen/pom.xml
@@ -23,13 +23,14 @@
     <packaging>maven-plugin</packaging>
 
     <properties>
+        <maven-plugin-api.version>3.9.2</maven-plugin-api.version>
+        <maven-plugin-annotations.version>3.9.0</maven-plugin-annotations.version>
+        <maven-core.version>3.9.2</maven-core.version>
+        <plexus-build-api.version>1.2.0</plexus-build-api.version>
     </properties>
 
     <dependencies>
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-clients</artifactId>
-        </dependency>
+        <!-- third party dependencies - runtime and compile -->
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-annotations</artifactId>
@@ -42,45 +43,48 @@
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.freemarker</groupId>
             <artifactId>freemarker</artifactId>
         </dependency>
-
         <dependency>
             <groupId>org.apache.maven</groupId>
-            <artifactId>maven-plugin-api</artifactId>
-            <version>3.9.2</version>
+            <artifactId>maven-core</artifactId>
+            <version>${maven-core.version}</version>
             <scope>provided</scope>
         </dependency>
         <dependency>
-            <groupId>org.apache.maven.plugin-tools</groupId>
-            <artifactId>maven-plugin-annotations</artifactId>
-            <version>3.9.0</version>
+            <groupId>org.apache.maven</groupId>
+            <artifactId>maven-plugin-api</artifactId>
+            <version>${maven-plugin-api.version}</version>
             <scope>provided</scope>
         </dependency>
         <dependency>
-            <groupId>org.apache.maven</groupId>
-            <artifactId>maven-core</artifactId>
-            <version>3.9.2</version>
+            <groupId>org.apache.maven.plugin-tools</groupId>
+            <artifactId>maven-plugin-annotations</artifactId>
+            <version>${maven-plugin-annotations.version}</version>
             <scope>provided</scope>
         </dependency>
         <dependency>
             <!-- according to https://github.com/codehaus-plexus/plexus-build-api this replaces org.sonatype.plexus:plexus-build-api -->
             <groupId>org.codehaus.plexus</groupId>
             <artifactId>plexus-build-api</artifactId>
-            <version>1.2.0</version>
+            <version>${plexus-build-api.version}</version>
         </dependency>
 
+        <!-- third party dependencies - test -->
         <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter</artifactId>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
-        <!-- https://mvnrepository.com/artifact/org.assertj/assertj-core -->
         <dependency>
-            <groupId>org.assertj</groupId>
-            <artifactId>assertj-core</artifactId>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
     </dependencies>
diff --git a/pom.xml b/pom.xml
index 1007a37f..b2ab7fda 100644
--- a/pom.xml
+++ b/pom.xml
@@ -108,15 +108,61 @@
 
     <dependencyManagement>
         <dependencies>
-            <!-- project dependencies -->
+            <!-- project dependencies - runtime and compile -->
+            <dependency>
+                <groupId>io.kroxylicious</groupId>
+                <artifactId>kroxylicious</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>io.kroxylicious</groupId>
+                <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>
+                <version>${project.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>io.kroxylicious</groupId>
+                <artifactId>kroxylicious-schema-validation</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+
+            <!-- project dependencies - test -->
             <dependency>
                 <groupId>io.kroxylicious</groupId>
                 <artifactId>kroxylicious-test-tools</artifactId>
                 <version>${project.version}</version>
                 <scope>test</scope>
             </dependency>
+            <dependency>
+                <groupId>io.kroxylicious</groupId>
+                <artifactId>kroxylicious-unit-test</artifactId>
+                <version>${project.version}</version>
+                <scope>test</scope>
+            </dependency>
 
-            <!-- third party dependencies -->
+            <dependency>
+                <groupId>io.kroxylicious.testing</groupId>
+                <artifactId>testing-junit5-extension</artifactId>
+                <version>${kroxy.extension.version}</version>
+                <scope>test</scope>
+            </dependency>
+            <dependency>
+                <groupId>io.kroxylicious.testing</groupId>
+                <artifactId>testing-api</artifactId>
+                <version>${kroxy.extension.version}</version>
+                <scope>test</scope>
+            </dependency>
+
+            <!-- third party dependencies - runtime and compile -->
             <dependency>
                 <groupId>com.fasterxml.jackson</groupId>
                 <artifactId>jackson-bom</artifactId>
@@ -131,11 +177,9 @@
                 <scope>provided</scope>
             </dependency>
             <dependency>
-                <groupId>io.netty</groupId>
-                <artifactId>netty-bom</artifactId>
-                <version>${netty.version}</version>
-                <type>pom</type>
-                <scope>import</scope>
+                <groupId>info.picocli</groupId>
+                <artifactId>picocli</artifactId>
+                <version>${picocli.version}</version>
             </dependency>
             <dependency>
                 <groupId>io.micrometer</groupId>
@@ -145,42 +189,41 @@
                 <scope>import</scope>
             </dependency>
             <dependency>
-                <groupId>org.apache.logging.log4j</groupId>
-                <artifactId>log4j-bom</artifactId>
-                <version>${log4j.version}</version>
-                <scope>import</scope>
+                <groupId>io.netty</groupId>
+                <artifactId>netty-bom</artifactId>
+                <version>${netty.version}</version>
                 <type>pom</type>
+                <scope>import</scope>
             </dependency>
             <dependency>
-                <groupId>org.junit</groupId>
-                <artifactId>junit-bom</artifactId>
-                <version>${junit.version}</version>
-                <type>pom</type>
-                <scope>import</scope>
+                <groupId>io.netty.incubator</groupId>
+                <artifactId>netty-incubator-transport-native-io_uring</artifactId>
+                <version>${netty.io_uring.version}</version>
+                <classifier>${netty.io_uring.classifier}</classifier>
             </dependency>
             <dependency>
-                <groupId>org.mockito</groupId>
-                <artifactId>mockito-bom</artifactId>
-                <version>${mockito.version}</version>
-                <type>pom</type>
-                <scope>import</scope>
+                <groupId>io.sundr</groupId>
+                <artifactId>builder-annotations</artifactId>
+                <version>${sundr-builder-annotations.version}</version>
+                <scope>provided</scope>
             </dependency>
             <dependency>
                 <groupId>org.apache.kafka</groupId>
                 <artifactId>kafka-clients</artifactId>
                 <version>${kafka.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.logging.log4j</groupId>
+                <artifactId>log4j-bom</artifactId>
+                <version>${log4j.version}</version>
+                <scope>import</scope>
+                <type>pom</type>
+            </dependency>
             <dependency>
                 <groupId>org.freemarker</groupId>
                 <artifactId>freemarker</artifactId>
                 <version>${freemarker.version}</version>
             </dependency>
-            <dependency>
-                <groupId>com.google.guava</groupId>
-                <artifactId>guava</artifactId>
-                <version>${guava.version}</version>
-                <scope>test</scope>
-            </dependency>
             <dependency>
                 <groupId>org.slf4j</groupId>
                 <artifactId>slf4j-api</artifactId>
@@ -191,35 +234,39 @@
                 <artifactId>slf4j-simple</artifactId>
                 <version>${slf4j.version}</version>
             </dependency>
+
+            <!-- third party dependencies - test -->
             <dependency>
-                <groupId>org.assertj</groupId>
-                <artifactId>assertj-core</artifactId>
-                <version>${assertj-core.version}</version>
+                <groupId>com.flipkart.zjsonpatch</groupId>
+                <artifactId>zjsonpatch</artifactId>
+                <version>${zjsonpatch.version}</version>
                 <scope>test</scope>
             </dependency>
             <dependency>
-                <groupId>io.kroxylicious.testing</groupId>
-                <artifactId>testing-junit5-extension</artifactId>
-                <version>${kroxy.extension.version}</version>
+                <groupId>com.google.guava</groupId>
+                <artifactId>guava</artifactId>
+                <version>${guava.version}</version>
                 <scope>test</scope>
             </dependency>
             <dependency>
-                <groupId>io.kroxylicious.testing</groupId>
-                <artifactId>testing-api</artifactId>
-                <version>${kroxy.extension.version}</version>
+                <groupId>org.assertj</groupId>
+                <artifactId>assertj-core</artifactId>
+                <version>${assertj-core.version}</version>
                 <scope>test</scope>
             </dependency>
             <dependency>
-                <groupId>com.flipkart.zjsonpatch</groupId>
-                <artifactId>zjsonpatch</artifactId>
-                <version>${zjsonpatch.version}</version>
-                <scope>test</scope>
+                <groupId>org.junit</groupId>
+                <artifactId>junit-bom</artifactId>
+                <version>${junit.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
             </dependency>
             <dependency>
-                <groupId>io.sundr</groupId>
-                <artifactId>builder-annotations</artifactId>
-                <version>${sundr-builder-annotations.version}</version>
-                <scope>provided</scope>
+                <groupId>org.mockito</groupId>
+                <artifactId>mockito-bom</artifactId>
+                <version>${mockito.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
             </dependency>
         </dependencies>
     </dependencyManagement>


Test patch:
diff --git a/api/kroxylicious-filter-api/src/test/java/io/kroxylicious/proxy/filter/FilterInvokersTest.java b/api/kroxylicious-filter-api/src/test/java/io/kroxylicious/proxy/filter/FilterInvokersTest.java
new file mode 100644
index 00000000..e2559c6d
--- /dev/null
+++ b/api/kroxylicious-filter-api/src/test/java/io/kroxylicious/proxy/filter/FilterInvokersTest.java
@@ -0,0 +1,212 @@
+/*
+ * 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.filter;
+
+import java.util.List;
+import java.util.stream.Stream;
+
+import org.apache.kafka.common.message.ApiVersionsRequestData;
+import org.apache.kafka.common.message.ApiVersionsResponseData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class FilterInvokersTest {
+
+    @ParameterizedTest
+    @MethodSource("invalidFilters")
+    public void testInvalidFilters(KrpcFilter invalid) {
+        Assertions.assertThrows(IllegalArgumentException.class, () -> FilterInvokers.from(invalid));
+    }
+
+    @ParameterizedTest
+    @MethodSource("validFilters")
+    public void testValidFilters(KrpcFilter invalid) {
+        assertThat(FilterInvokers.from(invalid)).isNotNull();
+    }
+
+    @Test
+    public void testCompositeFilter() {
+        MultipleSpecificFilter filterA = new MultipleSpecificFilter();
+        RequestResponseFilter filterB = new RequestResponseFilter();
+        RequestFilter filterC = (apiKey, header, body, filterContext) -> {
+
+        };
+        ResponseFilter filterD = (apiKey, header, body, filterContext) -> {
+
+        };
+        List<FilterAndInvoker> from = FilterInvokers.from(
+                (CompositeFilter) () -> List.of((CompositeFilter) () -> List.of(filterA, filterB), (CompositeFilter) () -> List.of(filterC, filterD)));
+        List<KrpcFilter> filters = from.stream().map(FilterAndInvoker::filter).toList();
+        assertThat(filters).containsExactly(filterA, filterB, filterC, filterD);
+    }
+
+    public static Stream<KrpcFilter> invalidFilters() {
+        KrpcFilter noKrpcFilterSubinterfacesImplemented = new KrpcFilter() {
+
+        };
+        return Stream.of(noKrpcFilterSubinterfacesImplemented,
+                new SpecificAndRequestFilter(),
+                new SpecificAndResponseFilter(),
+                new SpecificAndCompositeFilter(),
+                new CompositeAndRequestFilter(),
+                new CompositeAndResponseFilter(),
+                new TooDeeplyNestedCompositeFilter(),
+                new SelfReferencingCompositeFilter());
+    }
+
+    public static Stream<KrpcFilter> validFilters() {
+        ApiVersionsRequestFilter singleSpecificMessageFilter = (apiVersion, header, request, context) -> {
+
+        };
+        RequestFilter requestFilter = (apiKey, header, body, filterContext) -> {
+
+        };
+        ResponseFilter responseFilter = (apiKey, header, body, filterContext) -> {
+
+        };
+        return Stream.of(singleSpecificMessageFilter,
+                new MultipleSpecificFilter(),
+                requestFilter,
+                responseFilter,
+                new RequestResponseFilter(),
+                new NoRecursionCompositeFilter(),
+                new SingleRecursionCompositeFilter());
+    }
+
+    static class RequestResponseFilter implements RequestFilter, ResponseFilter {
+
+        @Override
+        public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+
+        }
+
+        @Override
+        public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+
+        }
+    }
+
+    static class NoRecursionCompositeFilter implements CompositeFilter {
+
+        @Override
+        public List<KrpcFilter> getFilters() {
+            return List.of(new MultipleSpecificFilter());
+        }
+    }
+
+    static class SingleRecursionCompositeFilter implements CompositeFilter {
+
+        @Override
+        public List<KrpcFilter> getFilters() {
+            return List.of((CompositeFilter) () -> List.of(new MultipleSpecificFilter()), (CompositeFilter) () -> List.of(new MultipleSpecificFilter()));
+        }
+    }
+
+    static class MultipleSpecificFilter implements ApiVersionsRequestFilter, ApiVersionsResponseFilter {
+
+        @Override
+        public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+
+        }
+
+        @Override
+        public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
+
+        }
+    }
+
+    static class CompositeAndRequestFilter implements RequestFilter, CompositeFilter {
+        @Override
+        public List<KrpcFilter> getFilters() {
+            return null;
+        }
+
+        @Override
+        public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+
+        }
+    }
+
+    /**
+     * We do not want to explode Composite Filters forever
+     */
+    static class SelfReferencingCompositeFilter implements CompositeFilter {
+        @Override
+        public List<KrpcFilter> getFilters() {
+            return List.of(this);
+        }
+    }
+
+    static class TooDeeplyNestedCompositeFilter implements CompositeFilter {
+        @Override
+        public List<KrpcFilter> getFilters() {
+            return List.of((CompositeFilter) () -> List.of((CompositeFilter) () -> List.of((ApiVersionsRequestFilter) (apiVersion, header, request, context) -> {
+
+            })));
+        }
+    }
+
+    static class CompositeAndResponseFilter implements ResponseFilter, CompositeFilter {
+        @Override
+        public List<KrpcFilter> getFilters() {
+            return null;
+        }
+
+        @Override
+        public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+
+        }
+    }
+
+    static class SpecificAndCompositeFilter implements ApiVersionsRequestFilter, CompositeFilter {
+
+        @Override
+        public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+
+        }
+
+        @Override
+        public List<KrpcFilter> getFilters() {
+            return null;
+        }
+    }
+
+    static class SpecificAndResponseFilter implements ApiVersionsRequestFilter, ResponseFilter {
+
+        @Override
+        public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+
+        }
+
+        @Override
+        public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+
+        }
+    }
+
+    static class SpecificAndRequestFilter implements ApiVersionsRequestFilter, RequestFilter {
+
+        @Override
+        public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+
+        }
+
+        @Override
+        public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+
+        }
+    }
+
+}
\ No newline at end of file
diff --git a/integrationtests/pom.xml b/integrationtests/pom.xml
index 337f7029..c483e34c 100644
--- a/integrationtests/pom.xml
+++ b/integrationtests/pom.xml
@@ -26,49 +26,51 @@
     </properties>
 
     <dependencies>
+
+        <!-- project dependencies - test -->
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious</artifactId>
-            <version>0.3.0-SNAPSHOT</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
-            <artifactId>kroxylicious-test-tools</artifactId>
-            <version>0.3.0-SNAPSHOT</version>
+            <artifactId>kroxylicious-multitenant</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
-            <artifactId>kroxylicious-multitenant</artifactId>
-            <version>0.3.0-SNAPSHOT</version>
+            <artifactId>kroxylicious-test-tools</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-schema-validation</artifactId>
-            <version>0.3.0-SNAPSHOT</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious.testing</groupId>
-            <artifactId>testing-junit5-extension</artifactId>
+            <artifactId>testing-api</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious.testing</groupId>
-            <artifactId>testing-api</artifactId>
+            <artifactId>testing-junit5-extension</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <!-- third party dependencies - runtime and compile -->
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+
+        <!-- third party dependencies - test -->
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>
             <artifactId>log4j-slf4j2-impl</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-api</artifactId>
-        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-params</artifactId>
@@ -92,6 +94,7 @@
                         <TESTCONTAINERS_RYUK_DISABLED>true</TESTCONTAINERS_RYUK_DISABLED>
                     </environmentVariables>
                     <systemPropertyVariables>
+                        <io.netty.leakDetection.level>paranoid</io.netty.leakDetection.level>
                         <container.logs.dir>${project.build.directory}/container-logs/</container.logs.dir>
                     </systemPropertyVariables>
                     <runOrder>random</runOrder>
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
index 2635e9c6..10ad6634 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
@@ -8,7 +8,6 @@ package io.kroxylicious.proxy;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -20,8 +19,15 @@ import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.MetadataResponseData;
 import org.apache.kafka.common.serialization.Serdes;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
@@ -29,16 +35,21 @@ import io.kroxylicious.proxy.config.FilterDefinitionBuilder;
 import io.kroxylicious.proxy.filter.CreateTopicRejectFilter;
 import io.kroxylicious.proxy.internal.filter.ByteBufferTransformation;
 import io.kroxylicious.proxy.service.HostPort;
+import io.kroxylicious.test.Request;
+import io.kroxylicious.test.Response;
+import io.kroxylicious.test.tester.MockServerKroxyliciousTester;
 import io.kroxylicious.testing.kafka.api.KafkaCluster;
 import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
 
 import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.proxy;
 import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.withDefaultFilters;
 import static io.kroxylicious.test.tester.KroxyliciousTesters.kroxyliciousTester;
+import static io.kroxylicious.test.tester.KroxyliciousTesters.mockKafkaKroxyliciousTester;
 import static org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG;
 import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG;
 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.common.protocol.ApiKeys.METADATA;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
@@ -54,6 +65,7 @@ public class KrpcFilterIT {
             (byte) 0x64, (byte) 0x67, (byte) 0x61, (byte) 0x59, (byte) 0x16 };
     private static final byte[] TOPIC_2_CIPHERTEXT = { (byte) 0xffffffa7, (byte) 0xffffffc4, (byte) 0xffffffcb, (byte) 0xffffffcb, (byte) 0xffffffce, (byte) 0xffffff8b,
             (byte) 0x7f, (byte) 0xffffffd6, (byte) 0xffffffce, (byte) 0xffffffd1, (byte) 0xffffffcb, (byte) 0xffffffc3, (byte) 0xffffff80 };
+    private static NettyLeakLogAppender appender;
 
     @BeforeAll
     public static void checkReversibleEncryption() {
@@ -64,6 +76,20 @@ public class KrpcFilterIT {
         assertEquals(PLAINTEXT, new String(decode(TOPIC_1, ByteBuffer.wrap(TOPIC_1_CIPHERTEXT)).array(), StandardCharsets.UTF_8));
         assertArrayEquals(TOPIC_2_CIPHERTEXT, encode(TOPIC_2, ByteBuffer.wrap(PLAINTEXT.getBytes(StandardCharsets.UTF_8))).array());
         assertEquals(PLAINTEXT, new String(decode(TOPIC_2, ByteBuffer.wrap(TOPIC_2_CIPHERTEXT)).array(), StandardCharsets.UTF_8));
+
+        final LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+        final Configuration config = ctx.getConfiguration();
+        appender = (NettyLeakLogAppender) config.getAppenders().get("NettyLeakLogAppender");
+    }
+
+    @BeforeEach
+    public void clearLeaks() {
+        appender.clear();
+    }
+
+    @AfterEach
+    public void checkNoNettyLeaks() {
+        appender.verifyNoLeaks();
     }
 
     public static class TestEncoder implements ByteBufferTransformation {
@@ -130,11 +156,29 @@ public class KrpcFilterIT {
 
         try (var tester = kroxyliciousTester(config);
                 var proxyAdmin = tester.admin()) {
-            assertThatExceptionOfType(ExecutionException.class)
-                    .isThrownBy(() -> proxyAdmin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1))).all().get())
-                    .withCauseInstanceOf(InvalidTopicException.class)
-                    .havingCause()
-                    .withMessage(CreateTopicRejectFilter.ERROR_MESSAGE);
+            assertCreatingTopicThrowsExpectedException(proxyAdmin);
+
+            // check no topic created on the cluster
+            Set<String> names = admin.listTopics().names().get(10, TimeUnit.SECONDS);
+            assertThat(names).isEmpty();
+        }
+    }
+
+    @Test
+    public void requestFiltersCanRespondWithoutProxyingDoesntLeakBuffers(KafkaCluster cluster, Admin admin) throws Exception {
+        var config = withDefaultFilters(proxy(cluster))
+                .addToFilters(new FilterDefinitionBuilder("CreateTopicRejectFilter").build());
+
+        try (var tester = kroxyliciousTester(config);
+                var proxyAdmin = tester.admin()) {
+            // loop because System.gc doesn't make any guarantees that the buffer will be collected
+            for (int i = 0; i < 20; i++) {
+                // CreateTopicRejectFilter allocates a buffer and then short-circuit responds
+                assertCreatingTopicThrowsExpectedException(proxyAdmin);
+                // buffers must be garbage collected before it causes leak detection during
+                // a subsequent buffer allocation
+                System.gc();
+            }
 
             // check no topic created on the cluster
             Set<String> names = admin.listTopics().names().get(10, TimeUnit.SECONDS);
@@ -142,6 +186,26 @@ public class KrpcFilterIT {
         }
     }
 
+    private static void assertCreatingTopicThrowsExpectedException(Admin proxyAdmin) {
+        assertThatExceptionOfType(ExecutionException.class)
+                .isThrownBy(() -> proxyAdmin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1))).all().get())
+                .withCauseInstanceOf(InvalidTopicException.class)
+                .havingCause()
+                .withMessage(CreateTopicRejectFilter.ERROR_MESSAGE);
+    }
+
+    @Test
+    public void testCompositeFilter() {
+        try (MockServerKroxyliciousTester tester = mockKafkaKroxyliciousTester((mockBootstrap) -> proxy(mockBootstrap)
+                .addToFilters(new FilterDefinitionBuilder("CompositePrefixingFixedClientId")
+                        .withConfig("clientId", "banana", "prefix", "123").build()));
+                var singleRequestClient = tester.singleRequestClient()) {
+            tester.setMockResponse(new Response(METADATA, METADATA.latestVersion(), new MetadataResponseData()));
+            singleRequestClient.getSync(new Request(METADATA, METADATA.latestVersion(), "client", new MetadataRequestData()));
+            assertEquals("123banana", tester.onlyRequest().clientIdHeader());
+        }
+    }
+
     @Test
     public void shouldModifyProduceMessage(KafkaCluster cluster, Admin admin) throws Exception {
         var proxyAddress = HostPort.parse("localhost:9192");
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/NettyLeakLogAppender.java b/integrationtests/src/test/java/io/kroxylicious/proxy/NettyLeakLogAppender.java
new file mode 100644
index 00000000..d32bf74c
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/NettyLeakLogAppender.java
@@ -0,0 +1,69 @@
+/*
+ * 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.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.Filter;
+import org.apache.logging.log4j.core.Layout;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Property;
+import org.apache.logging.log4j.core.config.plugins.Plugin;
+import org.apache.logging.log4j.core.config.plugins.PluginAttribute;
+import org.apache.logging.log4j.core.config.plugins.PluginElement;
+import org.apache.logging.log4j.core.config.plugins.PluginFactory;
+import org.apache.logging.log4j.core.layout.PatternLayout;
+import org.junit.jupiter.api.Assertions;
+
+import io.netty.util.ResourceLeakDetector;
+
+@Plugin(name = "NettyLeakLogAppender", category = "Core", elementType = "appender", printObject = true)
+public class NettyLeakLogAppender extends AbstractAppender {
+    private List<String> leaks = new ArrayList<>();
+
+    public void verifyNoLeaks() {
+        if (!leaks.isEmpty()) {
+            Assertions.fail("Netty resource leak detected " + leaks);
+        }
+    }
+
+    protected NettyLeakLogAppender(String name, Filter filter, Layout<? extends Serializable> layout) {
+        super(name, filter, layout, true, Property.EMPTY_ARRAY);
+    }
+
+    @Override
+    public void append(LogEvent event) {
+        if (event.getLoggerName().equals(ResourceLeakDetector.class.getName()) && event.getLevel().equals(Level.ERROR) && event.getMessage().getFormattedMessage()
+                .contains("LEAK:")) {
+            leaks.add(event.getMessage().getFormattedMessage());
+        }
+    }
+
+    @PluginFactory
+    public static NettyLeakLogAppender createAppender(
+                                                      @PluginAttribute("name") String name,
+                                                      @PluginElement("Layout") Layout<? extends Serializable> layout,
+                                                      @PluginElement("Filter") final Filter filter,
+                                                      @PluginAttribute("otherAttribute") String otherAttribute) {
+        if (name == null) {
+            LOGGER.error("No name provided for TestAppender");
+            return null;
+        }
+        if (layout == null) {
+            layout = PatternLayout.createDefaultLayout();
+        }
+        return new NettyLeakLogAppender(name, filter, layout);
+    }
+
+    public void clear() {
+        leaks.clear();
+    }
+}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CompositePrefixingFixedClientIdFilter.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CompositePrefixingFixedClientIdFilter.java
new file mode 100644
index 00000000..c8261875
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CompositePrefixingFixedClientIdFilter.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.filter;
+
+import java.util.List;
+
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import io.kroxylicious.proxy.config.BaseConfig;
+
+public class CompositePrefixingFixedClientIdFilter implements CompositeFilter {
+
+    private final CompositePrefixingFixedClientIdFilterConfig config;
+
+    public CompositePrefixingFixedClientIdFilter(CompositePrefixingFixedClientIdFilterConfig config) {
+        this.config = config;
+    }
+
+    private class PrefixingFilter implements RequestFilter {
+        @Override
+        public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+            header.setClientId(config.prefix + header.clientId());
+            filterContext.forwardRequest(header, body);
+        }
+    }
+
+    @Override
+    public List<KrpcFilter> getFilters() {
+        FixedClientIdFilter clientIdFilter = new FixedClientIdFilter(new FixedClientIdFilter.FixedClientIdFilterConfig(config.clientId));
+        return List.of(clientIdFilter, new PrefixingFilter());
+    }
+
+    public static class CompositePrefixingFixedClientIdFilterConfig extends BaseConfig {
+        private final String prefix;
+        private final String clientId;
+
+        @JsonCreator
+        public CompositePrefixingFixedClientIdFilterConfig(@JsonProperty("prefix") String prefix, @JsonProperty("clientId") String clientId) {
+            this.prefix = prefix;
+            this.clientId = clientId;
+        }
+    }
+}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CreateTopicRejectFilter.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CreateTopicRejectFilter.java
index 662fc33c..8a2283cc 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CreateTopicRejectFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/CreateTopicRejectFilter.java
@@ -19,6 +19,7 @@ public class CreateTopicRejectFilter implements CreateTopicsRequestFilter {
     public void onCreateTopicsRequest(short apiVersion, RequestHeaderData header, CreateTopicsRequestData request, KrpcFilterContext context) {
         CreateTopicsResponseData response = new CreateTopicsResponseData();
         CreateTopicsResponseData.CreatableTopicResultCollection topics = new CreateTopicsResponseData.CreatableTopicResultCollection();
+        allocateByteBufToTestKroxyliciousReleasesIt(context);
         request.topics().forEach(creatableTopic -> {
             CreateTopicsResponseData.CreatableTopicResult result = new CreateTopicsResponseData.CreatableTopicResult();
             result.setErrorCode(Errors.INVALID_TOPIC_EXCEPTION.code()).setErrorMessage(ERROR_MESSAGE);
@@ -28,4 +29,8 @@ public class CreateTopicRejectFilter implements CreateTopicsRequestFilter {
         response.setTopics(topics);
         context.forwardResponse(response);
     }
+
+    private static void allocateByteBufToTestKroxyliciousReleasesIt(KrpcFilterContext context) {
+        context.createByteBufferOutputStream(4000);
+    }
 }
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java
index 842b2882..e82eee70 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java
@@ -5,12 +5,14 @@
  */
 package io.kroxylicious.proxy.filter;
 
+import io.kroxylicious.proxy.filter.CompositePrefixingFixedClientIdFilter.CompositePrefixingFixedClientIdFilterConfig;
 import io.kroxylicious.proxy.service.BaseContributor;
 
 public class TestFilterContributor extends BaseContributor<KrpcFilter> implements FilterContributor {
 
     public static final BaseContributorBuilder<KrpcFilter> FILTERS = BaseContributor.<KrpcFilter> builder()
             .add("FixedClientId", FixedClientIdFilter.FixedClientIdFilterConfig.class, FixedClientIdFilter::new)
+            .add("CompositePrefixingFixedClientId", CompositePrefixingFixedClientIdFilterConfig.class, CompositePrefixingFixedClientIdFilter::new)
             .add("CreateTopicRejectFilter", CreateTopicRejectFilter::new);
 
     public TestFilterContributor() {
diff --git a/integrationtests/src/test/resources/log4j2-test.properties b/integrationtests/src/test/resources/log4j2-test.properties
index a95d06a5..2abaaa2b 100644
--- a/integrationtests/src/test/resources/log4j2-test.properties
+++ b/integrationtests/src/test/resources/log4j2-test.properties
@@ -11,11 +11,23 @@ appender.console.name = STDOUT
 appender.console.layout.type = PatternLayout
 appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %-5p %c:%L - %m%n
 
+appender.NettyLeakLogAppender.type = NettyLeakLogAppender
+appender.NettyLeakLogAppender.name = NettyLeakLogAppender
+appender.NettyLeakLogAppender.layout.type = PatternLayout
+appender.NettyLeakLogAppender.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %-5p %c:%L - %m%n
+
+logger.NettyResourceLeak.name = io.netty.util.ResourceLeakDetector
+logger.NettyResourceLeak.level = ERROR
+logger.NettyResourceLeak.additivity = true
+logger.NettyResourceLeak.appenderRefs = NettyLeakLogAppender
+logger.NettyResourceLeak.appenderRef.NettyLeakLogAppender.ref = NettyLeakLogAppender
+
 rootLogger.level = WARN
 rootLogger.appenderRefs = console
 rootLogger.appenderRef.console.ref = STDOUT
 rootLogger.additivity = false
 
+
 #logger.kproxy.name = io.kroxylicious.proxy
 #logger.kproxy.level = INFO
 #logger.kproxy.additivity = false
@@ -39,7 +51,6 @@ logger.broker.appenderRef.console.ref = STDOUT
 logger.net.name = org.apache.kafka.clients.NetworkClient
 logger.net.level = debug
 
-
 #
 #logger.scl.name = state.change.logger
 #logger.scl.level = WARN
diff --git a/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java b/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
index f92241b8..996ec724 100644
--- a/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
+++ b/kroxylicious-additional-filters/kroxylicious-multitenant/src/test/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantTransformationFilterTest.java
@@ -26,13 +26,14 @@ import com.flipkart.zjsonpatch.JsonDiff;
 import com.google.common.reflect.ClassPath;
 import com.google.common.reflect.ClassPath.ResourceInfo;
 
+import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.filter.FilterInvoker;
-import io.kroxylicious.proxy.filter.FilterInvokers;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.test.requestresponsetestdef.ApiMessageTestDef;
 import io.kroxylicious.test.requestresponsetestdef.RequestResponseTestDef;
 
 import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.Iterables.getOnlyElement;
 import static io.kroxylicious.test.requestresponsetestdef.KafkaApiMessageConverter.requestConverterFor;
 import static io.kroxylicious.test.requestresponsetestdef.KafkaApiMessageConverter.responseConverterFor;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -59,7 +60,7 @@ class MultiTenantTransformationFilterTest {
 
     private final MultiTenantTransformationFilter filter = new MultiTenantTransformationFilter();
 
-    private final FilterInvoker invoker = FilterInvokers.from(filter);
+    private final FilterInvoker invoker = getOnlyElement(FilterAndInvoker.build(filter)).invoker();
 
     private final KrpcFilterContext context = mock(KrpcFilterContext.class);
 
diff --git a/kroxylicious-test-tools/pom.xml b/kroxylicious-test-tools/pom.xml
index 351598c4..28e3b53d 100644
--- a/kroxylicious-test-tools/pom.xml
+++ b/kroxylicious-test-tools/pom.xml
@@ -22,24 +22,29 @@
     <name>Testing tools</name>
 
     <dependencies>
+
+        <!-- project dependencies - runtime and compile -->
         <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-clients</artifactId>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious</artifactId>
         </dependency>
+
+        <!-- project dependencies - test -->
         <dependency>
             <groupId>io.kroxylicious.testing</groupId>
-            <artifactId>testing-junit5-extension</artifactId>
+            <artifactId>testing-api</artifactId>
             <scope>compile</scope>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious.testing</groupId>
-            <artifactId>testing-api</artifactId>
+            <artifactId>testing-junit5-extension</artifactId>
             <scope>compile</scope>
         </dependency>
+
+        <!-- third party dependencies - runtime and compile -->
         <dependency>
-            <groupId>io.kroxylicious</groupId>
-            <artifactId>kroxylicious</artifactId>
-            <version>0.3.0-SNAPSHOT</version>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
         </dependency>
         <dependency>
             <groupId>io.netty</groupId>
@@ -77,24 +82,26 @@
             <groupId>io.sundr</groupId>
             <artifactId>builder-annotations</artifactId>
         </dependency>
+
+        <!-- third party dependencies - test -->
         <dependency>
             <groupId>com.flipkart.zjsonpatch</groupId>
             <artifactId>zjsonpatch</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter</artifactId>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter-params</artifactId>
+            <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.assertj</groupId>
-            <artifactId>assertj-core</artifactId>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
diff --git a/kroxylicious-unit-test/pom.xml b/kroxylicious-unit-test/pom.xml
index 53162101..4d961e74 100644
--- a/kroxylicious-unit-test/pom.xml
+++ b/kroxylicious-unit-test/pom.xml
@@ -23,10 +23,7 @@
     <description>Classes to help unit test Filter implementations</description>
 
     <dependencies>
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-clients</artifactId>
-        </dependency>
+        <!-- third party dependencies - runtime and compile -->
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
@@ -40,6 +37,10 @@
             <artifactId>guava</artifactId>
             <scope>compile</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+        </dependency>
     </dependencies>
     <build>
         <plugins>
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
index 75393a21..40bac954 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
@@ -15,10 +15,12 @@ import org.junit.jupiter.api.AfterEach;
 
 import io.netty.channel.embedded.EmbeddedChannel;
 
+import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.filter.KrpcFilter;
 import io.kroxylicious.proxy.frame.DecodedRequestFrame;
 import io.kroxylicious.proxy.frame.DecodedResponseFrame;
 
+import static com.google.common.collect.Iterables.getOnlyElement;
 import static org.junit.jupiter.api.Assertions.fail;
 
 /**
@@ -47,7 +49,7 @@ public abstract class FilterHarness {
      */
     protected void buildChannel(KrpcFilter filter, long timeoutMs) {
         this.filter = filter;
-        filterHandler = new FilterHandler(filter, timeoutMs, null);
+        filterHandler = new FilterHandler(getOnlyElement(FilterAndInvoker.build(filter)), timeoutMs, null);
         channel = new EmbeddedChannel(filterHandler);
     }
 
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderTest.java
index f1cb3899..7a3b2283 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderTest.java
@@ -19,6 +19,7 @@ import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 
 import io.kroxylicious.proxy.filter.ApiVersionsRequestFilter;
+import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.frame.DecodedRequestFrame;
 import io.kroxylicious.proxy.frame.OpaqueRequestFrame;
@@ -39,7 +40,8 @@ public class RequestDecoderTest extends AbstractCodecTest {
                         AbstractCodecTest::deserializeApiVersionsRequestUsingKafkaApis,
                         new KafkaRequestDecoder(
                                 DecodePredicate
-                                        .forFilters(List.of((ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request)))),
+                                        .forFilters(FilterAndInvoker.build(
+                                                (ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request)))),
                         DecodedRequestFrame.class,
                         (RequestHeaderData header) -> header),
                 "Unexpected correlation id");
@@ -53,8 +55,8 @@ public class RequestDecoderTest extends AbstractCodecTest {
                         ApiKeys.API_VERSIONS::requestHeaderVersion,
                         AbstractCodecTest::exampleRequestHeader,
                         AbstractCodecTest::exampleApiVersionsRequest,
-                        new KafkaRequestDecoder(DecodePredicate.forFilters(List.of(
-                                new ApiVersionsRequestFilter() {
+                        new KafkaRequestDecoder(DecodePredicate.forFilters(
+                                FilterAndInvoker.build(new ApiVersionsRequestFilter() {
 
                                     @Override
                                     public boolean shouldHandleApiVersionsRequest(short apiVersion) {
@@ -84,7 +86,8 @@ public class RequestDecoderTest extends AbstractCodecTest {
 
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
-                DecodePredicate.forFilters(List.of((ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request))))
+                DecodePredicate.forFilters(
+                        FilterAndInvoker.build((ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request))))
                 .decode(null, byteBuf, messages);
 
         assertEquals(List.of(), messageClasses(messages));
@@ -102,7 +105,8 @@ public class RequestDecoderTest extends AbstractCodecTest {
 
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
-                DecodePredicate.forFilters(List.of((ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request))))
+                DecodePredicate.forFilters(
+                        FilterAndInvoker.build((ApiVersionsRequestFilter) (version, header, request, context) -> context.forwardRequest(header, request))))
                 .decode(null, byteBuf, messages);
 
         assertEquals(List.of(), messageClasses(messages));
@@ -144,7 +148,8 @@ public class RequestDecoderTest extends AbstractCodecTest {
 
         var messages = new ArrayList<>();
         new KafkaRequestDecoder(
-                DecodePredicate.forFilters(List.of((ApiVersionsRequestFilter) (version, head, request, context) -> context.forwardRequest(header, request))))
+                DecodePredicate.forFilters(
+                        FilterAndInvoker.build((ApiVersionsRequestFilter) (version, head, request, context) -> context.forwardRequest(header, request))))
                 .decode(null, byteBuf, messages);
 
         assertEquals(List.of(DecodedRequestFrame.class, DecodedRequestFrame.class), messageClasses(messages));
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 15c62296..d37605ec 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
@@ -34,8 +34,8 @@ import org.mockito.junit.jupiter.MockitoExtension;
 import com.flipkart.zjsonpatch.JsonDiff;
 import com.google.common.reflect.ClassPath;
 
+import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.filter.FilterInvoker;
-import io.kroxylicious.proxy.filter.FilterInvokers;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 import io.kroxylicious.proxy.internal.net.EndpointReconciler;
 import io.kroxylicious.proxy.model.VirtualCluster;
@@ -44,6 +44,7 @@ import io.kroxylicious.test.requestresponsetestdef.ApiMessageTestDef;
 import io.kroxylicious.test.requestresponsetestdef.RequestResponseTestDef;
 
 import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.Iterables.getOnlyElement;
 import static io.kroxylicious.test.requestresponsetestdef.KafkaApiMessageConverter.responseConverterFor;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
@@ -102,7 +103,7 @@ class BrokerAddressFilterTest {
     @BeforeEach
     public void beforeEach() {
         filter = new BrokerAddressFilter(virtualCluster, endpointReconciler);
-        invoker = FilterInvokers.from(filter);
+        invoker = getOnlyElement(FilterAndInvoker.build(filter)).invoker();
         when(virtualCluster.getBrokerAddress(0)).thenReturn(HostPort.parse("downstream:19199"));
 
         var nodeMap = Map.of(0, HostPort.parse("upstream:9199"));
diff --git a/performance-tests/pom.xml b/performance-tests/pom.xml
index c4f6f8cf..d476484b 100644
--- a/performance-tests/pom.xml
+++ b/performance-tests/pom.xml
@@ -20,17 +20,24 @@
 
     <name>Micro benchmarks</name>
 
+    <properties>
+        <jmh.version>1.36</jmh.version>
+        <uberjar.name>benchmarks</uberjar.name>
+    </properties>
+
     <dependencies>
+
+        <!-- project dependencies - runtime and compile -->
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-filter-api</artifactId>
-            <version>${kroxyliciousApi.version}</version>
         </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-test-tools</artifactId>
-            <version>0.3.0-SNAPSHOT</version>
         </dependency>
+
+        <!-- third party dependencies - runtime and compile -->
         <dependency>
             <groupId>org.openjdk.jmh</groupId>
             <artifactId>jmh-core</artifactId>
@@ -44,11 +51,6 @@
         </dependency>
     </dependencies>
 
-    <properties>
-        <jmh.version>1.36</jmh.version>
-        <uberjar.name>benchmarks</uberjar.name>
-    </properties>
-
     <build>
         <plugins>
             <plugin>