Repository: kroxylicious/kroxylicious
Pull request #357 (2023-06-01 16:47:25)

Renaming CECP
### Type of change

_Select the type of your PR_

- Refactoring

### Description

Completes some review actions made during #271 .

_Please describe your pull request_

### Additional Context

_Why are you making this pull request?_



Patch:
diff --git a/.idea/compiler.xml b/.idea/compiler.xml
index f890df78..97d84287 100644
--- a/.idea/compiler.xml
+++ b/.idea/compiler.xml
@@ -16,9 +16,11 @@
         <module name="integrationtests" />
         <module name="integrationtests.test" />
         <module name="kroxylicious-krpc-plugin.main" />
+        <module name="kroxylicious-schema-validation" />
         <module name="kroxylicious.test" />
         <module name="kroxylicious-filter-api" />
         <module name="kroxylicious-krpc-plugin" />
+        <module name="performance-tests" />
         <module name="kroxylicious-krpc-plugin.test" />
       </profile>
     </annotationProcessing>
@@ -36,9 +38,11 @@
       <module name="kroxylicious-krpc-plugin.test" options="-parameters" />
       <module name="kroxylicious-multitenant" options="-parameters" />
       <module name="kroxylicious-parent" options="-parameters" />
+      <module name="kroxylicious-schema-validation" options="-parameters" />
       <module name="kroxylicious-test-tools" options="-parameters" />
       <module name="kroxylicious.main" options="-parameters" />
       <module name="kroxylicious.test" options="-parameters" />
+      <module name="performance-tests" options="-parameters" />
     </option>
   </component>
 </project>
\ No newline at end of file
diff --git a/.idea/encodings.xml b/.idea/encodings.xml
index fa575bd8..900f39a0 100644
--- a/.idea/encodings.xml
+++ b/.idea/encodings.xml
@@ -10,6 +10,8 @@
     <file url="file://$PROJECT_DIR$/integrationtests/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/kroxylicious-multitenant/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/kroxylicious-multitenant/src/main/resources" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/kroxylicious-schema-validation/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/kroxylicious-schema-validation/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/kroxylicious-test-tools/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/kroxylicious-test-tools/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/kroxylicious-test-tools/target/generated-sources/krpc" charset="UTF-8" />
@@ -19,6 +21,8 @@
     <file url="file://$PROJECT_DIR$/kroxylicious/target/generated-sources/krpc" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/krpc-code-gen/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/krpc-code-gen/src/main/resources" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/performance-tests/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/performance-tests/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/src/main/resources" charset="UTF-8" />
   </component>
diff --git a/.idea/misc.xml b/.idea/misc.xml
index de08349c..1dd8bf6f 100644
--- a/.idea/misc.xml
+++ b/.idea/misc.xml
@@ -11,5 +11,5 @@
   <component name="PDMPlugin">
     <option name="skipTestSources" value="false" />
   </component>
-  <component name="ProjectRootManager" version="2" languageLevel="JDK_19" default="true" project-jdk-name="temurin-19" project-jdk-type="JavaSDK" />
+  <component name="ProjectRootManager" version="2" languageLevel="JDK_20" default="true" project-jdk-name="temurin-20" project-jdk-type="JavaSDK" />
 </project>
\ No newline at end of file
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusterendpointprovider/ClusterEndpointProviderContributor.java b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusterendpointprovider/ClusterEndpointProviderContributor.java
deleted file mode 100644
index a325c71b..00000000
--- a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusterendpointprovider/ClusterEndpointProviderContributor.java
+++ /dev/null
@@ -1,16 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-package io.kroxylicious.proxy.clusterendpointprovider;
-
-import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
-import io.kroxylicious.proxy.service.Contributor;
-
-/**
- * ClusterEndpointConfigProvider is a pluggable source of Kroxylicious endpoint provider implementations.
- * @see Contributor
- */
-public interface ClusterEndpointProviderContributor extends Contributor<ClusterEndpointConfigProvider> {
-}
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusternetworkaddressconfigprovider/ClusterNetworkAddressConfigProviderContributor.java b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusternetworkaddressconfigprovider/ClusterNetworkAddressConfigProviderContributor.java
new file mode 100644
index 00000000..dd443b02
--- /dev/null
+++ b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/clusternetworkaddressconfigprovider/ClusterNetworkAddressConfigProviderContributor.java
@@ -0,0 +1,16 @@
+/*
+ * 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.clusternetworkaddressconfigprovider;
+
+import io.kroxylicious.proxy.service.ClusterNetworkAddressConfigProvider;
+import io.kroxylicious.proxy.service.Contributor;
+
+/**
+ * ClusterNetworkAddressConfigProviderContributor is a pluggable source of network address information.
+ * @see Contributor
+ */
+public interface ClusterNetworkAddressConfigProviderContributor extends Contributor<ClusterNetworkAddressConfigProvider> {
+}
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterEndpointConfigProvider.java b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterNetworkAddressConfigProvider.java
similarity index 97%
rename from api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterEndpointConfigProvider.java
rename to api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterNetworkAddressConfigProvider.java
index 39b38df9..8246ef31 100644
--- a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterEndpointConfigProvider.java
+++ b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ClusterNetworkAddressConfigProvider.java
@@ -11,7 +11,7 @@ import java.util.Set;
 /**
  * Provides the addresses of the network endpoints required by a virtual cluster.
  */
