Repository: kroxylicious/kroxylicious
Pull request #392 (2023-06-18 23:28:35)

Introduce CompositeFilters
Closes https://github.com/kroxylicious/kroxylicious/issues/348

### Type of change

- Enhancement / new feature

### Description

Filter Authors may now implement `CompositeFilter` if they wish a single filter definition in the configuration YAML to contribute multiple filters to the filter chain for a channel. `CompositeFilter` is not allowed to be combined with `RequestFilter`, `ResponseFilter `or any specific message Filter interfaces.

```
public interface CompositeFilter extends KrpcFilter {

    List<KrpcFilter> getFilters();

}
```

### Additional Context

In some cases it may be preferable if we can contribute multiple filters for a single Filter definition in the config.

For example for multi-tenancy we need to use per-MessageType filters to intercept specific RPCs. But we also will want to map all clientIds in request headers, which makes more sense as a RequestFilter that intercepts all request. These types of filters cannot be mixed in a single implementing class currently, so the Users would have to add two filter definitions to their YAML.

It would be better if they could declare a single type and have the required filters installed into the chain in the right order. The configuration is simplified and better for evolution too. It allows us to decompose a single filter into multiple, or recombine them without the user configuration having to change.

By making this an interface we enable some interesting options like users could implement anonymous or private Filter classes that share state.

### Checklist

- [x] Write tests
- [x] Make sure all tests pass
- [x] Make sure all Sonar-Lint warnings are addressed or are justifiably ignored.
- [ ] Update documentation
- [x] Reference relevant issue(s) and close them after merging
- [x] Update CHANGELOG.md


robobario: This is an alternative to https://github.com/kroxylicious/kroxylicious/pull/390

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 00b31a24..d177567d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,7 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 
 ## 0.3.0
 
+* [#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
@@ -19,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/api/kroxylicious-filter-api/pom.xml b/api/kroxylicious-filter-api/pom.xml
index 3ecbf2dd..786fc5e1 100644
--- a/api/kroxylicious-filter-api/pom.xml
+++ b/api/kroxylicious-filter-api/pom.xml
@@ -51,6 +51,21 @@
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-annotations</artifactId>
         </dependency>
+        <dependency>
+            <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>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
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/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/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 + "}";
             }
         };
     }


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/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
index eea3697b..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,6 +19,8 @@ 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;
@@ -34,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;
@@ -188,6 +194,18 @@ public class KrpcFilterIT {
                 .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/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/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/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/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"));