Repository: kroxylicious/kroxylicious
Pull request #354 (2023-05-31 09:49:20)

Remove ClusterEndpointConfigProvider from the Contributor instance methods
### Type of change

_Select the type of your PR_

- Refactoring

### Description

Why: The use case for `ClusterEndpointConfigProvider` being available to `Contributor` implementations was the `BrokerAddressFilter`.  Now BAF is an internal concern, that part of the API is redundant.


### Additional Context

_Why are you making this pull request?_


k-wall: @SamBarker I'll make the rename of ClusterEndpointConfigProvider after this is landed.

Patch:
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java
index 983af51a..7e975273 100644
--- a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java
+++ b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java
@@ -7,7 +7,6 @@ package io.kroxylicious.proxy.service;
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -37,28 +36,28 @@ public abstract class BaseContributor<T> implements Contributor<T> {
     }
 
     @Override
-    public T getInstance(String shortName, ClusterEndpointConfigProvider endpointConfigProvider, BaseConfig config) {
+    public T getInstance(String shortName, BaseConfig config) {
         InstanceBuilder<? extends BaseConfig, T> instanceBuilder = shortNameToInstanceBuilder.get(shortName);
-        return instanceBuilder == null ? null : instanceBuilder.construct(endpointConfigProvider, config);
+        return instanceBuilder == null ? null : instanceBuilder.construct(config);
     }
 
     private static class InstanceBuilder<T extends BaseConfig, L> {
 
         private final Class<T> configClass;
-        private final BiFunction<ClusterEndpointConfigProvider, T, L> instanceFunction;
+        private final Function<T, L> instanceFunction;
 
-        InstanceBuilder(Class<T> configClass, BiFunction<ClusterEndpointConfigProvider, T, L> instanceFunction) {
+        InstanceBuilder(Class<T> configClass, Function<T, L> instanceFunction) {
             this.configClass = configClass;
             this.instanceFunction = instanceFunction;
         }
 
-        L construct(ClusterEndpointConfigProvider endpointConfigProvider, BaseConfig config) {
+        L construct(BaseConfig config) {
             if (config == null) {
                 // tests pass in a null config, which some instance functions can tolerate
-                return instanceFunction.apply(endpointConfigProvider, null);
+                return instanceFunction.apply(null);
             }
             else if (configClass.isAssignableFrom(config.getClass())) {
-                return instanceFunction.apply(endpointConfigProvider, configClass.cast(config));
+                return instanceFunction.apply(configClass.cast(config));
             }
             else {
                 throw new IllegalArgumentException("config has the wrong type, expected "
@@ -80,7 +79,7 @@ public abstract class BaseContributor<T> implements Contributor<T> {
         private final Map<String, InstanceBuilder<?, L>> shortNameToInstanceBuilder = new HashMap<>();
 
         /**
-         * Registers registers a factory function for the construction of a service instance.
+         * Registers a factory function for the construction of a service instance.
          *
          * @param shortName service short name
          * @param configClass concrete type of configuration required by the service
@@ -88,8 +87,7 @@ public abstract class BaseContributor<T> implements Contributor<T> {
          * @return this
          * @param <T> the configuration concrete type
          */
-        public <T extends BaseConfig> BaseContributorBuilder<L> add(String shortName, Class<T> configClass,
-                                                                    BiFunction<ClusterEndpointConfigProvider, T, L> instanceFunction) {
+        public <T extends BaseConfig> BaseContributorBuilder<L> add(String shortName, Class<T> configClass, Function<T, L> instanceFunction) {
             if (shortNameToInstanceBuilder.containsKey(shortName)) {
                 throw new IllegalArgumentException(shortName + " already registered");
             }
@@ -98,41 +96,14 @@ public abstract class BaseContributor<T> implements Contributor<T> {
         }
 
         /**
-         * Registers registers a factory function for the construction of a service instance.
-         *
-         * @param shortName service short name
-         * @param instanceFunction function that constructs the service instance
-         * @return this
-         */
-        public BaseContributorBuilder<L> add(String shortName, Function<ClusterEndpointConfigProvider, L> instanceFunction) {
-            add(shortName, BaseConfig.class, (proxyConfig, config) -> instanceFunction.apply(proxyConfig));
-            return this;
-        }
-
-        /**
-         * Registers registers a factory function for the construction of a service instance.
-         *
-         * @param shortName service short name
-         * @param configClass concrete type of configuration required by the service
-         * @param instanceFunction function that constructs the service instance
-         * @return this
-         * @param <T> the configuration concrete type
-         */
-        public <T extends BaseConfig> BaseContributorBuilder<L> add(String shortName, Class<T> configClass, Function<T, L> instanceFunction) {
-            add(shortName, configClass, (proxyConfig, config) -> instanceFunction.apply(config));
-            return this;
-        }
-
-        /**
-         * Registers registers a factory function for the construction of a service instance.
+         * Registers a factory function for the construction of a service instance.
          *
          * @param shortName service short name
          * @param instanceFunction function that constructs the service instance
          * @return this
          */
         public BaseContributorBuilder<L> add(String shortName, Supplier<L> instanceFunction) {
-            add(shortName, BaseConfig.class, (proxyConfig, config) -> instanceFunction.get());
-            return this;
+            return add(shortName, BaseConfig.class, (config) -> instanceFunction.get());
         }
 
         Map<String, InstanceBuilder<?, L>> build() {
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/Contributor.java b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/Contributor.java
index 912cb783..baf094bd 100644
--- a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/Contributor.java
+++ b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/Contributor.java
@@ -27,9 +27,8 @@ public interface Contributor<T> {
      * Creates an instance of the service.
      *
      * @param shortName service short name
-     * @param endpointConfigProvider endpoint provider
-     * @param config service configuration which may be null if the service instance does not accept configuration.
+     * @param config    service configuration which may be null if the service instance does not accept configuration.
      * @return the service instance
      */
-    T getInstance(String shortName, ClusterEndpointConfigProvider endpointConfigProvider, BaseConfig config);
+    T getInstance(String shortName, BaseConfig config);
 }
diff --git a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/HostPort.java b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/HostPort.java
index 31e07206..f262a7a3 100644
--- a/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/HostPort.java
+++ b/api/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/HostPort.java
@@ -25,6 +25,8 @@ public final class HostPort {
 
     /**
      * Creates a host port.
+     * @param host Symbolic hostnames, FQDNs, IPv4, and IPv6 forms are supported
+     * @param port port number
      */
     public HostPort(String host, int port) {
         Objects.requireNonNull(host, "host cannot be null");
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java
index a7958ed5..bcfcfde4 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java
@@ -8,7 +8,6 @@ package io.kroxylicious.proxy.bootstrap;
 import java.util.List;
 
 import io.kroxylicious.proxy.config.Configuration;
-import io.kroxylicious.proxy.config.VirtualCluster;
 import io.kroxylicious.proxy.filter.KrpcFilter;
 import io.kroxylicious.proxy.internal.filter.FilterContributorManager;
 
@@ -20,11 +19,9 @@ import io.kroxylicious.proxy.internal.filter.FilterContributorManager;
 public class FilterChainFactory {
 
     private final Configuration config;
-    private final VirtualCluster virtualCluster;
 
-    public FilterChainFactory(Configuration config, VirtualCluster virtualCluster) {
+    public FilterChainFactory(Configuration config) {
         this.config = config;
-        this.virtualCluster = virtualCluster;
     }
 
     /**
@@ -37,7 +34,7 @@ public class FilterChainFactory {
 
         return config.filters()
                 .stream()
-                .map(f -> filterContributorManager.getFilter(f.type(), virtualCluster.getClusterEndpointProvider(), f.config()))
+                .map(f -> filterContributorManager.getFilter(f.type(), f.config()))
                 .toList();
     }
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
index 2703580c..434fd9ff 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java
@@ -177,33 +177,25 @@ class DefaultFilterContext implements KrpcFilterContext {
     @Override
     public void forwardResponse(ResponseHeaderData header, ApiMessage response) {
         // check it's a response
-        try {
-            String name = response.getClass().getName();
-            if (!name.endsWith("ResponseData")) {
-                throw new AssertionError("Attempt to use forwardResponse with a non-response: " + name);
+        String name = response.getClass().getName();
+        if (!name.endsWith("ResponseData")) {
+            throw new AssertionError("Attempt to use forwardResponse with a non-response: " + name);
+        }
+        if (decodedFrame instanceof RequestFrame) {
+            forwardShortCircuitResponse(header, response);
+        }
+        else {
+            // TODO check we've not forwarded it already
+            if (decodedFrame.body() != response) {
+                throw new AssertionError();
             }
-            if (decodedFrame instanceof RequestFrame) {
-                forwardShortCircuitResponse(header, response);
+            if (decodedFrame.header() != header) {
+                throw new AssertionError();
             }
-            else {
-                // TODO check we've not forwarded it already
-                if (decodedFrame.body() != response) {
-                    throw new AssertionError();
-                }
-                if (decodedFrame.header() != header) {
-                    throw new AssertionError();
-                }
-                if (LOGGER.isDebugEnabled()) {
-                    LOGGER.debug("{}: Forwarding response: {}", channelDescriptor(), decodedFrame);
-                }
-                channelContext.fireChannelRead(decodedFrame);
-            }
-        }
-        finally {
-            if (!channelContext.executor().inEventLoop()) {
-                // required to flush the message back to the client
-                channelContext.fireChannelReadComplete();
+            if (LOGGER.isDebugEnabled()) {
+                LOGGER.debug("{}: Forwarding response: {}", channelDescriptor(), decodedFrame);
             }
+            channelContext.fireChannelRead(decodedFrame);
         }
     }
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
index b9e1b8e1..60536cff 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
@@ -167,7 +167,7 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
         }
 
         var frontendHandler = new KafkaProxyFrontendHandler(context -> {
-            var filterChainFactory = new FilterChainFactory(config, virtualCluster);
+            var filterChainFactory = new FilterChainFactory(config);
 
             var filters = new ArrayList<>(filterChainFactory.createFilters());
             // Add internal 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
index e8adb887..b5e08607 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/ClusterEndpointConfigProviderContributorManager.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/clusterendpointprovider/ClusterEndpointConfigProviderContributorManager.java
@@ -41,7 +41,7 @@ public class ClusterEndpointConfigProviderContributorManager {
 
     public ClusterEndpointConfigProvider getClusterEndpointConfigProvider(String shortName, BaseConfig baseConfig) {
         for (ClusterEndpointProviderContributor contributor : contributors) {
-            var assigner = contributor.getInstance(shortName, null, baseConfig);
+            var assigner = contributor.getInstance(shortName, baseConfig);
             if (assigner != null) {
                 return assigner;
             }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FilterContributorManager.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FilterContributorManager.java
index be57ec58..b9cd8697 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FilterContributorManager.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FilterContributorManager.java
@@ -11,7 +11,6 @@ import java.util.ServiceLoader;
 import io.kroxylicious.proxy.config.BaseConfig;
 import io.kroxylicious.proxy.filter.FilterContributor;
 import io.kroxylicious.proxy.filter.KrpcFilter;
-import io.kroxylicious.proxy.service.ClusterEndpointConfigProvider;
 
 public class FilterContributorManager {
 
@@ -40,11 +39,11 @@ public class FilterContributorManager {
         throw new IllegalArgumentException("No filter found for name '" + shortName + "'");
     }
 
-    public KrpcFilter getFilter(String shortName, ClusterEndpointConfigProvider proxyConfig, BaseConfig filterConfig) {
+    public KrpcFilter getFilter(String shortName, BaseConfig filterConfig) {
         Iterator<FilterContributor> it = contributors.iterator();
         while (it.hasNext()) {
             FilterContributor contributor = it.next();
-            KrpcFilter filter = contributor.getInstance(shortName, proxyConfig, filterConfig);
+            KrpcFilter filter = contributor.getInstance(shortName, filterConfig);
             if (filter != null) {
                 return filter;
             }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/micrometer/MicrometerConfigurationHookContributorManager.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/micrometer/MicrometerConfigurationHookContributorManager.java
index f148a65e..30faa754 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/micrometer/MicrometerConfigurationHookContributorManager.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/micrometer/MicrometerConfigurationHookContributorManager.java
@@ -36,7 +36,7 @@ public class MicrometerConfigurationHookContributorManager {
 
     public MicrometerConfigurationHook getHook(String shortName, BaseConfig filterConfig) {
         for (MicrometerConfigurationHookContributor contributor : contributors) {
-            MicrometerConfigurationHook hook = contributor.getInstance(shortName, null, filterConfig);
+            MicrometerConfigurationHook hook = contributor.getInstance(shortName, filterConfig);
             if (hook != null) {
                 return hook;
             }


Test patch:
diff --git a/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java b/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java
index d0f505b3..91031ed6 100644
--- a/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java
+++ b/api/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java
@@ -11,7 +11,6 @@ import io.kroxylicious.proxy.config.BaseConfig;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.mockito.Mockito.mock;
 
 class BaseContributorTest {
     static class LongConfig extends BaseConfig {
@@ -38,7 +37,7 @@ class BaseContributorTest {
         builder.add("one", () -> 1L);
         BaseContributor<Long> baseContributor = new BaseContributor<>(builder) {
         };
-        Long instance = baseContributor.getInstance("one", mock(ClusterEndpointConfigProvider.class), new BaseConfig());
+        Long instance = baseContributor.getInstance("one", new BaseConfig());
         assertThat(instance).isEqualTo(1L);
     }
 
@@ -58,7 +57,7 @@ class BaseContributorTest {
         builder.add("fromBaseConfig", LongConfig.class, baseConfig -> baseConfig.value);
         BaseContributor<Long> baseContributor = new BaseContributor<>(builder) {
         };
-        Long instance = baseContributor.getInstance("fromBaseConfig", mock(ClusterEndpointConfigProvider.class), new LongConfig());
+        Long instance = baseContributor.getInstance("fromBaseConfig", new LongConfig());
         assertThat(instance).isEqualTo(2L);
     }
 
@@ -70,8 +69,8 @@ class BaseContributorTest {
         };
         AnotherConfig incompatibleConfig = new AnotherConfig();
         assertThatThrownBy(() -> {
-            baseContributor.getInstance("fromBaseConfig", mock(ClusterEndpointConfigProvider.class), incompatibleConfig);
+            baseContributor.getInstance("fromBaseConfig", incompatibleConfig);
         }).isInstanceOf(IllegalArgumentException.class);
     }
 
-}
\ No newline at end of file
+}
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 0995fa05..0a2ddadf 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
@@ -31,7 +31,6 @@ import org.apache.kafka.clients.admin.ConsumerGroupDescription;
 import org.apache.kafka.clients.admin.ConsumerGroupListing;
 import org.apache.kafka.clients.admin.CreateTopicsResult;
 import org.apache.kafka.clients.admin.DeleteTopicsResult;
-import org.apache.kafka.clients.admin.ListTopicsResult;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.admin.TopicListing;
@@ -135,9 +134,9 @@ public class MultiTenantIT {
                 var admin = tester.admin(TENANT_1_CLUSTER, commonConfig(Map.of()))) {
             var created = createTopics(admin, List.of(NEW_TOPIC_1, NEW_TOPIC_2));
 
-            ListTopicsResult listTopicsResult = admin.listTopics();
-            var topicMap = listTopicsResult.namesToListings().get();
-            assertEquals(2, topicMap.size());
+            var topicMap = await().atMost(Duration.ofSeconds(5)).until(() -> admin.listTopics().namesToListings().get(),
+                    n -> n.size() == 2);
+
             assertThat(topicMap).hasSize(2);
             assertThat(topicMap).hasEntrySatisfying(TOPIC_1,
                     allOf(matches(TopicListing::name, TOPIC_1), matches(TopicListing::topicId, created.topicId(TOPIC_1).get())));