-public interface ClusterEndpointConfigProvider {
+public interface ClusterNetworkAddressConfigProvider {
 
     /**
      * Address of the cluster's bootstrap address.
diff --git a/api/kroxylicious-filter-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl b/api/kroxylicious-filter-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl
index 71b18cb0..6ac5be2c 100644
--- a/api/kroxylicious-filter-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl
+++ b/api/kroxylicious-filter-api/src/main/templates/Kproxy/SpecificFilterArrayInvoker.ftl
@@ -66,13 +66,24 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
      * @param filterContext The filter context.
      */
     @Override
+    @SuppressWarnings("SwitchStatementWithTooFewBranches")
     public void onRequest(ApiKeys apiKey,
                            short apiVersion,
                            RequestHeaderData header,
                            ApiMessage body,
                            KrpcFilterContext filterContext) {
-        FilterInvoker invoker = requestInvokers[apiKey.id];
-        invoker.onRequest(apiKey, apiVersion, header, body, filterContext);
+        // We wrap the array lookup in a switch based on the API Key as it supports JIT optimisations around method dispatch.
+        // See the InvokerDispatchBenchmark micro benchmark for a comparison
+        switch (apiKey) {
+<#list messageSpecs as messageSpec>
+    <#if messageSpec.type?lower_case == 'request'>
+            case ${retrieveApiKey(messageSpec)} ->
+                requestInvokers[apiKey.id].onRequest(apiKey, apiVersion, header, body, filterContext);
+    </#if>
+</#list>
+            default -> throw new IllegalStateException("Unsupported RPC " + apiKey);
+        }
+
     }
 
     /**
@@ -84,13 +95,24 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
      * @param filterContext The filter context.
      */
     @Override
+    @SuppressWarnings("SwitchStatementWithTooFewBranches")
     public void onResponse(ApiKeys apiKey,
                             short apiVersion,
                             ResponseHeaderData header,
                             ApiMessage body,
                             KrpcFilterContext filterContext) {
-        FilterInvoker invoker = responseInvokers[apiKey.id];
-        invoker.onResponse(apiKey, apiVersion, header, body, filterContext);
+        // We wrap the array lookup in a switch based on the API Key as it supports JIT optimisations around method dispatch.
+        // See the InvokerDispatchBenchmark micro benchmark for a comparison
+        switch (apiKey) {
+<#list messageSpecs as messageSpec>
+    <#if messageSpec.type?lower_case == 'response'>
+            case ${retrieveApiKey(messageSpec)} ->
+                    responseInvokers[apiKey.id].onResponse(apiKey, apiVersion, header, body, filterContext);
+    </#if>
+</#list>
+            default -> throw new IllegalStateException("Unsupported RPC " + apiKey);
+        }
+
     }
 
     /**
@@ -110,9 +132,20 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
      * @return true if request should be deserialized
      */
     @Override
+    @SuppressWarnings("SwitchStatementWithTooFewBranches")
     public boolean shouldHandleRequest(ApiKeys apiKey, short apiVersion) {
-        FilterInvoker invoker = requestInvokers[apiKey.id];
-        return invoker.shouldHandleRequest(apiKey, apiVersion);
+        // We wrap the array lookup in a switch based on the API Key as it supports JIT optimisations around method dispatch.
+        // See the InvokerDispatchBenchmark micro benchmark for a comparison
+        return switch (apiKey) {
+<#list messageSpecs as messageSpec>
+    <#if messageSpec.type?lower_case == 'request'>
+            case ${retrieveApiKey(messageSpec)} ->
+                    requestInvokers[apiKey.id].shouldHandleRequest(apiKey, apiVersion);
+    </#if>
+</#list>
+            default -> throw new IllegalStateException("Unsupported RPC " + apiKey);
+        };
+
     }
 
     /**
@@ -132,9 +165,20 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
      * @return true if response should be deserialized
      */
     @Override
+    @SuppressWarnings("SwitchStatementWithTooFewBranches")
     public boolean shouldHandleResponse(ApiKeys apiKey, short apiVersion) {
-        FilterInvoker invoker = responseInvokers[apiKey.id];
-        return invoker.shouldHandleResponse(apiKey, apiVersion);
+        return switch (apiKey) {
+            // We wrap the array lookup in a switch based on the API Key as it supports JIT optimisations around method dispatch.
+            // See the InvokerDispatchBenchmark micro benchmark for a comparison
+<#list messageSpecs as messageSpec>
+    <#if messageSpec.type?lower_case == 'response'>
+            case ${retrieveApiKey(messageSpec)} ->
+                    responseInvokers[apiKey.id].shouldHandleResponse(apiKey, apiVersion);
+    </#if>
+</#list>
+            default -> throw new IllegalStateException("Unsupported RPC " + apiKey);
+        };
+
     }
 
     /**
@@ -149,9 +193,7 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
 
     private static FilterInvoker[] createHandleNothing() {
         FilterInvoker[] filterInvokers = emptyInvokerArraySizedForMessageTypes();
-        Arrays.stream(ApiMessageType.values()).mapToInt(ApiMessageType::apiKey).forEach(value -> {
-            filterInvokers[value] = FilterInvokers.handleNothingInvoker();
-        });
+        Arrays.stream(ApiMessageType.values()).mapToInt(ApiMessageType::apiKey).forEach(value -> filterInvokers[value] = FilterInvokers.handleNothingInvoker());
         return filterInvokers;
     }
 
@@ -160,9 +202,7 @@ class SpecificFilterArrayInvoker implements FilterInvoker {
             return HANDLE_NOTHING;
         }
         FilterInvoker[] filterInvokers = emptyInvokerArraySizedForMessageTypes();
-        Arrays.stream(ApiMessageType.values()).mapToInt(ApiMessageType::apiKey).forEach(value -> {
-            filterInvokers[value] = filterInvokersByApiMessageId.getOrDefault(value, FilterInvokers.handleNothingInvoker());
-        });
+        Arrays.stream(ApiMessageType.values()).mapToInt(ApiMessageType::apiKey).forEach(value -> filterInvokers[value] = filterInvokersByApiMessageId.getOrDefault(value, FilterInvokers.handleNothingInvoker()));
         return filterInvokers;
     }
 
diff --git a/etc/checkstyle-suppressions.xml b/etc/checkstyle-suppressions.xml
index 6d8be0c3..1a692996 100644
--- a/etc/checkstyle-suppressions.xml
+++ b/etc/checkstyle-suppressions.xml
@@ -15,5 +15,5 @@
     <suppress checks=".*"
               files="io[/\\]kroxylicious[/\\]proxy[/\\].*(Builder|Editable|Fluent|Nested|Visitor|Visitable).*\.java"/>
     <suppress checks=".*"
-              files="io[/\\]kroxylicious[/\\]jmh_generated[/\\].*"/>
+              files="io[/\\]kroxylicious[/\\]benchmarks[/\\]jmh_generated[/\\].*"/>
 </suppressions>
diff --git a/kroxylicious/example-proxy-config.yml b/kroxylicious/example-proxy-config.yml
index 095dcee2..c1413b11 100644
--- a/kroxylicious/example-proxy-config.yml
+++ b/kroxylicious/example-proxy-config.yml
@@ -12,7 +12,7 @@ virtualClusters:
   demo:
     targetCluster:
       bootstrap_servers: localhost:9092
-    clusterEndpointConfigProvider:
+    clusterNetworkAddressConfigProvider:
       type: PortPerBroker
       config:
         bootstrapAddress: localhost:9192
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderConverter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderConverter.java
deleted file mode 100644
index 3f39c0d6..00000000
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderConverter.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-
-package io.kroxylicious.proxy.config;
-
-import com.fasterxml.jackson.databind.util.StdConverter;
-
-import io.kroxylicious.proxy.internal.clusterendpointprovider.ClusterEndpointConfigProviderContributorManager;
-import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
-
-public class ClusterEndpointConfigProviderConverter extends StdConverter<ClusterEndpointConfigProviderDefinition, ClusterEndpointConfigProvider> {
-
-    @Override
-    public ClusterEndpointConfigProvider convert(ClusterEndpointConfigProviderDefinition value) {
-        return ClusterEndpointConfigProviderContributorManager.getInstance()
-                .getClusterEndpointConfigProvider(value.type(), value.config());
-
-    }
-
-}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterNetworkAddressConfigProviderConverter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterNetworkAddressConfigProviderConverter.java
new file mode 100644
index 00000000..992f3e6c
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterNetworkAddressConfigProviderConverter.java
@@ -0,0 +1,23 @@
+/*
+ * 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.config;
+
+import com.fasterxml.jackson.databind.util.StdConverter;
+
+import io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.ClusterNetworkAddressConfigProviderContributorManager;
+import io.kroxylicious.proxy.service.ClusterNetworkAddressConfigProvider;
+
+public class ClusterNetworkAddressConfigProviderConverter extends StdConverter<ClusterNetworkAddressConfigProviderDefinition, ClusterNetworkAddressConfigProvider> {
+
+    @Override
+    public ClusterNetworkAddressConfigProvider convert(ClusterNetworkAddressConfigProviderDefinition value) {
+        return ClusterNetworkAddressConfigProviderContributorManager.getInstance()
+                .getClusterEndpointConfigProvider(value.type(), value.config());
+
+    }
+
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderDefinition.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterNetworkAddressConfigProviderDefinition.java
similarity index 62%
rename from kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderDefinition.java
rename to kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterNetworkAddressConfigProviderDefinition.java
index ee0d2679..bcb573cf 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderDefinition.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterNetworkAddressConfigProviderDefinition.java
@@ -9,14 +9,14 @@ import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.databind.annotation.JsonTypeIdResolver;
 
-public class ClusterEndpointConfigProviderDefinition {
+public class ClusterNetworkAddressConfigProviderDefinition {
 
     private final String type;
     private final BaseConfig config;
 
     @JsonCreator
-    public ClusterEndpointConfigProviderDefinition(String type,
-                                                   @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.EXTERNAL_PROPERTY, property = "type") @JsonTypeIdResolver(ClusterEndpointConfigProviderTypeIdResolver.class) BaseConfig config) {
+    public ClusterNetworkAddressConfigProviderDefinition(String type,
+                                                         @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.EXTERNAL_PROPERTY, property = "type") @JsonTypeIdResolver(ClusterNetworkAddressConfigProviderTypeIdResolver.class) BaseConfig config) {
         this.type = type;
         this.config = config;
     }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderTypeIdResolver.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterNetworkAddressConfigProviderTypeIdResolver.java
similarity index 78%
rename from kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderTypeIdResolver.java
rename to kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterNetworkAddressConfigProviderTypeIdResolver.java
index 1e012c4f..67b2cc2a 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterEndpointConfigProviderTypeIdResolver.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/ClusterNetworkAddressConfigProviderTypeIdResolver.java
@@ -12,9 +12,9 @@ import com.fasterxml.jackson.databind.DatabindContext;
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.jsontype.impl.TypeIdResolverBase;
 
-import io.kroxylicious.proxy.internal.clusterendpointprovider.ClusterEndpointConfigProviderContributorManager;
+import io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.ClusterNetworkAddressConfigProviderContributorManager;
 
-public class ClusterEndpointConfigProviderTypeIdResolver extends TypeIdResolverBase {
+public class ClusterNetworkAddressConfigProviderTypeIdResolver extends TypeIdResolverBase {
     private JavaType superType;
 
     @Override
@@ -39,7 +39,7 @@ public class ClusterEndpointConfigProviderTypeIdResolver extends TypeIdResolverB
 
     @Override
     public JavaType typeFromId(DatabindContext context, String id) throws IOException {
-        Class<?> subType = ClusterEndpointConfigProviderContributorManager.getInstance().getConfigType(id);
+        Class<?> subType = ClusterNetworkAddressConfigProviderContributorManager.getInstance().getConfigType(id);
         return context.constructSpecializedType(superType, subType);
     }
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/VirtualCluster.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/VirtualCluster.java
index 9dabe8d1..354f3792 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/VirtualCluster.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/VirtualCluster.java
@@ -24,10 +24,10 @@ import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 
-import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
+import io.kroxylicious.proxy.service.ClusterNetworkAddressConfigProvider;
 import io.kroxylicious.proxy.service.HostPort;
 
-public class VirtualCluster implements ClusterEndpointConfigProvider {
+public class VirtualCluster implements ClusterNetworkAddressConfigProvider {
 
     private final TargetCluster targetCluster;
 
@@ -37,17 +37,19 @@ public class VirtualCluster implements ClusterEndpointConfigProvider {
     private final boolean logNetwork;
 
     private final boolean logFrames;
-    private final ClusterEndpointConfigProvider endpointProvider;
+
+    // There appear to be a Jackson defect that is triggered if this field name matches the name of the ctor's parameter causing it to ignore the converter.
+    private final ClusterNetworkAddressConfigProvider provider;
 
     public VirtualCluster(@JsonProperty(required = true) TargetCluster targetCluster,
-                          @JsonProperty(required = true) @JsonDeserialize(converter = ClusterEndpointConfigProviderConverter.class) ClusterEndpointConfigProvider clusterEndpointConfigProvider,
+                          @JsonProperty(required = true) @JsonDeserialize(converter = ClusterNetworkAddressConfigProviderConverter.class) ClusterNetworkAddressConfigProvider clusterNetworkAddressConfigProvider,
                           Optional<String> keyStoreFile,
                           Optional<String> keyPassword,
                           boolean logNetwork, boolean logFrames) {
-        if (clusterEndpointConfigProvider.requiresTls() && keyStoreFile.isEmpty()) {
+        if (clusterNetworkAddressConfigProvider.requiresTls() && keyStoreFile.isEmpty()) {
             throw new IllegalStateException("Cluster endpoint provider requires tls, but this virtual cluster does not define it");
         }
-        var conflicts = clusterEndpointConfigProvider.getExclusivePorts().stream().filter(p -> clusterEndpointConfigProvider.getSharedPorts().contains(p))
+        var conflicts = clusterNetworkAddressConfigProvider.getExclusivePorts().stream().filter(p -> clusterNetworkAddressConfigProvider.getSharedPorts().contains(p))
                 .collect(Collectors.toSet());
         if (!conflicts.isEmpty()) {
             throw new IllegalStateException(
@@ -58,15 +60,15 @@ public class VirtualCluster implements ClusterEndpointConfigProvider {
         this.logFrames = logFrames;
         this.keyStoreFile = keyStoreFile;
         this.keyPassword = keyPassword;
-        this.endpointProvider = clusterEndpointConfigProvider;
+        this.provider = clusterNetworkAddressConfigProvider;
     }
 
     public TargetCluster targetCluster() {
         return targetCluster;
     }
 
-    public ClusterEndpointConfigProvider getClusterEndpointProvider() {
-        return endpointProvider;
+    public ClusterNetworkAddressConfigProvider getClusterNetworkAddressConfigProvider() {
+        return provider;
     }
 
     public Optional<String> keyStoreFile() {
@@ -110,7 +112,7 @@ public class VirtualCluster implements ClusterEndpointConfigProvider {
     public String toString() {
         final StringBuilder sb = new StringBuilder("VirtualCluster [");
         sb.append("targetCluster=").append(targetCluster);
-        sb.append(", clusterEndpointProvider=").append(endpointProvider);
+        sb.append(", clusterNetworkAddressConfigProvider=").append(provider);
         sb.append(", keyStoreFile=").append(keyStoreFile);
         sb.append(", keyPassword=").append(keyPassword);
         sb.append(", logNetwork=").append(logNetwork);
@@ -121,31 +123,31 @@ public class VirtualCluster implements ClusterEndpointConfigProvider {
 
     @Override
     public HostPort getClusterBootstrapAddress() {
-        return endpointProvider.getClusterBootstrapAddress();
+        return provider.getClusterBootstrapAddress();
     }
 
     @Override
     public HostPort getBrokerAddress(int nodeId) throws IllegalArgumentException {
-        return endpointProvider.getBrokerAddress(nodeId);
+        return provider.getBrokerAddress(nodeId);
     }
 
     @Override
     public Optional<String> getBindAddress() {
-        return endpointProvider.getBindAddress();
+        return provider.getBindAddress();
     }
 
     @Override
     public boolean requiresTls() {
-        return endpointProvider.requiresTls();
+        return provider.requiresTls();
     }
 
     @Override
     public Set<Integer> getExclusivePorts() {
-        return endpointProvider.getExclusivePorts();
+        return provider.getExclusivePorts();
     }
 
     @Override
     public Set<Integer> getSharedPorts() {
-        return endpointProvider.getSharedPorts();
+        return provider.getSharedPorts();
     }
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/BuiltinClusterEndpointProviderContributor.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/BuiltinClusterEndpointProviderContributor.java
deleted file mode 100644
index 74158418..00000000
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/BuiltinClusterEndpointProviderContributor.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-package io.kroxylicious.proxy.internal.clusterendpointprovider;
-
-import io.kroxylicious.proxy.clusterendpointprovider.ClusterEndpointProviderContributor;
-import io.kroxylicious.proxy.internal.clusterendpointprovider.PortPerBrokerClusterEndpointConfigProvider.PortPerBrokerClusterEndpointProviderConfig;
-import io.kroxylicious.proxy.internal.clusterendpointprovider.SniRoutingClusterEndpointConfigProvider.SniRoutingClusterEndpointProviderConfig;
-import io.kroxylicious.proxy.service.BaseContributor;
-import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
-
-public class BuiltinClusterEndpointProviderContributor extends BaseContributor<ClusterEndpointConfigProvider> implements ClusterEndpointProviderContributor {
-
-    public static final BaseContributorBuilder<ClusterEndpointConfigProvider> FILTERS = BaseContributor.<ClusterEndpointConfigProvider> builder()
-            .add("PortPerBroker", PortPerBrokerClusterEndpointProviderConfig.class, PortPerBrokerClusterEndpointConfigProvider::new)
-            .add("SniRouting", SniRoutingClusterEndpointProviderConfig.class, SniRoutingClusterEndpointConfigProvider::new);
-
-    public BuiltinClusterEndpointProviderContributor() {
-        super(FILTERS);
-    }
-}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/ClusterEndpointConfigProviderContributorManager.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/ClusterEndpointConfigProviderContributorManager.java
deleted file mode 100644
index b5e08607..00000000
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/ClusterEndpointConfigProviderContributorManager.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * Copyright Kroxylicious Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-package io.kroxylicious.proxy.internal.clusterendpointprovider;
-
-import java.util.Iterator;
-import java.util.ServiceLoader;
-
-import io.kroxylicious.proxy.clusterendpointprovider.ClusterEndpointProviderContributor;
-import io.kroxylicious.proxy.config.BaseConfig;
-import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
-
-public class ClusterEndpointConfigProviderContributorManager {
-
-    private static final ClusterEndpointConfigProviderContributorManager INSTANCE = new ClusterEndpointConfigProviderContributorManager();
-
-    private final ServiceLoader<ClusterEndpointProviderContributor> contributors;
-
-    private ClusterEndpointConfigProviderContributorManager() {
-        this.contributors = ServiceLoader.load(ClusterEndpointProviderContributor.class);
-    }
-
-    public static ClusterEndpointConfigProviderContributorManager getInstance() {
-        return INSTANCE;
-    }
-
-    public Class<? extends BaseConfig> getConfigType(String shortName) {
-        Iterator<ClusterEndpointProviderContributor> it = contributors.iterator();
-        while (it.hasNext()) {
-            ClusterEndpointProviderContributor contributor = it.next();
-            Class<? extends BaseConfig> configType = contributor.getConfigType(shortName);
-            if (configType != null) {
-                return configType;
-            }
-        }
-
-        throw new IllegalArgumentException("No endpoint provider found for name '" + shortName + "'");
-    }
-
-    public ClusterEndpointConfigProvider getClusterEndpointConfigProvider(String shortName, BaseConfig baseConfig) {
-        for (ClusterEndpointProviderContributor contributor : contributors) {
-            var assigner = contributor.getInstance(shortName, baseConfig);
-            if (assigner != null) {
-                return assigner;
-            }
-        }
-
-        throw new IllegalArgumentException("No endpoint provider found for name '" + shortName + "'");
-    }
-}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/BuiltinClusterNetworkAddressConfigProviderContributor.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/BuiltinClusterNetworkAddressConfigProviderContributor.java
new file mode 100644
index 00000000..b06cefb7
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/BuiltinClusterNetworkAddressConfigProviderContributor.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider;
+
+import io.kroxylicious.proxy.clusternetworkaddressconfigprovider.ClusterNetworkAddressConfigProviderContributor;
+import io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.PortPerBrokerClusterNetworkAddressConfigProvider.PortPerBrokerClusterNetworkAddressConfigProviderConfig;
+import io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.SniRoutingClusterNetworkAddressConfigProvider.SniRoutingClusterNetworkAddressConfigProviderConfig;
+import io.kroxylicious.proxy.service.BaseContributor;
+import io.kroxylicious.proxy.service.ClusterNetworkAddressConfigProvider;
+
+public class BuiltinClusterNetworkAddressConfigProviderContributor extends BaseContributor<ClusterNetworkAddressConfigProvider>
+        implements ClusterNetworkAddressConfigProviderContributor {
+
+    public static final BaseContributorBuilder<ClusterNetworkAddressConfigProvider> FILTERS = BaseContributor.<ClusterNetworkAddressConfigProvider> builder()
+            .add("PortPerBroker", PortPerBrokerClusterNetworkAddressConfigProviderConfig.class, PortPerBrokerClusterNetworkAddressConfigProvider::new)
+            .add("SniRouting", SniRoutingClusterNetworkAddressConfigProviderConfig.class, SniRoutingClusterNetworkAddressConfigProvider::new);
+
+    public BuiltinClusterNetworkAddressConfigProviderContributor() {
+        super(FILTERS);
+    }
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/ClusterNetworkAddressConfigProviderContributorManager.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/ClusterNetworkAddressConfigProviderContributorManager.java
new file mode 100644
index 00000000..745ac09b
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/ClusterNetworkAddressConfigProviderContributorManager.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.internal.clusternetworkaddressconfigprovider;
+
+import java.util.Iterator;
+import java.util.ServiceLoader;
+
+import io.kroxylicious.proxy.clusternetworkaddressconfigprovider.ClusterNetworkAddressConfigProviderContributor;
+import io.kroxylicious.proxy.config.BaseConfig;
+import io.kroxylicious.proxy.service.ClusterNetworkAddressConfigProvider;
+
+public class ClusterNetworkAddressConfigProviderContributorManager {
+
+    private static final ClusterNetworkAddressConfigProviderContributorManager INSTANCE = new ClusterNetworkAddressConfigProviderContributorManager();
+
+    private final ServiceLoader<ClusterNetworkAddressConfigProviderContributor> contributors;
+
+    private ClusterNetworkAddressConfigProviderContributorManager() {
+        this.contributors = ServiceLoader.load(ClusterNetworkAddressConfigProviderContributor.class);
+    }
+
+    public static ClusterNetworkAddressConfigProviderContributorManager getInstance() {
+        return INSTANCE;
+    }
+
+    public Class<? extends BaseConfig> getConfigType(String shortName) {
+        Iterator<ClusterNetworkAddressConfigProviderContributor> it = contributors.iterator();
+        while (it.hasNext()) {
+            ClusterNetworkAddressConfigProviderContributor contributor = it.next();
+            Class<? extends BaseConfig> configType = contributor.getConfigType(shortName);
+            if (configType != null) {
+                return configType;
+            }
+        }
+
+        throw new IllegalArgumentException("No endpoint provider found for name '" + shortName + "'");
+    }
+
+    public ClusterNetworkAddressConfigProvider getClusterEndpointConfigProvider(String shortName, BaseConfig baseConfig) {
+        for (ClusterNetworkAddressConfigProviderContributor contributor : contributors) {
+            var assigner = contributor.getInstance(shortName, baseConfig);
+            if (assigner != null) {
+                return assigner;
+            }
+        }
+
+        throw new IllegalArgumentException("No endpoint provider found for name '" + shortName + "'");
+    }
+}
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/PortPerBrokerClusterEndpointConfigProvider.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProvider.java
similarity index 84%
rename from kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/PortPerBrokerClusterEndpointConfigProvider.java
rename to kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProvider.java
index 5924ce05..4b662a19 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/PortPerBrokerClusterEndpointConfigProvider.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProvider.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious.proxy.internal.clusterendpointprovider;
+package io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider;
 
 import java.util.Collections;
 import java.util.HashSet;
@@ -17,11 +17,11 @@ import java.util.stream.IntStream;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 import io.kroxylicious.proxy.config.BaseConfig;
-import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
+import io.kroxylicious.proxy.service.ClusterNetworkAddressConfigProvider;
 import io.kroxylicious.proxy.service.HostPort;
 
 /**
- * A ClusterEndpointConfigProvider implementation that uses a separate port per broker endpoint.
+ * A ClusterNetworkAddressConfigProvider implementation that uses a separate port per broker endpoint.
  * <br/>
  * The following configuration is supported:
  * <ul>
@@ -34,7 +34,7 @@ import io.kroxylicious.proxy.service.HostPort;
  *    <li>{@code numberOfBrokerPorts} (optional) defines the maximum number of broker ports that will be permitted. If omitted, it is defaulted to {$code 3}.</li>
  * </ul>
  */
-public class PortPerBrokerClusterEndpointConfigProvider implements ClusterEndpointConfigProvider {
+public class PortPerBrokerClusterNetworkAddressConfigProvider implements ClusterNetworkAddressConfigProvider {
 
     private final HostPort bootstrapAddress;
 
@@ -52,7 +52,7 @@ public class PortPerBrokerClusterEndpointConfigProvider implements ClusterEndpoi
      *
      * @param config configuration
      */
-    public PortPerBrokerClusterEndpointConfigProvider(PortPerBrokerClusterEndpointProviderConfig config) {
+    public PortPerBrokerClusterNetworkAddressConfigProvider(PortPerBrokerClusterNetworkAddressConfigProviderConfig config) {
         this.bootstrapAddress = config.bootstrapAddress;
         this.brokerAddressPattern = config.brokerAddressPattern;
         this.brokerStartPort = config.brokerStartPort;
@@ -94,16 +94,16 @@ public class PortPerBrokerClusterEndpointConfigProvider implements ClusterEndpoi
     /**
      * Creates the configuration for this provider.
      */
-    public static class PortPerBrokerClusterEndpointProviderConfig extends BaseConfig {
+    public static class PortPerBrokerClusterNetworkAddressConfigProviderConfig extends BaseConfig {
         private final HostPort bootstrapAddress;
         private final String brokerAddressPattern;
         private final int brokerStartPort;
         private final int numberOfBrokerPorts;
 
-        public PortPerBrokerClusterEndpointProviderConfig(@JsonProperty(required = true) HostPort bootstrapAddress,
-                                                          @JsonProperty(required = false) String brokerAddressPattern,
-                                                          @JsonProperty(required = false) Integer brokerStartPort,
-                                                          @JsonProperty(required = false, defaultValue = "3") Integer numberOfBrokerPorts) {
+        public PortPerBrokerClusterNetworkAddressConfigProviderConfig(@JsonProperty(required = true) HostPort bootstrapAddress,
+                                                                      @JsonProperty(required = false) String brokerAddressPattern,
+                                                                      @JsonProperty(required = false) Integer brokerStartPort,
+                                                                      @JsonProperty(required = false, defaultValue = "3") Integer numberOfBrokerPorts) {
             Objects.requireNonNull(bootstrapAddress, "bootstrapAddress cannot be null");
 
             this.bootstrapAddress = bootstrapAddress;
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/SniRoutingClusterEndpointConfigProvider.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProvider.java
similarity index 80%
rename from kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/SniRoutingClusterEndpointConfigProvider.java
rename to kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProvider.java
index 8cdb227d..e9a44ece 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/SniRoutingClusterEndpointConfigProvider.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProvider.java
@@ -4,21 +4,21 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious.proxy.internal.clusterendpointprovider;
+package io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider;
 
 import java.util.Set;
 import java.util.regex.Pattern;
 
 import io.kroxylicious.proxy.config.BaseConfig;
-import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
+import io.kroxylicious.proxy.service.ClusterNetworkAddressConfigProvider;
 import io.kroxylicious.proxy.service.HostPort;
 
 /**
- * A ClusterEndpointConfigProvider implementation that uses a single port for bootstrap and
+ * A ClusterNetworkAddressConfigProvider implementation that uses a single port for bootstrap and
  * all brokers.  SNI information is used to route the connection to the correct target.
  *
  */
-public class SniRoutingClusterEndpointConfigProvider implements ClusterEndpointConfigProvider {
+public class SniRoutingClusterNetworkAddressConfigProvider implements ClusterNetworkAddressConfigProvider {
 
     public static final String LITERAL_NODE_ID = "$(nodeId)";
     private static final Pattern NODE_ID_TOKEN_RE = Pattern.compile(Pattern.quote(LITERAL_NODE_ID));
@@ -30,7 +30,7 @@ public class SniRoutingClusterEndpointConfigProvider implements ClusterEndpointC
      *
      * @param config configuration
      */
-    public SniRoutingClusterEndpointConfigProvider(SniRoutingClusterEndpointProviderConfig config) {
+    public SniRoutingClusterNetworkAddressConfigProvider(SniRoutingClusterNetworkAddressConfigProviderConfig config) {
         this.bootstrapAddress = config.bootstrapAddress;
         this.brokerAddressPattern = config.brokerAddressPattern;
     }
@@ -63,12 +63,12 @@ public class SniRoutingClusterEndpointConfigProvider implements ClusterEndpointC
     /**
      * Creates the configuration for this provider.
      */
-    public static class SniRoutingClusterEndpointProviderConfig extends BaseConfig {
+    public static class SniRoutingClusterNetworkAddressConfigProviderConfig extends BaseConfig {
 
         private final HostPort bootstrapAddress;
         private final String brokerAddressPattern;
 
-        public SniRoutingClusterEndpointProviderConfig(HostPort bootstrapAddress, String brokerAddressPattern) {
+        public SniRoutingClusterNetworkAddressConfigProviderConfig(HostPort bootstrapAddress, String brokerAddressPattern) {
             if (bootstrapAddress == null) {
                 throw new IllegalArgumentException("bootstrapAddress cannot be null");
             }
diff --git a/kroxylicious/src/main/resources/META-INF/services/io.kroxylicious.proxy.clusterendpointprovider.ClusterEndpointProviderContributor b/kroxylicious/src/main/resources/META-INF/services/io.kroxylicious.proxy.clusterendpointprovider.ClusterEndpointProviderContributor
deleted file mode 100644
index 9b78922e..00000000
--- a/kroxylicious/src/main/resources/META-INF/services/io.kroxylicious.proxy.clusterendpointprovider.ClusterEndpointProviderContributor
+++ /dev/null
@@ -1,1 +0,0 @@
-io.kroxylicious.proxy.internal.clusterendpointprovider.BuiltinClusterEndpointProviderContributor
diff --git a/kroxylicious/src/main/resources/META-INF/services/io.kroxylicious.proxy.clusternetworkaddressconfigprovider.ClusterNetworkAddressConfigProviderContributor b/kroxylicious/src/main/resources/META-INF/services/io.kroxylicious.proxy.clusternetworkaddressconfigprovider.ClusterNetworkAddressConfigProviderContributor
new file mode 100644
index 00000000..7d83e4b4
--- /dev/null
+++ b/kroxylicious/src/main/resources/META-INF/services/io.kroxylicious.proxy.clusternetworkaddressconfigprovider.ClusterNetworkAddressConfigProviderContributor
@@ -0,0 +1,1 @@
+io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.BuiltinClusterNetworkAddressConfigProviderContributor


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/ExpositionIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/ExpositionIT.java
index 9044bb05..8d81578e 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/ExpositionIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/ExpositionIT.java
@@ -77,10 +77,10 @@ public class ExpositionIT {
                         .withNewTargetCluster()
                         .withBootstrapServers(bootstrapServers)
                         .endTargetCluster()
-                        .withNewClusterEndpointConfigProvider()
+                        .withNewClusterNetworkAddressConfigProvider()
                         .withType("PortPerBroker")
                         .withConfig(Map.of("bootstrapAddress", PROXY_ADDRESS.toString()))
-                        .endClusterEndpointConfigProvider()
+                        .endClusterNetworkAddressConfigProvider()
                         .build())
                 .addNewFilter().withType("ApiVersions").endFilter();
 
@@ -122,10 +122,10 @@ public class ExpositionIT {
         for (int i = 0; i < clusterProxyAddresses.size(); i++) {
             var bootstrap = clusterProxyAddresses.get(i);
             var virtualCluster = new VirtualClusterBuilder(base)
-                    .withNewClusterEndpointConfigProvider()
+                    .withNewClusterNetworkAddressConfigProvider()
                     .withType("PortPerBroker")
                     .withConfig(Map.of("bootstrapAddress", bootstrap))
-                    .endClusterEndpointConfigProvider()
+                    .endClusterNetworkAddressConfigProvider()
                     .build();
             builder.addToVirtualClusters("cluster" + i, virtualCluster);
 
@@ -169,11 +169,11 @@ public class ExpositionIT {
             clientTrust.add(new ClientTrust(clientTrustStore, brokerCertificateGenerator.getPassword()));
 
             var virtualCluster = new VirtualClusterBuilder(base)
-                    .withNewClusterEndpointConfigProvider()
+                    .withNewClusterNetworkAddressConfigProvider()
                     .withType("SniRouting")
                     .withConfig(Map.of("bootstrapAddress", virtualClusterFQDN + ":9192",
                             "brokerAddressPattern", virtualClusterBrokerAddressPattern.formatted(i)))
-                    .endClusterEndpointConfigProvider()
+                    .endClusterNetworkAddressConfigProvider()
                     .withKeyPassword(brokerCertificateGenerator.getPassword())
                     .withKeyStoreFile(brokerCertificateGenerator.getKeyStoreLocation())
                     .withLogNetwork(true)
@@ -205,10 +205,10 @@ public class ExpositionIT {
                         .withNewTargetCluster()
                         .withBootstrapServers(cluster.getBootstrapServers())
                         .endTargetCluster()
-                        .withNewClusterEndpointConfigProvider()
+                        .withNewClusterNetworkAddressConfigProvider()
                         .withType("PortPerBroker")
                         .withConfig(Map.of("bootstrapAddress", PROXY_ADDRESS.toString()))
-                        .endClusterEndpointConfigProvider()
+                        .endClusterNetworkAddressConfigProvider()
                         .build())
                 .addNewFilter().withType("ApiVersions").endFilter();
 
@@ -234,10 +234,10 @@ public class ExpositionIT {
                         .withNewTargetCluster()
                         .withBootstrapServers(cluster.getBootstrapServers())
                         .endTargetCluster()
-                        .withNewClusterEndpointConfigProvider()
+                        .withNewClusterNetworkAddressConfigProvider()
                         .withType("PortPerBroker")
                         .withConfig(Map.of("bootstrapAddress", PROXY_ADDRESS.toString()))
-                        .endClusterEndpointConfigProvider()
+                        .endClusterNetworkAddressConfigProvider()
                         .build())
                 .addNewFilter().withType("ApiVersions").endFilter();
 
@@ -268,10 +268,10 @@ public class ExpositionIT {
                         .withNewTargetCluster()
                         .withBootstrapServers(cluster.getBootstrapServers())
                         .endTargetCluster()
-                        .withNewClusterEndpointConfigProvider()
+                        .withNewClusterNetworkAddressConfigProvider()
                         .withType("PortPerBroker")
                         .withConfig(Map.of("bootstrapAddress", PROXY_ADDRESS.toString()))
-                        .endClusterEndpointConfigProvider()
+                        .endClusterNetworkAddressConfigProvider()
                         .build())
                 .addNewFilter().withType("ApiVersions").endFilter();
 
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
index 69b02940..453592c9 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
@@ -21,7 +21,6 @@ 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.serialization.Serdes;
-import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -39,6 +38,8 @@ import static org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET
 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.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
@@ -128,7 +129,7 @@ public class KrpcFilterIT {
 
         try (var tester = kroxyliciousTester(config);
                 var proxyAdmin = tester.admin()) {
-            Assertions.assertThatExceptionOfType(ExecutionException.class)
+            assertThatExceptionOfType(ExecutionException.class)
                     .isThrownBy(() -> proxyAdmin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1))).all().get())
                     .withCauseInstanceOf(InvalidTopicException.class)
                     .havingCause()
@@ -139,7 +140,7 @@ public class KrpcFilterIT {
             // remove once https://github.com/kroxylicious/kroxylicious-junit5-extension/issues/114 is fixed.
             names = new HashSet<>(names);
             names.removeIf(n -> n.startsWith("__org_kroxylicious_testing"));
-            assertEquals(Set.of(), names);
+            assertThat(names).isEmpty();
         }
     }
 
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
index 0a2ddadf..ee719d09 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
@@ -583,10 +583,10 @@ public class MultiTenantIT {
                         .withNewTargetCluster()
                         .withBootstrapServers(cluster.getBootstrapServers())
                         .endTargetCluster()
-                        .withNewClusterEndpointConfigProvider()
+                        .withNewClusterNetworkAddressConfigProvider()
                         .withType("PortPerBroker")
                         .withConfig(Map.of("bootstrapAddress", TENANT_1_PROXY_ADDRESS.toString()))
-                        .endClusterEndpointConfigProvider()
+                        .endClusterNetworkAddressConfigProvider()
                         .withKeyPassword(certificateGenerator.getPassword())
                         .withKeyStoreFile(certificateGenerator.getKeyStoreLocation())
                         .build())
@@ -594,10 +594,10 @@ public class MultiTenantIT {
                         .withNewTargetCluster()
                         .withBootstrapServers(cluster.getBootstrapServers())
                         .endTargetCluster()
-                        .withNewClusterEndpointConfigProvider()
+                        .withNewClusterNetworkAddressConfigProvider()
                         .withType("PortPerBroker")
                         .withConfig(Map.of("bootstrapAddress", TENANT_2_PROXY_ADDRESS.toString()))
-                        .endClusterEndpointConfigProvider()
+                        .endClusterNetworkAddressConfigProvider()
                         .withKeyPassword(certificateGenerator.getPassword())
                         .withKeyStoreFile(certificateGenerator.getKeyStoreLocation())
                         .build())
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/proxy/ClusterEndpointConfigProvider.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/proxy/ClusterNetworkAddressConfigProvider.java
similarity index 78%
rename from kroxylicious-test-tools/src/main/java/io/kroxylicious/proxy/ClusterEndpointConfigProvider.java
rename to kroxylicious-test-tools/src/main/java/io/kroxylicious/proxy/ClusterNetworkAddressConfigProvider.java
index 85df3baa..1d012438 100644
--- a/kroxylicious-test-tools/src/main/java/io/kroxylicious/proxy/ClusterEndpointConfigProvider.java
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/proxy/ClusterNetworkAddressConfigProvider.java
@@ -15,5 +15,5 @@ import io.sundr.builder.annotations.Buildable;
 import static com.fasterxml.jackson.annotation.JsonInclude.Include.NON_EMPTY;
 
 @Buildable(editableEnabled = false)
-public record ClusterEndpointConfigProvider(String type, @JsonInclude(NON_EMPTY) Map<String, Object> config) {
+public record ClusterNetworkAddressConfigProvider(String type, @JsonInclude(NON_EMPTY) Map<String, Object> config) {
 }
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/proxy/VirtualCluster.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/proxy/VirtualCluster.java
index 8ac08a8d..e1db1452 100644
--- a/kroxylicious-test-tools/src/main/java/io/kroxylicious/proxy/VirtualCluster.java
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/proxy/VirtualCluster.java
@@ -9,5 +9,5 @@ package io.kroxylicious.proxy;
 import io.sundr.builder.annotations.Buildable;
 
 @Buildable(editableEnabled = false)
-public record VirtualCluster(TargetCluster targetCluster, ClusterEndpointConfigProvider clusterEndpointConfigProvider, String keyStoreFile, String keyPassword, boolean logNetwork, boolean logFrames) {
+public record VirtualCluster(TargetCluster targetCluster, ClusterNetworkAddressConfigProvider clusterNetworkAddressConfigProvider, String keyStoreFile, String keyPassword, boolean logNetwork, boolean logFrames) {
 }
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousConfigUtils.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousConfigUtils.java
index 6102d26c..4ed454b8 100644
--- a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousConfigUtils.java
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousConfigUtils.java
@@ -8,7 +8,7 @@ package io.kroxylicious.test.tester;
 
 import java.util.Map;
 
-import io.kroxylicious.proxy.ClusterEndpointConfigProvider;
+import io.kroxylicious.proxy.ClusterNetworkAddressConfigProvider;
 import io.kroxylicious.proxy.KroxyliciousConfig;
 import io.kroxylicious.proxy.KroxyliciousConfigBuilder;
 import io.kroxylicious.proxy.VirtualCluster;
@@ -35,10 +35,10 @@ public class KroxyliciousConfigUtils {
                 .withNewTargetCluster()
                 .withBootstrapServers(clusterBootstrapServers)
                 .endTargetCluster()
-                .withNewClusterEndpointConfigProvider()
+                .withNewClusterNetworkAddressConfigProvider()
                 .withType("PortPerBroker")
                 .withConfig(Map.of("bootstrapAddress", DEFAULT_PROXY_BOOTSTRAP.toString()))
-                .endClusterEndpointConfigProvider()
+                .endClusterNetworkAddressConfigProvider()
                 .build());
     }
 
@@ -74,7 +74,7 @@ public class KroxyliciousConfigUtils {
         if (cluster == null) {
             throw new IllegalArgumentException("virtualCluster " + virtualCluster + " not found in config: " + config);
         }
-        ClusterEndpointConfigProvider provider = cluster.clusterEndpointConfigProvider();
+        ClusterNetworkAddressConfigProvider provider = cluster.clusterNetworkAddressConfigProvider();
         if (provider.type().equals("PortPerBroker") || provider.type().equals("SniRouting")) {
             Object bootstrapAddress = provider.config().get("bootstrapAddress");
             return (String) bootstrapAddress;
diff --git a/kroxylicious-test-tools/src/test/java/io/kroxylicious/proxy/KroxyliciousConfigTest.java b/kroxylicious-test-tools/src/test/java/io/kroxylicious/proxy/KroxyliciousConfigTest.java
index 3afcc701..5b6ccc30 100644
--- a/kroxylicious-test-tools/src/test/java/io/kroxylicious/proxy/KroxyliciousConfigTest.java
+++ b/kroxylicious-test-tools/src/test/java/io/kroxylicious/proxy/KroxyliciousConfigTest.java
@@ -34,17 +34,17 @@ public class KroxyliciousConfigTest {
     }
 
     @Test
-    public void testVirtualClusterConfigWithClusterEndpointProvider() throws Exception {
+    public void testVirtualClusterConfigWithClusterNetworkAddressConfigProvider() throws Exception {
         ObjectNode deserializedConfig = serializeAndDeserialize(
                 builder().addToVirtualClusters("demo", new VirtualClusterBuilder()
-                        .withNewClusterEndpointConfigProvider()
+                        .withNewClusterNetworkAddressConfigProvider()
                         .withType("providertype")
                         .withConfig(Map.of("a", "b"))
-                        .endClusterEndpointConfigProvider()
+                        .endClusterNetworkAddressConfigProvider()
                         .build()).build());
         ObjectNode clusterObj = assertObjectField(deserializedConfig, "virtualClusters");
         ObjectNode demoObj = assertObjectField(clusterObj, "demo");
-        ObjectNode endpointAssignerObj = assertObjectField(demoObj, "clusterEndpointConfigProvider");
+        ObjectNode endpointAssignerObj = assertObjectField(demoObj, "clusterNetworkAddressConfigProvider");
         assertTextField(endpointAssignerObj, "type", "providertype");
         ObjectNode endpointAssignerConfig = assertObjectField(endpointAssignerObj, "config");
         assertTextField(endpointAssignerConfig, "a", "b");
diff --git a/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/tester/KroxyliciousTestersTest.java b/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/tester/KroxyliciousTestersTest.java
index 7ab4528f..a20cd502 100644
--- a/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/tester/KroxyliciousTestersTest.java
+++ b/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/tester/KroxyliciousTestersTest.java
@@ -184,10 +184,10 @@ public class KroxyliciousTestersTest {
                 .withNewTargetCluster()
                 .withBootstrapServers(clusterBootstrapServers)
                 .endTargetCluster()
-                .withNewClusterEndpointConfigProvider()
+                .withNewClusterNetworkAddressConfigProvider()
                 .withType("PortPerBroker")
                 .withConfig(Map.of("bootstrapAddress", defaultProxyBootstrap))
-                .endClusterEndpointConfigProvider()
+                .endClusterNetworkAddressConfigProvider()
                 .build());
     }
 
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/KafkaProxyTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/KafkaProxyTest.java
index d5824908..6f3cb20d 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/KafkaProxyTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/KafkaProxyTest.java
@@ -26,7 +26,7 @@ class KafkaProxyTest {
                   demo1:
                     targetCluster:
                       bootstrap_servers: kafka.example:1234
-                    clusterEndpointConfigProvider:
+                    clusterNetworkAddressConfigProvider:
                       type: PortPerBroker
                       config:
                         bootstrapAddress: localhost:9192
@@ -34,7 +34,7 @@ class KafkaProxyTest {
                   demo2:
                     targetCluster:
                       bootstrap_servers: kafka.example:1234
-                    clusterEndpointConfigProvider:
+                    clusterNetworkAddressConfigProvider:
                       type: PortPerBroker
                       config:
                         bootstrapAddress: localhost:9192 # Conflict
@@ -45,7 +45,7 @@ class KafkaProxyTest {
                           demo1:
                             targetCluster:
                               bootstrap_servers: kafka.example:1234
-                            clusterEndpointConfigProvider:
+                            clusterNetworkAddressConfigProvider:
                               type: PortPerBroker
                               config:
                                 bootstrapAddress: localhost:9192
@@ -54,7 +54,7 @@ class KafkaProxyTest {
                           demo2:
                             targetCluster:
                               bootstrap_servers: kafka.example:1234
-                            clusterEndpointConfigProvider:
+                            clusterNetworkAddressConfigProvider:
                               type: PortPerBroker
                               config:
                                 bootstrapAddress: localhost:8192
@@ -66,14 +66,14 @@ class KafkaProxyTest {
                           demo1:
                             targetCluster:
                               bootstrap_servers: kafka.example:1234
-                            clusterEndpointConfigProvider:
+                            clusterNetworkAddressConfigProvider:
                               type: PortPerBroker
                               config:
                                 bootstrapAddress: localhost:9192
                           demo2:
                             targetCluster:
                               bootstrap_servers: kafka.example:1234
-                            clusterEndpointConfigProvider:
+                            clusterNetworkAddressConfigProvider:
                               type: SniRouting
                               config:
                                 bootstrapAddress: localhost:9192
@@ -96,13 +96,13 @@ class KafkaProxyTest {
         return Stream.of(Arguments.of("tls mismatch", """
                 virtualClusters:
                   demo1:
-                    targetCluster:
-                      bootstrap_servers: kafka.example:1234
-                    clusterEndpointConfigProvider:
+                    clusterNetworkAddressConfigProvider:
                       type: SniRouting
                       config:
                         bootstrapAddress: cluster1:9192
                         brokerAddressPattern:  broker-$(nodeId)
+                    targetCluster:
+                      bootstrap_servers: kafka.example:1234
                 """, "Cluster endpoint provider requires tls, but this virtual cluster does not define it"));
     }
 
@@ -124,7 +124,7 @@ class KafkaProxyTest {
                           demo1:
                             targetCluster:
                               bootstrap_servers: kafka.invalid:1234
-                            clusterEndpointConfigProvider:
+                            clusterNetworkAddressConfigProvider:
                               type: SniRouting
                               config:
                                 bootstrapAddress: cluster1:9192
@@ -134,7 +134,7 @@ class KafkaProxyTest {
                           demo2:
                             targetCluster:
                               bootstrap_servers: kafka.invalid:1234
-                            clusterEndpointConfigProvider:
+                            clusterNetworkAddressConfigProvider:
                               type: SniRouting
                               config:
                                 bootstrapAddress: cluster2:9192
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusterendpointprovider/PortPerBrokerClusterEndpointConfigProviderTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProviderTest.java
similarity index 61%
rename from kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusterendpointprovider/PortPerBrokerClusterEndpointConfigProviderTest.java
rename to kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProviderTest.java
index 2ea388c9..7d911c36 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusterendpointprovider/PortPerBrokerClusterEndpointConfigProviderTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/PortPerBrokerClusterNetworkAddressConfigProviderTest.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious.proxy.internal.clusterendpointprovider;
+package io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -13,19 +13,19 @@ import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.EmptySource;
 import org.junit.jupiter.params.provider.ValueSource;
 
-import io.kroxylicious.proxy.internal.clusterendpointprovider.PortPerBrokerClusterEndpointConfigProvider.PortPerBrokerClusterEndpointProviderConfig;
+import io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.PortPerBrokerClusterNetworkAddressConfigProvider.PortPerBrokerClusterNetworkAddressConfigProviderConfig;
 
 import static io.kroxylicious.proxy.service.HostPort.parse;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
-class PortPerBrokerClusterEndpointConfigProviderTest {
+class PortPerBrokerClusterNetworkAddressConfigProviderTest {
 
     @ParameterizedTest
     @CsvSource({ "0,1", "-1,1", "4,-1", "10,0" })
     void badBrokerPortDefinition(int brokerStartPort, int numberOfBrokerPorts) {
         Assertions.assertThrows(IllegalArgumentException.class, () -> {
-            new PortPerBrokerClusterEndpointProviderConfig(parse("localhost:1235"), "localhost:$(portNumber)", brokerStartPort, numberOfBrokerPorts);
+            new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"), "localhost:$(portNumber)", brokerStartPort, numberOfBrokerPorts);
         });
     }
 
@@ -33,7 +33,7 @@ class PortPerBrokerClusterEndpointConfigProviderTest {
     @CsvSource({ "1235,1", "1234,3" })
     void bootstrapBrokerAddressCollision(int brokerStartPort, int numberOfBrokerPorts) {
         Assertions.assertThrows(IllegalArgumentException.class, () -> {
-            new PortPerBrokerClusterEndpointProviderConfig(parse("localhost:1235"), "localhost:$(portNumber)", brokerStartPort, numberOfBrokerPorts);
+            new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"), "localhost:$(portNumber)", brokerStartPort, numberOfBrokerPorts);
         });
     }
 
@@ -42,13 +42,13 @@ class PortPerBrokerClusterEndpointConfigProviderTest {
             "not.at.$(portNumber).end:1234" })
     @EmptySource
     void invalidBrokerPattern(String input) {
-        assertThrows(IllegalArgumentException.class, () -> new PortPerBrokerClusterEndpointProviderConfig(parse("good:1235"), input, 1, 5));
+        assertThrows(IllegalArgumentException.class, () -> new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("good:1235"), input, 1, 5));
     }
 
     @Test
     void portsExhausted() {
-        var provider = new PortPerBrokerClusterEndpointConfigProvider(
-                new PortPerBrokerClusterEndpointProviderConfig(parse("localhost:1235"), "localhost:$(portNumber)", 1236, 1));
+        var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
+                new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"), "localhost:$(portNumber)", 1236, 1));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("localhost:1236"));
         Assertions.assertThrows(IllegalArgumentException.class, () -> {
             provider.getBrokerAddress(1);
@@ -57,24 +57,24 @@ class PortPerBrokerClusterEndpointConfigProviderTest {
 
     @Test
     void defaultsBrokerPatternBasedOnBootstrapHost() {
-        var provider = new PortPerBrokerClusterEndpointConfigProvider(
-                new PortPerBrokerClusterEndpointProviderConfig(parse("mycluster:1235"), null, 1236, 1237));
+        var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
+                new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("mycluster:1235"), null, 1236, 1237));
         assertThat(provider.getClusterBootstrapAddress()).isEqualTo(parse("mycluster:1235"));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("mycluster:1236"));
     }
 
     @Test
     void defaultsBrokerStartPortBasedOnBootstrapPort() {
-        var provider = new PortPerBrokerClusterEndpointConfigProvider(
-                new PortPerBrokerClusterEndpointProviderConfig(parse("mycluster:1235"), null, null, 1237));
+        var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
+                new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("mycluster:1235"), null, null, 1237));
         assertThat(provider.getClusterBootstrapAddress()).isEqualTo(parse("mycluster:1235"));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("mycluster:1236"));
     }
 
     @Test
     void defaultsNumberOfBrokerPorts() {
-        var provider = new PortPerBrokerClusterEndpointConfigProvider(
-                new PortPerBrokerClusterEndpointProviderConfig(parse("mycluster:1235"), null, null, null));
+        var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
+                new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("mycluster:1235"), null, null, null));
         assertThat(provider.getClusterBootstrapAddress()).isEqualTo(parse("mycluster:1235"));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("mycluster:1236"));
         assertThat(provider.getBrokerAddress(1)).isEqualTo(parse("mycluster:1237"));
@@ -86,16 +86,16 @@ class PortPerBrokerClusterEndpointConfigProviderTest {
 
     @Test
     void definesExclusiveAndSharedCorrectly() {
-        var provider = new PortPerBrokerClusterEndpointConfigProvider(
-                new PortPerBrokerClusterEndpointProviderConfig(parse("localhost:1235"), "localhost:$(portNumber)", 1236, 2));
+        var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
+                new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"), "localhost:$(portNumber)", 1236, 2));
         assertThat(provider.getExclusivePorts()).containsExactlyInAnyOrder(1235, 1236, 1237);
         assertThat(provider.getSharedPorts()).isEmpty();
     }
 
     @Test
     void generatesBrokerAddresses() {
-        var provider = new PortPerBrokerClusterEndpointConfigProvider(
-                new PortPerBrokerClusterEndpointProviderConfig(parse("localhost:1235"), "localhost:$(portNumber)", 1236, 3));
+        var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
+                new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("localhost:1235"), "localhost:$(portNumber)", 1236, 3));
         assertThat(provider.getClusterBootstrapAddress()).isEqualTo(parse("localhost:1235"));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("localhost:1236"));
         assertThat(provider.getBrokerAddress(1)).isEqualTo(parse("localhost:1237"));
@@ -103,8 +103,9 @@ class PortPerBrokerClusterEndpointConfigProviderTest {
 
     @Test
     void fullyQualifiedHostNames() {
-        var provider = new PortPerBrokerClusterEndpointConfigProvider(
-                new PortPerBrokerClusterEndpointProviderConfig(parse("cluster.kafka.example.com:1235"), "broker.kafka.example.com:$(portNumber)", 1236, 1238));
+        var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
+                new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("cluster.kafka.example.com:1235"), "broker.kafka.example.com:$(portNumber)", 1236,
+                        1238));
         assertThat(provider.getClusterBootstrapAddress()).isEqualTo(parse("cluster.kafka.example.com:1235"));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("broker.kafka.example.com:1236"));
         assertThat(provider.getBrokerAddress(1)).isEqualTo(parse("broker.kafka.example.com:1237"));
@@ -112,8 +113,9 @@ class PortPerBrokerClusterEndpointConfigProviderTest {
 
     @Test
     void fullyQualifiedHostNamesWithNodeInterpolation() {
-        var provider = new PortPerBrokerClusterEndpointConfigProvider(
-                new PortPerBrokerClusterEndpointProviderConfig(parse("cluster.kafka.example.com:1235"), "broker$(nodeId).kafka.example.com:$(portNumber)", 1236, 1238));
+        var provider = new PortPerBrokerClusterNetworkAddressConfigProvider(
+                new PortPerBrokerClusterNetworkAddressConfigProviderConfig(parse("cluster.kafka.example.com:1235"), "broker$(nodeId).kafka.example.com:$(portNumber)",
+                        1236, 1238));
         assertThat(provider.getClusterBootstrapAddress()).isEqualTo(parse("cluster.kafka.example.com:1235"));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(parse("broker0.kafka.example.com:1236"));
         assertThat(provider.getBrokerAddress(1)).isEqualTo(parse("broker1.kafka.example.com:1237"));
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusterendpointprovider/SniRoutingClusterEndpointConfigProviderTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProviderTest.java
similarity index 59%
rename from kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusterendpointprovider/SniRoutingClusterEndpointConfigProviderTest.java
rename to kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProviderTest.java
index 46f320cc..92b30bc6 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusterendpointprovider/SniRoutingClusterEndpointConfigProviderTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/SniRoutingClusterNetworkAddressConfigProviderTest.java
@@ -4,25 +4,25 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious.proxy.internal.clusterendpointprovider;
+package io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.NullAndEmptySource;
 import org.junit.jupiter.params.provider.ValueSource;
 
-import io.kroxylicious.proxy.internal.clusterendpointprovider.SniRoutingClusterEndpointConfigProvider.SniRoutingClusterEndpointProviderConfig;
+import io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.SniRoutingClusterNetworkAddressConfigProvider.SniRoutingClusterNetworkAddressConfigProviderConfig;
 import io.kroxylicious.proxy.service.HostPort;
 
 import static io.kroxylicious.proxy.service.HostPort.parse;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
-class SniRoutingClusterEndpointConfigProviderTest {
+class SniRoutingClusterNetworkAddressConfigProviderTest {
 
     @Test
     void valid() {
-        new SniRoutingClusterEndpointProviderConfig(
+        new SniRoutingClusterNetworkAddressConfigProviderConfig(
                 parse("good:1235"), "broker$(nodeId)-good");
 
     }
@@ -31,18 +31,19 @@ class SniRoutingClusterEndpointConfigProviderTest {
     @ValueSource(strings = { "notgood", "toomany$(nodeId)$(nodeId)", "toomanyrecursive$(nodeId$(nodeId))", "shouty$(NODEID)badtoo" })
     @NullAndEmptySource
     void invalidBrokerPattern(String input) {
-        assertThrows(IllegalArgumentException.class, () -> new SniRoutingClusterEndpointProviderConfig(parse("good:1235"), input));
+        assertThrows(IllegalArgumentException.class, () -> new SniRoutingClusterNetworkAddressConfigProviderConfig(parse("good:1235"), input));
     }
 
     @Test
     void getBrokerAddress() {
-        var provider = new SniRoutingClusterEndpointConfigProvider(new SniRoutingClusterEndpointProviderConfig(parse("boot.kafka:1234"), "broker-$(nodeId).kafka"));
+        var provider = new SniRoutingClusterNetworkAddressConfigProvider(
+                new SniRoutingClusterNetworkAddressConfigProviderConfig(parse("boot.kafka:1234"), "broker-$(nodeId).kafka"));
         assertThat(provider.getBrokerAddress(0)).isEqualTo(HostPort.parse("broker-0.kafka:1234"));
     }
 
     @Test
     void badNodeId() {
-        assertThrows(IllegalArgumentException.class, () -> new SniRoutingClusterEndpointConfigProvider(
-                new SniRoutingClusterEndpointProviderConfig(parse("boot.kafka:1234"), "broker-$(nodeId).kafka")).getBrokerAddress(-1));
+        assertThrows(IllegalArgumentException.class, () -> new SniRoutingClusterNetworkAddressConfigProvider(
+                new SniRoutingClusterNetworkAddressConfigProviderConfig(parse("boot.kafka:1234"), "broker-$(nodeId).kafka")).getBrokerAddress(-1));
     }
 }
\ No newline at end of file
diff --git a/performance-tests/README.md b/performance-tests/README.md
index bc04f0c8..52cdbf70 100644
--- a/performance-tests/README.md
+++ b/performance-tests/README.md
@@ -14,14 +14,14 @@ mvn clean install -DskipTests
 
 ### 2. Run Benchmarks
 
-To run a benchmark like [io.kroxylicious.InvokerScalabilityBenchmark](src/main/java/io/kroxylicious/InvokerScalabilityBenchmark.java)
+To run a benchmark like [io.kroxylicious.benchmarks.InvokerScalabilityBenchmark](src/main/java/io/kroxylicious/benchmarks/InvokerScalabilityBenchmark.java)
 ```shell
-java -jar performance-tests/target/benchmarks.jar io.kroxylicious.InvokerScalabilityBenchmark
+java -jar performance-tests/target/benchmarks.jar io.kroxylicious.benchmarks.InvokerScalabilityBenchmark
 ```
 
 To run specific benchmark methods within a class you can use regex matching:
 ```shell
-java -jar performance-tests/target/benchmarks.jar io.kroxylicious.InvokerScalabilityBenchmark.methodNamePattern
+java -jar performance-tests/target/benchmarks.jar io.kroxylicious.benchmarks.InvokerScalabilityBenchmark.methodNamePattern
 ```
 
 You can control how many forks/warmup-iterations/iterations etc. are run using command line arguments, see 
diff --git a/performance-tests/src/main/java/io/kroxylicious/InvokerBenchmark.java b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerBenchmark.java
similarity index 90%
rename from performance-tests/src/main/java/io/kroxylicious/InvokerBenchmark.java
rename to performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerBenchmark.java
index 095fbb83..8ad8d3df 100644
--- a/performance-tests/src/main/java/io/kroxylicious/InvokerBenchmark.java
+++ b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerBenchmark.java
@@ -4,13 +4,17 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious;
+package io.kroxylicious.benchmarks;
 
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.Scope;
 import org.openjdk.jmh.infra.Blackhole;
 
+import io.kroxylicious.filters.EightInterfaceFilter;
+import io.kroxylicious.filters.FourInterfaceFilter;
+import io.kroxylicious.filters.OneInterfaceFilter;
+import io.kroxylicious.filters.TwoInterfaceFilter;
 import io.kroxylicious.proxy.filter.FilterInvoker;
 import io.kroxylicious.proxy.filter.FilterInvokers;
 
diff --git a/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
new file mode 100644
index 00000000..45995764
--- /dev/null
+++ b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
@@ -0,0 +1,192 @@
+/*
+ * 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.benchmarks;
+
+import java.util.Map;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.kafka.common.message.ApiVersionsRequestData;
+import org.apache.kafka.common.message.FetchRequestData;
+import org.apache.kafka.common.message.ProduceRequestData;
+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.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import io.kroxylicious.filters.FourInterfaceFilter0;
+import io.kroxylicious.filters.FourInterfaceFilter1;
+import io.kroxylicious.filters.FourInterfaceFilter2;
+import io.kroxylicious.filters.FourInterfaceFilter3;
+import io.kroxylicious.proxy.filter.ArrayFilterInvoker;
+import io.kroxylicious.proxy.filter.FilterInvoker;
+import io.kroxylicious.proxy.filter.FilterInvokers;
+import io.kroxylicious.proxy.filter.KrpcFilter;
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+import io.kroxylicious.proxy.filter.SpecificFilterInvoker;
+
+// try hard to make shouldHandleXYZ to observe different receivers concrete types, saving unrolling to bias a specific call-site to a specific concrete type
+@Fork(value = 2, jvmArgsAppend = "-XX:LoopUnrollLimit=1")
+@Warmup(iterations = 5, time = 1)
+@Measurement(iterations = 5, time = 500, timeUnit = TimeUnit.MILLISECONDS)
+public class InvokerDispatchBenchmark {
+
+    // A low and constant number of tokens allows us to balance the amount of work done in the benchmarks with still being able to observe the effects of the different dispatch mechanisms.
+    public static final int CONSUME_TOKENS = 5;
+
+    public enum Invoker {
+        array {
+            @Override
+            FilterInvoker invokerWith(KrpcFilter filter) {
+                return new ArrayFilterInvoker(filter);
+            }
+        },
+        specific {
+            @Override
+            FilterInvoker invokerWith(KrpcFilter filter) {
+                return new SpecificFilterInvoker(filter);
+            }
+        },
+        switching {
+            @Override
+            FilterInvoker invokerWith(KrpcFilter filter) {
+                return FilterInvokers.arrayInvoker(filter);
+            }
+        };
+
+        abstract FilterInvoker invokerWith(KrpcFilter filter);
+    }
+
+    @State(Scope.Benchmark)
+    public static class BenchState {
+        FilterInvoker[] invokers;
+
+        ApiKeys[] keys;
+
+        @Param({ "array", "specific", "switching" })
+        String invoker;
+
+        private RequestHeaderData requestHeaders;
+        private KrpcFilterContext filterContext;
+        private Map.Entry<ApiKeys, ApiMessage>[] apiMessages;
+
+        @SuppressWarnings("unchecked")
+        @Setup
+        public void init() {
+            Invoker invokerType = Invoker.valueOf(invoker);
+            invokers = new FilterInvoker[]{
+                    invokerType.invokerWith(new FourInterfaceFilter0()),
+                    invokerType.invokerWith(new FourInterfaceFilter1()),
+                    invokerType.invokerWith(new FourInterfaceFilter2()),
+                    invokerType.invokerWith(new FourInterfaceFilter3()),
+                    invokerType.invokerWith(new FourInterfaceFilter0()),
+                    invokerType.invokerWith(new FourInterfaceFilter1()),
+                    invokerType.invokerWith(new FourInterfaceFilter2()),
+                    invokerType.invokerWith(new FourInterfaceFilter3())
+            };
+            final Map<ApiKeys, ApiMessage> messages = Map.of(ApiKeys.PRODUCE, new ProduceRequestData(), ApiKeys.API_VERSIONS, new ApiVersionsRequestData(), ApiKeys.FETCH,
+                    new FetchRequestData());
+            apiMessages = messages.entrySet().toArray(new Map.Entry[0]); // Avoids iterator.next showing up in the benchmarks
+            requestHeaders = new RequestHeaderData();
+            filterContext = new StubFilterContext();
+            keys = messages.keySet().toArray(new ApiKeys[0]);
+        }
+    }
+
+    @Benchmark
+    public void testDispatchToShouldHandle(BenchState state, Blackhole blackhole) {
+        invokeShouldHandle(blackhole, state.invokers, state.keys);
+    }
+
+    @Benchmark
+    public void testDispatchToHandleRequest(BenchState state) {
+        invokeHandleRequest(state.invokers, state.apiMessages, state.requestHeaders, state.filterContext);
+    }
+
+    @Benchmark
+    @Threads(4)
+    public void test4ThreadsDispatchToShouldHandle(BenchState state, Blackhole blackhole) {
+        invokeShouldHandle(blackhole, state.invokers, state.keys);
+    }
+
+    @Benchmark
+    @Threads(4)
+    public void test4ThreadsDispatchToHandleRequest(BenchState state) {
+        invokeHandleRequest(state.invokers, state.apiMessages, state.requestHeaders, state.filterContext);
+    }
+
+    private static void invokeShouldHandle(Blackhole blackhole, FilterInvoker[] filters, ApiKeys[] apiKeys) {
+        for (ApiKeys apiKey : apiKeys) {
+            final short apiVersion = apiKey.latestVersion();
+            for (FilterInvoker invoker : filters) {
+                blackhole.consume(invoker.shouldHandleRequest(apiKey, apiVersion));
+                blackhole.consume(invoker.shouldHandleResponse(apiKey, apiVersion));
+            }
+        }
+    }
+
+    private static void invokeHandleRequest(FilterInvoker[] filters, Map.Entry<ApiKeys, ApiMessage>[] apiMessages, RequestHeaderData requestHeaders,
+                                            KrpcFilterContext filterContext) {
+        for (Map.Entry<ApiKeys, ApiMessage> entry : apiMessages) {
+            final ApiKeys apiKey = entry.getKey();
+            final short apiVersion = apiKey.latestVersion();
+            for (FilterInvoker invoker : filters) {
+                if (invoker.shouldHandleRequest(apiKey, apiVersion)) {
+                    invoker.onRequest(apiKey, apiVersion, requestHeaders, entry.getValue(), filterContext);
+                }
+            }
+        }
+    }
+
+    private static class StubFilterContext implements KrpcFilterContext {
+        @Override
+        public String channelDescriptor() {
+            return null;
+        }
+
+        @Override
+        public ByteBufferOutputStream createByteBufferOutputStream(int initialCapacity) {
+            return null;
+        }
+
+        @Override
+        public String sniHostname() {
+            return null;
+        }
+
+        @Override
+        public void forwardRequest(RequestHeaderData header, ApiMessage request) {
+
+        }
+
+        @Override
+        public <T extends ApiMessage> CompletionStage<T> sendRequest(short apiVersion, ApiMessage request) {
+            return null;
+        }
+
+        @Override
+        public void forwardResponse(ResponseHeaderData header, ApiMessage response) {
+
+        }
+
+        @Override
+        public void forwardResponse(ApiMessage response) {
+
+        }
+    }
+}
diff --git a/performance-tests/src/main/java/io/kroxylicious/InvokerScalabilityBenchmark.java b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerScalabilityBenchmark.java
similarity index 80%
rename from performance-tests/src/main/java/io/kroxylicious/InvokerScalabilityBenchmark.java
rename to performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerScalabilityBenchmark.java
index fec6df5a..c8564121 100644
--- a/performance-tests/src/main/java/io/kroxylicious/InvokerScalabilityBenchmark.java
+++ b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerScalabilityBenchmark.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious;
+package io.kroxylicious.benchmarks;
 
 import java.util.concurrent.TimeUnit;
 
@@ -19,9 +19,13 @@ import org.openjdk.jmh.annotations.Threads;
 import org.openjdk.jmh.annotations.Warmup;
 import org.openjdk.jmh.infra.Blackhole;
 
+import io.kroxylicious.filters.TwoInterfaceFilter0;
+import io.kroxylicious.filters.TwoInterfaceFilter1;
+import io.kroxylicious.proxy.filter.ArrayFilterInvoker;
 import io.kroxylicious.proxy.filter.FilterInvoker;
 import io.kroxylicious.proxy.filter.FilterInvokers;
 import io.kroxylicious.proxy.filter.KrpcFilter;
+import io.kroxylicious.proxy.filter.SpecificFilterInvoker;
 
 // try hard to make shouldHandleXYZ to observe different receivers concrete types, saving unrolling to bias a specific call-site to a specific concrete type
 @Fork(value = 2, jvmArgsAppend = "-XX:LoopUnrollLimit=1")
@@ -33,7 +37,7 @@ public class InvokerScalabilityBenchmark {
         array {
             @Override
             FilterInvoker invokerWith(KrpcFilter filter) {
-                return FilterInvokers.arrayInvoker(filter);
+                return new ArrayFilterInvoker(filter);
             }
         },
         specific {
@@ -41,6 +45,12 @@ public class InvokerScalabilityBenchmark {
             FilterInvoker invokerWith(KrpcFilter filter) {
                 return new SpecificFilterInvoker(filter);
             }
+        },
+        switching {
+            @Override
+            FilterInvoker invokerWith(KrpcFilter filter) {
+                return FilterInvokers.arrayInvoker(filter);
+            }
         };
 
         abstract FilterInvoker invokerWith(KrpcFilter filter);
@@ -52,7 +62,7 @@ public class InvokerScalabilityBenchmark {
 
         ApiKeys key;
 
-        @Param({ "array", "specific" })
+        @Param({ "array", "specific", "switching" })
         String invoker;
 
         @Setup
@@ -78,9 +88,10 @@ public class InvokerScalabilityBenchmark {
     }
 
     private static void invoke(Blackhole blackhole, FilterInvoker[] filters, ApiKeys key) {
+        final short apiVersion = key.latestVersion();
         for (FilterInvoker invoker : filters) {
-            blackhole.consume(invoker.shouldHandleRequest(key, key.latestVersion()));
-            blackhole.consume(invoker.shouldHandleResponse(key, key.latestVersion()));
+            blackhole.consume(invoker.shouldHandleRequest(key, apiVersion));
+            blackhole.consume(invoker.shouldHandleResponse(key, apiVersion));
         }
     }
 
diff --git a/performance-tests/src/main/java/io/kroxylicious/EightInterfaceFilter.java b/performance-tests/src/main/java/io/kroxylicious/filters/EightInterfaceFilter.java
similarity index 98%
rename from performance-tests/src/main/java/io/kroxylicious/EightInterfaceFilter.java
rename to performance-tests/src/main/java/io/kroxylicious/filters/EightInterfaceFilter.java
index dc67dc8a..5e372c9d 100644
--- a/performance-tests/src/main/java/io/kroxylicious/EightInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/EightInterfaceFilter.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious;
+package io.kroxylicious.filters;
 
 import org.apache.kafka.common.message.CreateTopicsRequestData;
 import org.apache.kafka.common.message.CreateTopicsResponseData;
diff --git a/performance-tests/src/main/java/io/kroxylicious/FourInterfaceFilter.java b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter.java
similarity index 98%
rename from performance-tests/src/main/java/io/kroxylicious/FourInterfaceFilter.java
rename to performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter.java
index 9d9db53e..c3a22e0c 100644
--- a/performance-tests/src/main/java/io/kroxylicious/FourInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious;
+package io.kroxylicious.filters;
 
 import org.apache.kafka.common.message.CreateTopicsRequestData;
 import org.apache.kafka.common.message.CreateTopicsResponseData;
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter0.java b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter0.java
new file mode 100644
index 00000000..db734d4e
--- /dev/null
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter0.java
@@ -0,0 +1,46 @@
+/*
+ * 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.filters;
+
+import org.apache.kafka.common.message.ApiVersionsRequestData;
+import org.apache.kafka.common.message.ApiVersionsResponseData;
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.ProduceResponseData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.openjdk.jmh.infra.Blackhole;
+
+import io.kroxylicious.proxy.filter.ApiVersionsRequestFilter;
+import io.kroxylicious.proxy.filter.ApiVersionsResponseFilter;
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+import io.kroxylicious.proxy.filter.ProduceRequestFilter;
+import io.kroxylicious.proxy.filter.ProduceResponseFilter;
+
+import static io.kroxylicious.benchmarks.InvokerDispatchBenchmark.CONSUME_TOKENS;
+
+public class FourInterfaceFilter0 implements ProduceResponseFilter, ProduceRequestFilter, ApiVersionsRequestFilter, ApiVersionsResponseFilter {
+
+    @Override
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+
+    @Override
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+
+    @Override
+    public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+
+    @Override
+    public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+}
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter1.java b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter1.java
new file mode 100644
index 00000000..36c690a8
--- /dev/null
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter1.java
@@ -0,0 +1,45 @@
+/*
+ * 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.filters;
+
+import org.apache.kafka.common.message.ApiVersionsRequestData;
+import org.apache.kafka.common.message.ApiVersionsResponseData;
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.ProduceResponseData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.openjdk.jmh.infra.Blackhole;
+
+import io.kroxylicious.proxy.filter.ApiVersionsRequestFilter;
+import io.kroxylicious.proxy.filter.ApiVersionsResponseFilter;
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+import io.kroxylicious.proxy.filter.ProduceRequestFilter;
+import io.kroxylicious.proxy.filter.ProduceResponseFilter;
+
+import static io.kroxylicious.benchmarks.InvokerDispatchBenchmark.CONSUME_TOKENS;
+
+public class FourInterfaceFilter1 implements ProduceResponseFilter, ProduceRequestFilter, ApiVersionsRequestFilter, ApiVersionsResponseFilter {
+    @Override
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+
+    @Override
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+
+    @Override
+    public void onApiVersionsRequest(short apiVersion, RequestHeaderData header, ApiVersionsRequestData request, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+
+    @Override
+    public void onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData response, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+}
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter2.java b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter2.java
new file mode 100644
index 00000000..eaa7240b
--- /dev/null
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter2.java
@@ -0,0 +1,46 @@
+/*
+ * 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.filters;
+
+import org.apache.kafka.common.message.FetchRequestData;
+import org.apache.kafka.common.message.FetchResponseData;
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.ProduceResponseData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.openjdk.jmh.infra.Blackhole;
+
+import io.kroxylicious.proxy.filter.FetchRequestFilter;
+import io.kroxylicious.proxy.filter.FetchResponseFilter;
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+import io.kroxylicious.proxy.filter.ProduceRequestFilter;
+import io.kroxylicious.proxy.filter.ProduceResponseFilter;
+
+import static io.kroxylicious.benchmarks.InvokerDispatchBenchmark.CONSUME_TOKENS;
+
+public class FourInterfaceFilter2 implements ProduceResponseFilter, ProduceRequestFilter, FetchRequestFilter, FetchResponseFilter {
+
+    @Override
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+
+    @Override
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+
+    @Override
+    public void onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+
+    @Override
+    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+}
diff --git a/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter3.java b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter3.java
new file mode 100644
index 00000000..90c65d97
--- /dev/null
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/FourInterfaceFilter3.java
@@ -0,0 +1,46 @@
+/*
+ * 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.filters;
+
+import org.apache.kafka.common.message.FetchRequestData;
+import org.apache.kafka.common.message.FetchResponseData;
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.ProduceResponseData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.openjdk.jmh.infra.Blackhole;
+
+import io.kroxylicious.proxy.filter.FetchRequestFilter;
+import io.kroxylicious.proxy.filter.FetchResponseFilter;
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+import io.kroxylicious.proxy.filter.ProduceRequestFilter;
+import io.kroxylicious.proxy.filter.ProduceResponseFilter;
+
+import static io.kroxylicious.benchmarks.InvokerDispatchBenchmark.CONSUME_TOKENS;
+
+public class FourInterfaceFilter3 implements ProduceResponseFilter, ProduceRequestFilter, FetchRequestFilter, FetchResponseFilter {
+
+    @Override
+    public void onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+
+    @Override
+    public void onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+
+    @Override
+    public void onFetchRequest(short apiVersion, RequestHeaderData header, FetchRequestData request, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+
+    @Override
+    public void onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, KrpcFilterContext context) {
+        Blackhole.consumeCPU(CONSUME_TOKENS);
+    }
+}
diff --git a/performance-tests/src/main/java/io/kroxylicious/OneInterfaceFilter.java b/performance-tests/src/main/java/io/kroxylicious/filters/OneInterfaceFilter.java
similarity index 94%
rename from performance-tests/src/main/java/io/kroxylicious/OneInterfaceFilter.java
rename to performance-tests/src/main/java/io/kroxylicious/filters/OneInterfaceFilter.java
index 1ed6dfbb..bacb5e2f 100644
--- a/performance-tests/src/main/java/io/kroxylicious/OneInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/OneInterfaceFilter.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious;
+package io.kroxylicious.filters;
 
 import org.apache.kafka.common.message.ProduceResponseData;
 import org.apache.kafka.common.message.ResponseHeaderData;
diff --git a/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter.java b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter.java
similarity index 96%
rename from performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter.java
rename to performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter.java
index 326404fd..e1c3459d 100644
--- a/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious;
+package io.kroxylicious.filters;
 
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceResponseData;
diff --git a/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter0.java b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter0.java
similarity index 96%
rename from performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter0.java
rename to performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter0.java
index 9b9d2059..9c249754 100644
--- a/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter0.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter0.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious;
+package io.kroxylicious.filters;
 
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceResponseData;
diff --git a/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter1.java b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter1.java
similarity index 96%
rename from performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter1.java
rename to performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter1.java
index 6cb67618..a58cd4b7 100644
--- a/performance-tests/src/main/java/io/kroxylicious/TwoInterfaceFilter1.java
+++ b/performance-tests/src/main/java/io/kroxylicious/filters/TwoInterfaceFilter1.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious;
+package io.kroxylicious.filters;
 
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceResponseData;
diff --git a/performance-tests/src/main/java/io/kroxylicious/proxy/filter/ArrayFilterInvoker.java b/performance-tests/src/main/java/io/kroxylicious/proxy/filter/ArrayFilterInvoker.java
new file mode 100644
index 00000000..f7275cca
--- /dev/null
+++ b/performance-tests/src/main/java/io/kroxylicious/proxy/filter/ArrayFilterInvoker.java
@@ -0,0 +1,553 @@
+/*
+ * 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.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.OptionalInt;
+
+import org.apache.kafka.common.message.ApiMessageType;
+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;
+
+/**
+ * Invoker for KrpcFilters that implement any number of Specific Message interfaces (for
+ * example {@link AlterConfigsResponseFilter}.
+ */
+public class ArrayFilterInvoker implements FilterInvoker {
+
+    private static final FilterInvoker[] HANDLE_NOTHING = createHandleNothing();
+
+    private final FilterInvoker[] requestInvokers;
+    private final FilterInvoker[] responseInvokers;
+
+    public ArrayFilterInvoker(KrpcFilter filter) {
+        Map<Integer, FilterInvoker> requestInvokers = new HashMap<>();
+        Map<Integer, FilterInvoker> responseInvokers = new HashMap<>();
+        if (filter instanceof AddOffsetsToTxnRequestFilter) {
+            requestInvokers.put(25, new AddOffsetsToTxnRequestFilterInvoker((AddOffsetsToTxnRequestFilter) filter));
+        }
+        if (filter instanceof AddOffsetsToTxnResponseFilter) {
+            responseInvokers.put(25, new AddOffsetsToTxnResponseFilterInvoker((AddOffsetsToTxnResponseFilter) filter));
+        }
+        if (filter instanceof AddPartitionsToTxnRequestFilter) {
+            requestInvokers.put(24, new AddPartitionsToTxnRequestFilterInvoker((AddPartitionsToTxnRequestFilter) filter));
+        }
+        if (filter instanceof AddPartitionsToTxnResponseFilter) {
+            responseInvokers.put(24, new AddPartitionsToTxnResponseFilterInvoker((AddPartitionsToTxnResponseFilter) filter));
+        }
+        if (filter instanceof AllocateProducerIdsRequestFilter) {
+            requestInvokers.put(67, new AllocateProducerIdsRequestFilterInvoker((AllocateProducerIdsRequestFilter) filter));
+        }
+        if (filter instanceof AllocateProducerIdsResponseFilter) {
+            responseInvokers.put(67, new AllocateProducerIdsResponseFilterInvoker((AllocateProducerIdsResponseFilter) filter));
+        }
+        if (filter instanceof AlterClientQuotasRequestFilter) {
+            requestInvokers.put(49, new AlterClientQuotasRequestFilterInvoker((AlterClientQuotasRequestFilter) filter));
+        }
+        if (filter instanceof AlterClientQuotasResponseFilter) {
+            responseInvokers.put(49, new AlterClientQuotasResponseFilterInvoker((AlterClientQuotasResponseFilter) filter));
+        }
+        if (filter instanceof AlterConfigsRequestFilter) {
+            requestInvokers.put(33, new AlterConfigsRequestFilterInvoker((AlterConfigsRequestFilter) filter));
+        }
+        if (filter instanceof AlterConfigsResponseFilter) {
+            responseInvokers.put(33, new AlterConfigsResponseFilterInvoker((AlterConfigsResponseFilter) filter));
+        }
+        if (filter instanceof AlterPartitionReassignmentsRequestFilter) {
+            requestInvokers.put(45, new AlterPartitionReassignmentsRequestFilterInvoker((AlterPartitionReassignmentsRequestFilter) filter));
+        }
+        if (filter instanceof AlterPartitionReassignmentsResponseFilter) {
+            responseInvokers.put(45, new AlterPartitionReassignmentsResponseFilterInvoker((AlterPartitionReassignmentsResponseFilter) filter));
+        }
+        if (filter instanceof AlterPartitionRequestFilter) {
+            requestInvokers.put(56, new AlterPartitionRequestFilterInvoker((AlterPartitionRequestFilter) filter));
+        }
+        if (filter instanceof AlterPartitionResponseFilter) {
+            responseInvokers.put(56, new AlterPartitionResponseFilterInvoker((AlterPartitionResponseFilter) filter));
+        }
+        if (filter instanceof AlterReplicaLogDirsRequestFilter) {
+            requestInvokers.put(34, new AlterReplicaLogDirsRequestFilterInvoker((AlterReplicaLogDirsRequestFilter) filter));
+        }
+        if (filter instanceof AlterReplicaLogDirsResponseFilter) {
+            responseInvokers.put(34, new AlterReplicaLogDirsResponseFilterInvoker((AlterReplicaLogDirsResponseFilter) filter));
+        }
+        if (filter instanceof AlterUserScramCredentialsRequestFilter) {
+            requestInvokers.put(51, new AlterUserScramCredentialsRequestFilterInvoker((AlterUserScramCredentialsRequestFilter) filter));
+        }
+        if (filter instanceof AlterUserScramCredentialsResponseFilter) {
+            responseInvokers.put(51, new AlterUserScramCredentialsResponseFilterInvoker((AlterUserScramCredentialsResponseFilter) filter));
+        }
+        if (filter instanceof ApiVersionsRequestFilter) {
+            requestInvokers.put(18, new ApiVersionsRequestFilterInvoker((ApiVersionsRequestFilter) filter));
+        }
+        if (filter instanceof ApiVersionsResponseFilter) {
+            responseInvokers.put(18, new ApiVersionsResponseFilterInvoker((ApiVersionsResponseFilter) filter));
+        }
+        if (filter instanceof BeginQuorumEpochRequestFilter) {
+            requestInvokers.put(53, new BeginQuorumEpochRequestFilterInvoker((BeginQuorumEpochRequestFilter) filter));
+        }
+        if (filter instanceof BeginQuorumEpochResponseFilter) {
+            responseInvokers.put(53, new BeginQuorumEpochResponseFilterInvoker((BeginQuorumEpochResponseFilter) filter));
+        }
+        if (filter instanceof BrokerHeartbeatRequestFilter) {
+            requestInvokers.put(63, new BrokerHeartbeatRequestFilterInvoker((BrokerHeartbeatRequestFilter) filter));
+        }
+        if (filter instanceof BrokerHeartbeatResponseFilter) {
+            responseInvokers.put(63, new BrokerHeartbeatResponseFilterInvoker((BrokerHeartbeatResponseFilter) filter));
+        }
+        if (filter instanceof BrokerRegistrationRequestFilter) {
+            requestInvokers.put(62, new BrokerRegistrationRequestFilterInvoker((BrokerRegistrationRequestFilter) filter));
+        }
+        if (filter instanceof BrokerRegistrationResponseFilter) {
+            responseInvokers.put(62, new BrokerRegistrationResponseFilterInvoker((BrokerRegistrationResponseFilter) filter));
+        }
+        if (filter instanceof ControlledShutdownRequestFilter) {
+            requestInvokers.put(7, new ControlledShutdownRequestFilterInvoker((ControlledShutdownRequestFilter) filter));
+        }
+        if (filter instanceof ControlledShutdownResponseFilter) {
+            responseInvokers.put(7, new ControlledShutdownResponseFilterInvoker((ControlledShutdownResponseFilter) filter));
+        }
+        if (filter instanceof CreateAclsRequestFilter) {
+            requestInvokers.put(30, new CreateAclsRequestFilterInvoker((CreateAclsRequestFilter) filter));
+        }
+        if (filter instanceof CreateAclsResponseFilter) {
+            responseInvokers.put(30, new CreateAclsResponseFilterInvoker((CreateAclsResponseFilter) filter));
+        }
+        if (filter instanceof CreateDelegationTokenRequestFilter) {
+            requestInvokers.put(38, new CreateDelegationTokenRequestFilterInvoker((CreateDelegationTokenRequestFilter) filter));
+        }
+        if (filter instanceof CreateDelegationTokenResponseFilter) {
+            responseInvokers.put(38, new CreateDelegationTokenResponseFilterInvoker((CreateDelegationTokenResponseFilter) filter));
+        }
+        if (filter instanceof CreatePartitionsRequestFilter) {
+            requestInvokers.put(37, new CreatePartitionsRequestFilterInvoker((CreatePartitionsRequestFilter) filter));
+        }
+        if (filter instanceof CreatePartitionsResponseFilter) {
+            responseInvokers.put(37, new CreatePartitionsResponseFilterInvoker((CreatePartitionsResponseFilter) filter));
+        }
+        if (filter instanceof CreateTopicsRequestFilter) {
+            requestInvokers.put(19, new CreateTopicsRequestFilterInvoker((CreateTopicsRequestFilter) filter));
+        }
+        if (filter instanceof CreateTopicsResponseFilter) {
+            responseInvokers.put(19, new CreateTopicsResponseFilterInvoker((CreateTopicsResponseFilter) filter));
+        }
+        if (filter instanceof DeleteAclsRequestFilter) {
+            requestInvokers.put(31, new DeleteAclsRequestFilterInvoker((DeleteAclsRequestFilter) filter));
+        }
+        if (filter instanceof DeleteAclsResponseFilter) {
+            responseInvokers.put(31, new DeleteAclsResponseFilterInvoker((DeleteAclsResponseFilter) filter));
+        }
+        if (filter instanceof DeleteGroupsRequestFilter) {
+            requestInvokers.put(42, new DeleteGroupsRequestFilterInvoker((DeleteGroupsRequestFilter) filter));
+        }
+        if (filter instanceof DeleteGroupsResponseFilter) {
+            responseInvokers.put(42, new DeleteGroupsResponseFilterInvoker((DeleteGroupsResponseFilter) filter));
+        }
+        if (filter instanceof DeleteRecordsRequestFilter) {
+            requestInvokers.put(21, new DeleteRecordsRequestFilterInvoker((DeleteRecordsRequestFilter) filter));
+        }
+        if (filter instanceof DeleteRecordsResponseFilter) {
+            responseInvokers.put(21, new DeleteRecordsResponseFilterInvoker((DeleteRecordsResponseFilter) filter));
+        }
+        if (filter instanceof DeleteTopicsRequestFilter) {
+            requestInvokers.put(20, new DeleteTopicsRequestFilterInvoker((DeleteTopicsRequestFilter) filter));
+        }
+        if (filter instanceof DeleteTopicsResponseFilter) {
+            responseInvokers.put(20, new DeleteTopicsResponseFilterInvoker((DeleteTopicsResponseFilter) filter));
+        }
+        if (filter instanceof DescribeAclsRequestFilter) {
+            requestInvokers.put(29, new DescribeAclsRequestFilterInvoker((DescribeAclsRequestFilter) filter));
+        }
+        if (filter instanceof DescribeAclsResponseFilter) {
+            responseInvokers.put(29, new DescribeAclsResponseFilterInvoker((DescribeAclsResponseFilter) filter));
+        }
+        if (filter instanceof DescribeClientQuotasRequestFilter) {
+            requestInvokers.put(48, new DescribeClientQuotasRequestFilterInvoker((DescribeClientQuotasRequestFilter) filter));
+        }
+        if (filter instanceof DescribeClientQuotasResponseFilter) {
+            responseInvokers.put(48, new DescribeClientQuotasResponseFilterInvoker((DescribeClientQuotasResponseFilter) filter));
+        }
+        if (filter instanceof DescribeClusterRequestFilter) {
+            requestInvokers.put(60, new DescribeClusterRequestFilterInvoker((DescribeClusterRequestFilter) filter));
+        }
+        if (filter instanceof DescribeClusterResponseFilter) {
+            responseInvokers.put(60, new DescribeClusterResponseFilterInvoker((DescribeClusterResponseFilter) filter));
+        }
+        if (filter instanceof DescribeConfigsRequestFilter) {
+            requestInvokers.put(32, new DescribeConfigsRequestFilterInvoker((DescribeConfigsRequestFilter) filter));
+        }
+        if (filter instanceof DescribeConfigsResponseFilter) {
+            responseInvokers.put(32, new DescribeConfigsResponseFilterInvoker((DescribeConfigsResponseFilter) filter));
+        }
+        if (filter instanceof DescribeDelegationTokenRequestFilter) {
+            requestInvokers.put(41, new DescribeDelegationTokenRequestFilterInvoker((DescribeDelegationTokenRequestFilter) filter));
+        }
+        if (filter instanceof DescribeDelegationTokenResponseFilter) {
+            responseInvokers.put(41, new DescribeDelegationTokenResponseFilterInvoker((DescribeDelegationTokenResponseFilter) filter));
+        }
+        if (filter instanceof DescribeGroupsRequestFilter) {
+            requestInvokers.put(15, new DescribeGroupsRequestFilterInvoker((DescribeGroupsRequestFilter) filter));
+        }
+        if (filter instanceof DescribeGroupsResponseFilter) {
+            responseInvokers.put(15, new DescribeGroupsResponseFilterInvoker((DescribeGroupsResponseFilter) filter));
+        }
+        if (filter instanceof DescribeLogDirsRequestFilter) {
+            requestInvokers.put(35, new DescribeLogDirsRequestFilterInvoker((DescribeLogDirsRequestFilter) filter));
+        }
+        if (filter instanceof DescribeLogDirsResponseFilter) {
+            responseInvokers.put(35, new DescribeLogDirsResponseFilterInvoker((DescribeLogDirsResponseFilter) filter));
+        }
+        if (filter instanceof DescribeProducersRequestFilter) {
+            requestInvokers.put(61, new DescribeProducersRequestFilterInvoker((DescribeProducersRequestFilter) filter));
+        }
+        if (filter instanceof DescribeProducersResponseFilter) {
+            responseInvokers.put(61, new DescribeProducersResponseFilterInvoker((DescribeProducersResponseFilter) filter));
+        }
+        if (filter instanceof DescribeQuorumRequestFilter) {
+            requestInvokers.put(55, new DescribeQuorumRequestFilterInvoker((DescribeQuorumRequestFilter) filter));
+        }
+        if (filter instanceof DescribeQuorumResponseFilter) {
+            responseInvokers.put(55, new DescribeQuorumResponseFilterInvoker((DescribeQuorumResponseFilter) filter));
+        }
+        if (filter instanceof DescribeTransactionsRequestFilter) {
+            requestInvokers.put(65, new DescribeTransactionsRequestFilterInvoker((DescribeTransactionsRequestFilter) filter));
+        }
+        if (filter instanceof DescribeTransactionsResponseFilter) {
+            responseInvokers.put(65, new DescribeTransactionsResponseFilterInvoker((DescribeTransactionsResponseFilter) filter));
+        }
+        if (filter instanceof DescribeUserScramCredentialsRequestFilter) {
+            requestInvokers.put(50, new DescribeUserScramCredentialsRequestFilterInvoker((DescribeUserScramCredentialsRequestFilter) filter));
+        }
+        if (filter instanceof DescribeUserScramCredentialsResponseFilter) {
+            responseInvokers.put(50, new DescribeUserScramCredentialsResponseFilterInvoker((DescribeUserScramCredentialsResponseFilter) filter));
+        }
+        if (filter instanceof ElectLeadersRequestFilter) {
+            requestInvokers.put(43, new ElectLeadersRequestFilterInvoker((ElectLeadersRequestFilter) filter));
+        }
+        if (filter instanceof ElectLeadersResponseFilter) {
+            responseInvokers.put(43, new ElectLeadersResponseFilterInvoker((ElectLeadersResponseFilter) filter));
+        }
+        if (filter instanceof EndQuorumEpochRequestFilter) {
+            requestInvokers.put(54, new EndQuorumEpochRequestFilterInvoker((EndQuorumEpochRequestFilter) filter));
+        }
+        if (filter instanceof EndQuorumEpochResponseFilter) {
+            responseInvokers.put(54, new EndQuorumEpochResponseFilterInvoker((EndQuorumEpochResponseFilter) filter));
+        }
+        if (filter instanceof EndTxnRequestFilter) {
+            requestInvokers.put(26, new EndTxnRequestFilterInvoker((EndTxnRequestFilter) filter));
+        }
+        if (filter instanceof EndTxnResponseFilter) {
+            responseInvokers.put(26, new EndTxnResponseFilterInvoker((EndTxnResponseFilter) filter));
+        }
+        if (filter instanceof EnvelopeRequestFilter) {
+            requestInvokers.put(58, new EnvelopeRequestFilterInvoker((EnvelopeRequestFilter) filter));
+        }
+        if (filter instanceof EnvelopeResponseFilter) {
+            responseInvokers.put(58, new EnvelopeResponseFilterInvoker((EnvelopeResponseFilter) filter));
+        }
+        if (filter instanceof ExpireDelegationTokenRequestFilter) {
+            requestInvokers.put(40, new ExpireDelegationTokenRequestFilterInvoker((ExpireDelegationTokenRequestFilter) filter));
+        }
+        if (filter instanceof ExpireDelegationTokenResponseFilter) {
+            responseInvokers.put(40, new ExpireDelegationTokenResponseFilterInvoker((ExpireDelegationTokenResponseFilter) filter));
+        }
+        if (filter instanceof FetchRequestFilter) {
+            requestInvokers.put(1, new FetchRequestFilterInvoker((FetchRequestFilter) filter));
+        }
+        if (filter instanceof FetchResponseFilter) {
+            responseInvokers.put(1, new FetchResponseFilterInvoker((FetchResponseFilter) filter));
+        }
+        if (filter instanceof FetchSnapshotRequestFilter) {
+            requestInvokers.put(59, new FetchSnapshotRequestFilterInvoker((FetchSnapshotRequestFilter) filter));
+        }
+        if (filter instanceof FetchSnapshotResponseFilter) {
+            responseInvokers.put(59, new FetchSnapshotResponseFilterInvoker((FetchSnapshotResponseFilter) filter));
+        }
+        if (filter instanceof FindCoordinatorRequestFilter) {
+            requestInvokers.put(10, new FindCoordinatorRequestFilterInvoker((FindCoordinatorRequestFilter) filter));
+        }
+        if (filter instanceof FindCoordinatorResponseFilter) {
+            responseInvokers.put(10, new FindCoordinatorResponseFilterInvoker((FindCoordinatorResponseFilter) filter));
+        }
+        if (filter instanceof HeartbeatRequestFilter) {
+            requestInvokers.put(12, new HeartbeatRequestFilterInvoker((HeartbeatRequestFilter) filter));
+        }
+        if (filter instanceof HeartbeatResponseFilter) {
+            responseInvokers.put(12, new HeartbeatResponseFilterInvoker((HeartbeatResponseFilter) filter));
+        }
+        if (filter instanceof IncrementalAlterConfigsRequestFilter) {
+            requestInvokers.put(44, new IncrementalAlterConfigsRequestFilterInvoker((IncrementalAlterConfigsRequestFilter) filter));
+        }
+        if (filter instanceof IncrementalAlterConfigsResponseFilter) {
+            responseInvokers.put(44, new IncrementalAlterConfigsResponseFilterInvoker((IncrementalAlterConfigsResponseFilter) filter));
+        }
+        if (filter instanceof InitProducerIdRequestFilter) {
+            requestInvokers.put(22, new InitProducerIdRequestFilterInvoker((InitProducerIdRequestFilter) filter));
+        }
+        if (filter instanceof InitProducerIdResponseFilter) {
+            responseInvokers.put(22, new InitProducerIdResponseFilterInvoker((InitProducerIdResponseFilter) filter));
+        }
+        if (filter instanceof JoinGroupRequestFilter) {
+            requestInvokers.put(11, new JoinGroupRequestFilterInvoker((JoinGroupRequestFilter) filter));
+        }
+        if (filter instanceof JoinGroupResponseFilter) {
+            responseInvokers.put(11, new JoinGroupResponseFilterInvoker((JoinGroupResponseFilter) filter));
+        }
+        if (filter instanceof LeaderAndIsrRequestFilter) {
+            requestInvokers.put(4, new LeaderAndIsrRequestFilterInvoker((LeaderAndIsrRequestFilter) filter));
+        }
+        if (filter instanceof LeaderAndIsrResponseFilter) {
+            responseInvokers.put(4, new LeaderAndIsrResponseFilterInvoker((LeaderAndIsrResponseFilter) filter));
+        }
+        if (filter instanceof LeaveGroupRequestFilter) {
+            requestInvokers.put(13, new LeaveGroupRequestFilterInvoker((LeaveGroupRequestFilter) filter));
+        }
+        if (filter instanceof LeaveGroupResponseFilter) {
+            responseInvokers.put(13, new LeaveGroupResponseFilterInvoker((LeaveGroupResponseFilter) filter));
+        }
+        if (filter instanceof ListGroupsRequestFilter) {
+            requestInvokers.put(16, new ListGroupsRequestFilterInvoker((ListGroupsRequestFilter) filter));
+        }
+        if (filter instanceof ListGroupsResponseFilter) {
+            responseInvokers.put(16, new ListGroupsResponseFilterInvoker((ListGroupsResponseFilter) filter));
+        }
+        if (filter instanceof ListOffsetsRequestFilter) {
+            requestInvokers.put(2, new ListOffsetsRequestFilterInvoker((ListOffsetsRequestFilter) filter));
+        }
+        if (filter instanceof ListOffsetsResponseFilter) {
+            responseInvokers.put(2, new ListOffsetsResponseFilterInvoker((ListOffsetsResponseFilter) filter));
+        }
+        if (filter instanceof ListPartitionReassignmentsRequestFilter) {
+            requestInvokers.put(46, new ListPartitionReassignmentsRequestFilterInvoker((ListPartitionReassignmentsRequestFilter) filter));
+        }
+        if (filter instanceof ListPartitionReassignmentsResponseFilter) {
+            responseInvokers.put(46, new ListPartitionReassignmentsResponseFilterInvoker((ListPartitionReassignmentsResponseFilter) filter));
+        }
+        if (filter instanceof ListTransactionsRequestFilter) {
+            requestInvokers.put(66, new ListTransactionsRequestFilterInvoker((ListTransactionsRequestFilter) filter));
+        }
+        if (filter instanceof ListTransactionsResponseFilter) {
+            responseInvokers.put(66, new ListTransactionsResponseFilterInvoker((ListTransactionsResponseFilter) filter));
+        }
+        if (filter instanceof MetadataRequestFilter) {
+            requestInvokers.put(3, new MetadataRequestFilterInvoker((MetadataRequestFilter) filter));
+        }
+        if (filter instanceof MetadataResponseFilter) {
+            responseInvokers.put(3, new MetadataResponseFilterInvoker((MetadataResponseFilter) filter));
+        }
+        if (filter instanceof OffsetCommitRequestFilter) {
+            requestInvokers.put(8, new OffsetCommitRequestFilterInvoker((OffsetCommitRequestFilter) filter));
+        }
+        if (filter instanceof OffsetCommitResponseFilter) {
+            responseInvokers.put(8, new OffsetCommitResponseFilterInvoker((OffsetCommitResponseFilter) filter));
+        }
+        if (filter instanceof OffsetDeleteRequestFilter) {
+            requestInvokers.put(47, new OffsetDeleteRequestFilterInvoker((OffsetDeleteRequestFilter) filter));
+        }
+        if (filter instanceof OffsetDeleteResponseFilter) {
+            responseInvokers.put(47, new OffsetDeleteResponseFilterInvoker((OffsetDeleteResponseFilter) filter));
+        }
+        if (filter instanceof OffsetFetchRequestFilter) {
+            requestInvokers.put(9, new OffsetFetchRequestFilterInvoker((OffsetFetchRequestFilter) filter));
+        }
+        if (filter instanceof OffsetFetchResponseFilter) {
+            responseInvokers.put(9, new OffsetFetchResponseFilterInvoker((OffsetFetchResponseFilter) filter));
+        }
+        if (filter instanceof OffsetForLeaderEpochRequestFilter) {
+            requestInvokers.put(23, new OffsetForLeaderEpochRequestFilterInvoker((OffsetForLeaderEpochRequestFilter) filter));
+        }
+        if (filter instanceof OffsetForLeaderEpochResponseFilter) {
+            responseInvokers.put(23, new OffsetForLeaderEpochResponseFilterInvoker((OffsetForLeaderEpochResponseFilter) filter));
+        }
+        if (filter instanceof ProduceRequestFilter) {
+            requestInvokers.put(0, new ProduceRequestFilterInvoker((ProduceRequestFilter) filter));
+        }
+        if (filter instanceof ProduceResponseFilter) {
+            responseInvokers.put(0, new ProduceResponseFilterInvoker((ProduceResponseFilter) filter));
+        }
+        if (filter instanceof RenewDelegationTokenRequestFilter) {
+            requestInvokers.put(39, new RenewDelegationTokenRequestFilterInvoker((RenewDelegationTokenRequestFilter) filter));
+        }
+        if (filter instanceof RenewDelegationTokenResponseFilter) {
+            responseInvokers.put(39, new RenewDelegationTokenResponseFilterInvoker((RenewDelegationTokenResponseFilter) filter));
+        }
+        if (filter instanceof SaslAuthenticateRequestFilter) {
+            requestInvokers.put(36, new SaslAuthenticateRequestFilterInvoker((SaslAuthenticateRequestFilter) filter));
+        }
+        if (filter instanceof SaslAuthenticateResponseFilter) {
+            responseInvokers.put(36, new SaslAuthenticateResponseFilterInvoker((SaslAuthenticateResponseFilter) filter));
+        }
+        if (filter instanceof SaslHandshakeRequestFilter) {
+            requestInvokers.put(17, new SaslHandshakeRequestFilterInvoker((SaslHandshakeRequestFilter) filter));
+        }
+        if (filter instanceof SaslHandshakeResponseFilter) {
+            responseInvokers.put(17, new SaslHandshakeResponseFilterInvoker((SaslHandshakeResponseFilter) filter));
+        }
+        if (filter instanceof StopReplicaRequestFilter) {
+            requestInvokers.put(5, new StopReplicaRequestFilterInvoker((StopReplicaRequestFilter) filter));
+        }
+        if (filter instanceof StopReplicaResponseFilter) {
+            responseInvokers.put(5, new StopReplicaResponseFilterInvoker((StopReplicaResponseFilter) filter));
+        }
+        if (filter instanceof SyncGroupRequestFilter) {
+            requestInvokers.put(14, new SyncGroupRequestFilterInvoker((SyncGroupRequestFilter) filter));
+        }
+        if (filter instanceof SyncGroupResponseFilter) {
+            responseInvokers.put(14, new SyncGroupResponseFilterInvoker((SyncGroupResponseFilter) filter));
+        }
+        if (filter instanceof TxnOffsetCommitRequestFilter) {
+            requestInvokers.put(28, new TxnOffsetCommitRequestFilterInvoker((TxnOffsetCommitRequestFilter) filter));
+        }
+        if (filter instanceof TxnOffsetCommitResponseFilter) {
+            responseInvokers.put(28, new TxnOffsetCommitResponseFilterInvoker((TxnOffsetCommitResponseFilter) filter));
+        }
+        if (filter instanceof UnregisterBrokerRequestFilter) {
+            requestInvokers.put(64, new UnregisterBrokerRequestFilterInvoker((UnregisterBrokerRequestFilter) filter));
+        }
+        if (filter instanceof UnregisterBrokerResponseFilter) {
+            responseInvokers.put(64, new UnregisterBrokerResponseFilterInvoker((UnregisterBrokerResponseFilter) filter));
+        }
+        if (filter instanceof UpdateFeaturesRequestFilter) {
+            requestInvokers.put(57, new UpdateFeaturesRequestFilterInvoker((UpdateFeaturesRequestFilter) filter));
+        }
+        if (filter instanceof UpdateFeaturesResponseFilter) {
+            responseInvokers.put(57, new UpdateFeaturesResponseFilterInvoker((UpdateFeaturesResponseFilter) filter));
+        }
+        if (filter instanceof UpdateMetadataRequestFilter) {
+            requestInvokers.put(6, new UpdateMetadataRequestFilterInvoker((UpdateMetadataRequestFilter) filter));
+        }
+        if (filter instanceof UpdateMetadataResponseFilter) {
+            responseInvokers.put(6, new UpdateMetadataResponseFilterInvoker((UpdateMetadataResponseFilter) filter));
+        }
+        if (filter instanceof VoteRequestFilter) {
+            requestInvokers.put(52, new VoteRequestFilterInvoker((VoteRequestFilter) filter));
+        }
+        if (filter instanceof VoteResponseFilter) {
+            responseInvokers.put(52, new VoteResponseFilterInvoker((VoteResponseFilter) filter));
+        }
+        if (filter instanceof WriteTxnMarkersRequestFilter) {
+            requestInvokers.put(27, new WriteTxnMarkersRequestFilterInvoker((WriteTxnMarkersRequestFilter) filter));
+        }
+        if (filter instanceof WriteTxnMarkersResponseFilter) {
+            responseInvokers.put(27, new WriteTxnMarkersResponseFilterInvoker((WriteTxnMarkersResponseFilter) filter));
+        }
+        this.requestInvokers = createFrom(requestInvokers);
+        this.responseInvokers = createFrom(responseInvokers);
+    }
+
+    /**
+     * Apply the filter to the given {@code header} and {@code body} using the given {@code filterContext}.
+     * @param apiKey The request api key.
+     * @param apiVersion The request api version.
+     * @param header The request header.
+     * @param body The request body.
+     * @param filterContext The filter context.
+     */
+    @Override
+    public void onRequest(ApiKeys apiKey,
+                          short apiVersion,
+                          RequestHeaderData header,
+                          ApiMessage body,
+                          KrpcFilterContext filterContext) {
+        FilterInvoker invoker = requestInvokers[apiKey.id];
+        invoker.onRequest(apiKey, apiVersion, header, body, filterContext);
+    }
+
+    /**
+     * Apply the filter to the given {@code header} and {@code body} using the given {@code filterContext}.
+     * @param apiKey The request api key.
+     * @param apiVersion The api version.
+     * @param header The request header.
+     * @param body The request body.
+     * @param filterContext The filter context.
+     */
+    @Override
+    public void onResponse(ApiKeys apiKey,
+                           short apiVersion,
+                           ResponseHeaderData header,
+                           ApiMessage body,
+                           KrpcFilterContext filterContext) {
+        FilterInvoker invoker = responseInvokers[apiKey.id];
+        invoker.onResponse(apiKey, apiVersion, header, body, filterContext);
+    }
+
+    /**
+     * <p>Determines whether a request with the given {@code apiKey} and {@code apiVersion} should be deserialized.
+     * Note that it is not guaranteed that this method will be called once per request,
+     * or that two consecutive calls refer to the same request.
+     * That is, the sequences of invocations like the following are allowed:</p>
+     * <ol>
+     *     <li>{@code shouldHandleRequest} on request A</li>
+     *     <li>{@code shouldHandleRequest} on request B</li>
+     *     <li>{@code shouldHandleRequest} on request A</li>
+     *     <li>{@code onRequest} on request A</li>
+     *     <li>{@code onRequest} on request B</li>
+     * </ol>
+     * @param apiKey The API key
+     * @param apiVersion The API version
+     * @return true if request should be deserialized
+     */
+    @Override
+    public boolean shouldHandleRequest(ApiKeys apiKey, short apiVersion) {
+        FilterInvoker invoker = requestInvokers[apiKey.id];
+        return invoker.shouldHandleRequest(apiKey, apiVersion);
+    }
+
+    /**
+     * <p>Determines whether a response with the given {@code apiKey} and {@code apiVersion} should be deserialized.
+     * Note that it is not guaranteed that this method will be called once per response,
+     * or that two consecutive calls refer to the same response.
+     * That is, the sequences of invocations like the following are allowed:</p>
+     * <ol>
+     *     <li>{@code shouldHandleResponse} on response A</li>
+     *     <li>{@code shouldHandleResponse} on response B</li>
+     *     <li>{@code shouldHandleResponse} on response A</li>
+     *     <li>{@code apply} on response A</li>
+     *     <li>{@code apply} on response B</li>
+     * </ol>
+     * @param apiKey The API key
+     * @param apiVersion The API version
+     * @return true if response should be deserialized
+     */
+    @Override
+    public boolean shouldHandleResponse(ApiKeys apiKey, short apiVersion) {
+        FilterInvoker invoker = responseInvokers[apiKey.id];
+        return invoker.shouldHandleResponse(apiKey, apiVersion);
+    }
+
+    private static FilterInvoker[] createHandleNothing() {
+        FilterInvoker[] filterInvokers = emptyInvokerArraySizedForMessageTypes();
+        Arrays.stream(ApiMessageType.values()).mapToInt(ApiMessageType::apiKey).forEach(value -> {
+            filterInvokers[value] = FilterInvokers.handleNothingInvoker();
+        });
+        return filterInvokers;
+    }
+
+    private static FilterInvoker[] createFrom(Map<Integer, FilterInvoker> filterInvokersByApiMessageId) {
+        if (filterInvokersByApiMessageId.isEmpty()) {
+            return HANDLE_NOTHING;
+        }
+        FilterInvoker[] filterInvokers = emptyInvokerArraySizedForMessageTypes();
+        Arrays.stream(ApiMessageType.values()).mapToInt(ApiMessageType::apiKey).forEach(value -> {
+            filterInvokers[value] = filterInvokersByApiMessageId.getOrDefault(value, FilterInvokers.handleNothingInvoker());
+        });
+        return filterInvokers;
+    }
+
+    private static FilterInvoker[] emptyInvokerArraySizedForMessageTypes() {
+        OptionalInt maybeMaxId = Arrays.stream(ApiMessageType.values()).mapToInt(ApiMessageType::apiKey).max();
+        if (maybeMaxId.isEmpty()) {
+            throw new IllegalStateException("no maximum id found");
+        }
+        int arraySize = maybeMaxId.getAsInt() + 1;
+        return new FilterInvoker[arraySize];
+    }
+
+}
diff --git a/performance-tests/src/main/java/io/kroxylicious/SpcificFilterInvoker.java b/performance-tests/src/main/java/io/kroxylicious/proxy/filter/SpecificFilterInvoker.java
similarity index 77%
rename from performance-tests/src/main/java/io/kroxylicious/SpcificFilterInvoker.java
rename to performance-tests/src/main/java/io/kroxylicious/proxy/filter/SpecificFilterInvoker.java
index 1061bc1c..d3d586dd 100644
--- a/performance-tests/src/main/java/io/kroxylicious/SpcificFilterInvoker.java
+++ b/performance-tests/src/main/java/io/kroxylicious/proxy/filter/SpecificFilterInvoker.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious;
+package io.kroxylicious.proxy.filter;
 
 import org.apache.kafka.common.message.AddOffsetsToTxnRequestData;
 import org.apache.kafka.common.message.AddOffsetsToTxnResponseData;
@@ -147,155 +147,15 @@ import org.apache.kafka.common.message.WriteTxnMarkersResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ApiMessage;
 
-import io.kroxylicious.proxy.filter.AddOffsetsToTxnRequestFilter;
-import io.kroxylicious.proxy.filter.AddOffsetsToTxnResponseFilter;
-import io.kroxylicious.proxy.filter.AddPartitionsToTxnRequestFilter;
-import io.kroxylicious.proxy.filter.AddPartitionsToTxnResponseFilter;
-import io.kroxylicious.proxy.filter.AllocateProducerIdsRequestFilter;
-import io.kroxylicious.proxy.filter.AllocateProducerIdsResponseFilter;
-import io.kroxylicious.proxy.filter.AlterClientQuotasRequestFilter;
-import io.kroxylicious.proxy.filter.AlterClientQuotasResponseFilter;
-import io.kroxylicious.proxy.filter.AlterConfigsRequestFilter;
-import io.kroxylicious.proxy.filter.AlterConfigsResponseFilter;
-import io.kroxylicious.proxy.filter.AlterPartitionReassignmentsRequestFilter;
-import io.kroxylicious.proxy.filter.AlterPartitionReassignmentsResponseFilter;
-import io.kroxylicious.proxy.filter.AlterPartitionRequestFilter;
-import io.kroxylicious.proxy.filter.AlterPartitionResponseFilter;
-import io.kroxylicious.proxy.filter.AlterReplicaLogDirsRequestFilter;
-import io.kroxylicious.proxy.filter.AlterReplicaLogDirsResponseFilter;
-import io.kroxylicious.proxy.filter.AlterUserScramCredentialsRequestFilter;
-import io.kroxylicious.proxy.filter.AlterUserScramCredentialsResponseFilter;
-import io.kroxylicious.proxy.filter.ApiVersionsRequestFilter;
-import io.kroxylicious.proxy.filter.ApiVersionsResponseFilter;
-import io.kroxylicious.proxy.filter.BeginQuorumEpochRequestFilter;
-import io.kroxylicious.proxy.filter.BeginQuorumEpochResponseFilter;
-import io.kroxylicious.proxy.filter.BrokerHeartbeatRequestFilter;
-import io.kroxylicious.proxy.filter.BrokerHeartbeatResponseFilter;
-import io.kroxylicious.proxy.filter.BrokerRegistrationRequestFilter;
-import io.kroxylicious.proxy.filter.BrokerRegistrationResponseFilter;
-import io.kroxylicious.proxy.filter.ControlledShutdownRequestFilter;
-import io.kroxylicious.proxy.filter.ControlledShutdownResponseFilter;
-import io.kroxylicious.proxy.filter.CreateAclsRequestFilter;
-import io.kroxylicious.proxy.filter.CreateAclsResponseFilter;
-import io.kroxylicious.proxy.filter.CreateDelegationTokenRequestFilter;
-import io.kroxylicious.proxy.filter.CreateDelegationTokenResponseFilter;
-import io.kroxylicious.proxy.filter.CreatePartitionsRequestFilter;
-import io.kroxylicious.proxy.filter.CreatePartitionsResponseFilter;
-import io.kroxylicious.proxy.filter.CreateTopicsRequestFilter;
-import io.kroxylicious.proxy.filter.CreateTopicsResponseFilter;
-import io.kroxylicious.proxy.filter.DeleteAclsRequestFilter;
-import io.kroxylicious.proxy.filter.DeleteAclsResponseFilter;
-import io.kroxylicious.proxy.filter.DeleteGroupsRequestFilter;
-import io.kroxylicious.proxy.filter.DeleteGroupsResponseFilter;
-import io.kroxylicious.proxy.filter.DeleteRecordsRequestFilter;
-import io.kroxylicious.proxy.filter.DeleteRecordsResponseFilter;
-import io.kroxylicious.proxy.filter.DeleteTopicsRequestFilter;
-import io.kroxylicious.proxy.filter.DeleteTopicsResponseFilter;
-import io.kroxylicious.proxy.filter.DescribeAclsRequestFilter;
-import io.kroxylicious.proxy.filter.DescribeAclsResponseFilter;
-import io.kroxylicious.proxy.filter.DescribeClientQuotasRequestFilter;
-import io.kroxylicious.proxy.filter.DescribeClientQuotasResponseFilter;
-import io.kroxylicious.proxy.filter.DescribeClusterRequestFilter;
-import io.kroxylicious.proxy.filter.DescribeClusterResponseFilter;
-import io.kroxylicious.proxy.filter.DescribeConfigsRequestFilter;
-import io.kroxylicious.proxy.filter.DescribeConfigsResponseFilter;
-import io.kroxylicious.proxy.filter.DescribeDelegationTokenRequestFilter;
-import io.kroxylicious.proxy.filter.DescribeDelegationTokenResponseFilter;
-import io.kroxylicious.proxy.filter.DescribeGroupsRequestFilter;
-import io.kroxylicious.proxy.filter.DescribeGroupsResponseFilter;
-import io.kroxylicious.proxy.filter.DescribeLogDirsRequestFilter;
-import io.kroxylicious.proxy.filter.DescribeLogDirsResponseFilter;
-import io.kroxylicious.proxy.filter.DescribeProducersRequestFilter;
-import io.kroxylicious.proxy.filter.DescribeProducersResponseFilter;
-import io.kroxylicious.proxy.filter.DescribeQuorumRequestFilter;
-import io.kroxylicious.proxy.filter.DescribeQuorumResponseFilter;
-import io.kroxylicious.proxy.filter.DescribeTransactionsRequestFilter;
-import io.kroxylicious.proxy.filter.DescribeTransactionsResponseFilter;
-import io.kroxylicious.proxy.filter.DescribeUserScramCredentialsRequestFilter;
-import io.kroxylicious.proxy.filter.DescribeUserScramCredentialsResponseFilter;
-import io.kroxylicious.proxy.filter.ElectLeadersRequestFilter;
-import io.kroxylicious.proxy.filter.ElectLeadersResponseFilter;
-import io.kroxylicious.proxy.filter.EndQuorumEpochRequestFilter;
-import io.kroxylicious.proxy.filter.EndQuorumEpochResponseFilter;
-import io.kroxylicious.proxy.filter.EndTxnRequestFilter;
-import io.kroxylicious.proxy.filter.EndTxnResponseFilter;
-import io.kroxylicious.proxy.filter.EnvelopeRequestFilter;
-import io.kroxylicious.proxy.filter.EnvelopeResponseFilter;
-import io.kroxylicious.proxy.filter.ExpireDelegationTokenRequestFilter;
-import io.kroxylicious.proxy.filter.ExpireDelegationTokenResponseFilter;
-import io.kroxylicious.proxy.filter.FetchRequestFilter;
-import io.kroxylicious.proxy.filter.FetchResponseFilter;
-import io.kroxylicious.proxy.filter.FetchSnapshotRequestFilter;
-import io.kroxylicious.proxy.filter.FetchSnapshotResponseFilter;
-import io.kroxylicious.proxy.filter.FilterInvoker;
-import io.kroxylicious.proxy.filter.FindCoordinatorRequestFilter;
-import io.kroxylicious.proxy.filter.FindCoordinatorResponseFilter;
-import io.kroxylicious.proxy.filter.HeartbeatRequestFilter;
-import io.kroxylicious.proxy.filter.HeartbeatResponseFilter;
-import io.kroxylicious.proxy.filter.IncrementalAlterConfigsRequestFilter;
-import io.kroxylicious.proxy.filter.IncrementalAlterConfigsResponseFilter;
-import io.kroxylicious.proxy.filter.InitProducerIdRequestFilter;
-import io.kroxylicious.proxy.filter.InitProducerIdResponseFilter;
-import io.kroxylicious.proxy.filter.JoinGroupRequestFilter;
-import io.kroxylicious.proxy.filter.JoinGroupResponseFilter;
-import io.kroxylicious.proxy.filter.KrpcFilter;
-import io.kroxylicious.proxy.filter.KrpcFilterContext;
-import io.kroxylicious.proxy.filter.LeaderAndIsrRequestFilter;
-import io.kroxylicious.proxy.filter.LeaderAndIsrResponseFilter;
-import io.kroxylicious.proxy.filter.LeaveGroupRequestFilter;
-import io.kroxylicious.proxy.filter.LeaveGroupResponseFilter;
-import io.kroxylicious.proxy.filter.ListGroupsRequestFilter;
-import io.kroxylicious.proxy.filter.ListGroupsResponseFilter;
-import io.kroxylicious.proxy.filter.ListOffsetsRequestFilter;
-import io.kroxylicious.proxy.filter.ListOffsetsResponseFilter;
-import io.kroxylicious.proxy.filter.ListPartitionReassignmentsRequestFilter;
-import io.kroxylicious.proxy.filter.ListPartitionReassignmentsResponseFilter;
-import io.kroxylicious.proxy.filter.ListTransactionsRequestFilter;
-import io.kroxylicious.proxy.filter.ListTransactionsResponseFilter;
-import io.kroxylicious.proxy.filter.MetadataRequestFilter;
-import io.kroxylicious.proxy.filter.MetadataResponseFilter;
-import io.kroxylicious.proxy.filter.OffsetCommitRequestFilter;
-import io.kroxylicious.proxy.filter.OffsetCommitResponseFilter;
-import io.kroxylicious.proxy.filter.OffsetDeleteRequestFilter;
-import io.kroxylicious.proxy.filter.OffsetDeleteResponseFilter;
-import io.kroxylicious.proxy.filter.OffsetFetchRequestFilter;
-import io.kroxylicious.proxy.filter.OffsetFetchResponseFilter;
-import io.kroxylicious.proxy.filter.OffsetForLeaderEpochRequestFilter;
-import io.kroxylicious.proxy.filter.OffsetForLeaderEpochResponseFilter;
-import io.kroxylicious.proxy.filter.ProduceRequestFilter;
-import io.kroxylicious.proxy.filter.ProduceResponseFilter;
-import io.kroxylicious.proxy.filter.RenewDelegationTokenRequestFilter;
-import io.kroxylicious.proxy.filter.RenewDelegationTokenResponseFilter;
-import io.kroxylicious.proxy.filter.SaslAuthenticateRequestFilter;
-import io.kroxylicious.proxy.filter.SaslAuthenticateResponseFilter;
-import io.kroxylicious.proxy.filter.SaslHandshakeRequestFilter;
-import io.kroxylicious.proxy.filter.SaslHandshakeResponseFilter;
-import io.kroxylicious.proxy.filter.StopReplicaRequestFilter;
-import io.kroxylicious.proxy.filter.StopReplicaResponseFilter;
-import io.kroxylicious.proxy.filter.SyncGroupRequestFilter;
-import io.kroxylicious.proxy.filter.SyncGroupResponseFilter;
-import io.kroxylicious.proxy.filter.TxnOffsetCommitRequestFilter;
-import io.kroxylicious.proxy.filter.TxnOffsetCommitResponseFilter;
-import io.kroxylicious.proxy.filter.UnregisterBrokerRequestFilter;
-import io.kroxylicious.proxy.filter.UnregisterBrokerResponseFilter;
-import io.kroxylicious.proxy.filter.UpdateFeaturesRequestFilter;
-import io.kroxylicious.proxy.filter.UpdateFeaturesResponseFilter;
-import io.kroxylicious.proxy.filter.UpdateMetadataRequestFilter;
-import io.kroxylicious.proxy.filter.UpdateMetadataResponseFilter;
-import io.kroxylicious.proxy.filter.VoteRequestFilter;
-import io.kroxylicious.proxy.filter.VoteResponseFilter;
-import io.kroxylicious.proxy.filter.WriteTxnMarkersRequestFilter;
-import io.kroxylicious.proxy.filter.WriteTxnMarkersResponseFilter;
-
 /**
  * Invoker for KrpcFilters that implement any number of Specific Message interfaces (for
  * example {@link io.kroxylicious.proxy.filter.AlterConfigsResponseFilter}.
  */
-class SpecificFilterInvoker implements FilterInvoker {
+public class SpecificFilterInvoker implements FilterInvoker {
 
     private final KrpcFilter filter;
 
-    SpecificFilterInvoker(KrpcFilter filter) {
+    public SpecificFilterInvoker(KrpcFilter filter) {
         this.filter = filter;
     }
 
@@ -805,149 +665,4 @@ class SpecificFilterInvoker implements FilterInvoker {
             default -> throw new IllegalStateException("Unsupported API key " + apiKey);
         };
     }
-
-    /**
-     * Check if a KrpcFilter implements any of the Specific Message Filter interfaces
-     * @param filter the filter
-     * @return true if the filter implements any Specific Message Filter interfaces
-     */
-    public static boolean implementsAnySpecificFilterInterface(KrpcFilter filter) {
-        return filter instanceof AddOffsetsToTxnRequestFilter ||
-                filter instanceof AddOffsetsToTxnResponseFilter ||
-                filter instanceof AddPartitionsToTxnRequestFilter ||
-                filter instanceof AddPartitionsToTxnResponseFilter ||
-                filter instanceof AllocateProducerIdsRequestFilter ||
-                filter instanceof AllocateProducerIdsResponseFilter ||
-                filter instanceof AlterClientQuotasRequestFilter ||
-                filter instanceof AlterClientQuotasResponseFilter ||
-                filter instanceof AlterConfigsRequestFilter ||
-                filter instanceof AlterConfigsResponseFilter ||
-                filter instanceof AlterPartitionReassignmentsRequestFilter ||
-                filter instanceof AlterPartitionReassignmentsResponseFilter ||
-                filter instanceof AlterPartitionRequestFilter ||
-                filter instanceof AlterPartitionResponseFilter ||
-                filter instanceof AlterReplicaLogDirsRequestFilter ||
-                filter instanceof AlterReplicaLogDirsResponseFilter ||
-                filter instanceof AlterUserScramCredentialsRequestFilter ||
-                filter instanceof AlterUserScramCredentialsResponseFilter ||
-                filter instanceof ApiVersionsRequestFilter ||
-                filter instanceof ApiVersionsResponseFilter ||
-                filter instanceof BeginQuorumEpochRequestFilter ||
-                filter instanceof BeginQuorumEpochResponseFilter ||
-                filter instanceof BrokerHeartbeatRequestFilter ||
-                filter instanceof BrokerHeartbeatResponseFilter ||
-                filter instanceof BrokerRegistrationRequestFilter ||
-                filter instanceof BrokerRegistrationResponseFilter ||
-                filter instanceof ControlledShutdownRequestFilter ||
-                filter instanceof ControlledShutdownResponseFilter ||
-                filter instanceof CreateAclsRequestFilter ||
-                filter instanceof CreateAclsResponseFilter ||
-                filter instanceof CreateDelegationTokenRequestFilter ||
-                filter instanceof CreateDelegationTokenResponseFilter ||
-                filter instanceof CreatePartitionsRequestFilter ||
-                filter instanceof CreatePartitionsResponseFilter ||
-                filter instanceof CreateTopicsRequestFilter ||
-                filter instanceof CreateTopicsResponseFilter ||
-                filter instanceof DeleteAclsRequestFilter ||
-                filter instanceof DeleteAclsResponseFilter ||
-                filter instanceof DeleteGroupsRequestFilter ||
-                filter instanceof DeleteGroupsResponseFilter ||
-                filter instanceof DeleteRecordsRequestFilter ||
-                filter instanceof DeleteRecordsResponseFilter ||
-                filter instanceof DeleteTopicsRequestFilter ||
-                filter instanceof DeleteTopicsResponseFilter ||
-                filter instanceof DescribeAclsRequestFilter ||
-                filter instanceof DescribeAclsResponseFilter ||
-                filter instanceof DescribeClientQuotasRequestFilter ||
-                filter instanceof DescribeClientQuotasResponseFilter ||
-                filter instanceof DescribeClusterRequestFilter ||
-                filter instanceof DescribeClusterResponseFilter ||
-                filter instanceof DescribeConfigsRequestFilter ||
-                filter instanceof DescribeConfigsResponseFilter ||
-                filter instanceof DescribeDelegationTokenRequestFilter ||
-                filter instanceof DescribeDelegationTokenResponseFilter ||
-                filter instanceof DescribeGroupsRequestFilter ||
-                filter instanceof DescribeGroupsResponseFilter ||
-                filter instanceof DescribeLogDirsRequestFilter ||
-                filter instanceof DescribeLogDirsResponseFilter ||
-                filter instanceof DescribeProducersRequestFilter ||
-                filter instanceof DescribeProducersResponseFilter ||
-                filter instanceof DescribeQuorumRequestFilter ||
-                filter instanceof DescribeQuorumResponseFilter ||
-                filter instanceof DescribeTransactionsRequestFilter ||
-                filter instanceof DescribeTransactionsResponseFilter ||
-                filter instanceof DescribeUserScramCredentialsRequestFilter ||
-                filter instanceof DescribeUserScramCredentialsResponseFilter ||
-                filter instanceof ElectLeadersRequestFilter ||
-                filter instanceof ElectLeadersResponseFilter ||
-                filter instanceof EndQuorumEpochRequestFilter ||
-                filter instanceof EndQuorumEpochResponseFilter ||
-                filter instanceof EndTxnRequestFilter ||
-                filter instanceof EndTxnResponseFilter ||
-                filter instanceof EnvelopeRequestFilter ||
-                filter instanceof EnvelopeResponseFilter ||
-                filter instanceof ExpireDelegationTokenRequestFilter ||
-                filter instanceof ExpireDelegationTokenResponseFilter ||
-                filter instanceof FetchRequestFilter ||
-                filter instanceof FetchResponseFilter ||
-                filter instanceof FetchSnapshotRequestFilter ||
-                filter instanceof FetchSnapshotResponseFilter ||
-                filter instanceof FindCoordinatorRequestFilter ||
-                filter instanceof FindCoordinatorResponseFilter ||
-                filter instanceof HeartbeatRequestFilter ||
-                filter instanceof HeartbeatResponseFilter ||
-                filter instanceof IncrementalAlterConfigsRequestFilter ||
-                filter instanceof IncrementalAlterConfigsResponseFilter ||
-                filter instanceof InitProducerIdRequestFilter ||
-                filter instanceof InitProducerIdResponseFilter ||
-                filter instanceof JoinGroupRequestFilter ||
-                filter instanceof JoinGroupResponseFilter ||
-                filter instanceof LeaderAndIsrRequestFilter ||
-                filter instanceof LeaderAndIsrResponseFilter ||
-                filter instanceof LeaveGroupRequestFilter ||
-                filter instanceof LeaveGroupResponseFilter ||
-                filter instanceof ListGroupsRequestFilter ||
-                filter instanceof ListGroupsResponseFilter ||
-                filter instanceof ListOffsetsRequestFilter ||
-                filter instanceof ListOffsetsResponseFilter ||
-                filter instanceof ListPartitionReassignmentsRequestFilter ||
-                filter instanceof ListPartitionReassignmentsResponseFilter ||
-                filter instanceof ListTransactionsRequestFilter ||
-                filter instanceof ListTransactionsResponseFilter ||
-                filter instanceof MetadataRequestFilter ||
-                filter instanceof MetadataResponseFilter ||
-                filter instanceof OffsetCommitRequestFilter ||
-                filter instanceof OffsetCommitResponseFilter ||
-                filter instanceof OffsetDeleteRequestFilter ||
-                filter instanceof OffsetDeleteResponseFilter ||
-                filter instanceof OffsetFetchRequestFilter ||
-                filter instanceof OffsetFetchResponseFilter ||
-                filter instanceof OffsetForLeaderEpochRequestFilter ||
-                filter instanceof OffsetForLeaderEpochResponseFilter ||
-                filter instanceof ProduceRequestFilter ||
-                filter instanceof ProduceResponseFilter ||
-                filter instanceof RenewDelegationTokenRequestFilter ||
-                filter instanceof RenewDelegationTokenResponseFilter ||
-                filter instanceof SaslAuthenticateRequestFilter ||
-                filter instanceof SaslAuthenticateResponseFilter ||
-                filter instanceof SaslHandshakeRequestFilter ||
-                filter instanceof SaslHandshakeResponseFilter ||
-                filter instanceof StopReplicaRequestFilter ||
-                filter instanceof StopReplicaResponseFilter ||
-                filter instanceof SyncGroupRequestFilter ||
-                filter instanceof SyncGroupResponseFilter ||
-                filter instanceof TxnOffsetCommitRequestFilter ||
-                filter instanceof TxnOffsetCommitResponseFilter ||
-                filter instanceof UnregisterBrokerRequestFilter ||
-                filter instanceof UnregisterBrokerResponseFilter ||
-                filter instanceof UpdateFeaturesRequestFilter ||
-                filter instanceof UpdateFeaturesResponseFilter ||
-                filter instanceof UpdateMetadataRequestFilter ||
-                filter instanceof UpdateMetadataResponseFilter ||
-                filter instanceof VoteRequestFilter ||
-                filter instanceof VoteResponseFilter ||
-                filter instanceof WriteTxnMarkersRequestFilter ||
-                filter instanceof WriteTxnMarkersResponseFilter;
-    }
-
 }