Repository: kroxylicious/kroxylicious (java)

Issue:
Remove ApiVersionsService
Outcome of #641 was a deprecation of the class. It is now eligible for deletion as it will be more than 3 minor releases since 0.3.0

Pull request:
Remove api versions service
### Type of change

- Refactoring

### Description

Closes #1378

Remove ApiVersionsService
    
Deprecated since 0.3.0 and now eligible for destruction by our policy.
    
    We haven't used it yet and the utility of it to Filters is unclear. The
    original rationale was that Filters can hijack the connection and use
    sendRequest to send messages to the broker, messages constructed by the
    Filter. We need to ensure we only use fields that the broker
    understands, so by offering an API to access the supported versions a
    Filter could discover what's available and react to that.
    
    For now we will remove the deprecated class and can re-implement
    it if anyone needs this functionality. For now the Filters can work
    around it by intercepting the ApiVersionsResponse themselves which
    should work for the typical case where the Filter wants to do some work
    after the api versions are negotiated.

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [ ] Write tests
- [ ] Make sure all tests pass
- [ ] Review performance test results. Ensure that any degradations to performance numbers are understood and justified.
- [ ] Make sure all Sonarcloud warnings are addressed or are justifiably ignored.
- [ ] Update documentation
- [ ] Reference relevant issue(s) and close them after merging
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


Patch:
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/ApiVersionsService.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/ApiVersionsService.java
deleted file mode 100644
index 4410f4b1..00000000
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/ApiVersionsService.java
+++ /dev/null
@@ -1,45 +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;
-
-import java.util.Optional;
-import java.util.concurrent.CompletionStage;
-
-import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion;
-import org.apache.kafka.common.protocol.ApiKeys;
-
-import edu.umd.cs.findbugs.annotations.NonNull;
-
-/**
- * Service used to obtain the upstream cluster's ApiVersions and intersect them with the versions
- * supported by the proxy. Filter Authors can then use this information when sending extra requests
- * to the upstream cluster.
- * @deprecated functionality to be moved up into FilterContext
- */
-@Deprecated(since = "0.3.0", forRemoval = true)
-public interface ApiVersionsService {
-
-    /**
-     * Information about version ranges for an ApiKey supported by the upstream and Kroxylicious
-     * @param upstream the version range supported by the upstream server, or null if this key is not supported by upstream
-     * @param intersected the version range supported by both Kroxylicious and the upstream server, or null if this ApiKey is not supported
-     * @deprecated upstream versions not very useful, so can be replaced with a return type of ApiVersion and only return intersected
-     */
-    @Deprecated(since = "0.3.0", forRemoval = true)
-    record ApiVersionRanges(@NonNull ApiVersion upstream, @NonNull ApiVersion intersected) {}
-
-    /**
-     * Get the supported version ranges for an ApiKey. Will contain the upstream supported
-     * version range, and the intersected version range supported by the proxy and upstream.
-     * Filters will likely want to work with the intersected range as both the proxy and the
-     * upstream can parse those versions.
-     * @param keys keys
-     * @return a CompletionStage that will be completed with the upstream ApiVersionRanges,
-     * or an empty optional if the upstream doesn't support this key
-     */
-    CompletionStage<Optional<ApiVersionRanges>> getApiVersionRanges(ApiKeys keys);
-}
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContext.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContext.java
index 0081a3d0..cbaba8d6 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContext.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContext.java
@@ -14,8 +14,6 @@ import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiMessage;
 import org.apache.kafka.common.utils.ByteBufferOutputStream;
 
-import io.kroxylicious.proxy.ApiVersionsService;
-
 import edu.umd.cs.findbugs.annotations.NonNull;
 
 /**
@@ -142,10 +140,4 @@ public interface FilterContext {
     String getVirtualClusterName();
     // TODO an API to allow a filter to add/remove another filter from the pipeline
 
-    /**
-     * Gets a Service responsible for exposing the API versions supported by the
-     * upstream broker and kroxylicious
-     * @return apiVersionsService
-     */
-    ApiVersionsService getApiVersionsService();
 }
diff --git a/kroxylicious-microbenchmarks/src/main/java/io/kroxylicious/microbenchmarks/InvokerDispatchBenchmark.java b/kroxylicious-microbenchmarks/src/main/java/io/kroxylicious/microbenchmarks/InvokerDispatchBenchmark.java
index cbbffda1..59ec845c 100644
--- a/kroxylicious-microbenchmarks/src/main/java/io/kroxylicious/microbenchmarks/InvokerDispatchBenchmark.java
+++ b/kroxylicious-microbenchmarks/src/main/java/io/kroxylicious/microbenchmarks/InvokerDispatchBenchmark.java
@@ -33,7 +33,6 @@ import io.kroxylicious.filters.FourInterfaceFilter0;
 import io.kroxylicious.filters.FourInterfaceFilter1;
 import io.kroxylicious.filters.FourInterfaceFilter2;
 import io.kroxylicious.filters.FourInterfaceFilter3;
-import io.kroxylicious.proxy.ApiVersionsService;
 import io.kroxylicious.proxy.filter.ArrayFilterInvoker;
 import io.kroxylicious.proxy.filter.Filter;
 import io.kroxylicious.proxy.filter.FilterContext;
@@ -179,11 +178,6 @@ public class InvokerDispatchBenchmark {
             return null;
         }
 
-        @Override
-        public ApiVersionsService getApiVersionsService() {
-            return null;
-        }
-
         @Override
         public <M extends ApiMessage> CompletionStage<M> sendRequest(RequestHeaderData header, ApiMessage request) {
             return null;
diff --git a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImpl.java b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImpl.java
index 1fa2d10d..510c4497 100644
--- a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImpl.java
+++ b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImpl.java
@@ -7,36 +7,20 @@
 package io.kroxylicious.proxy.internal;
 
 import java.util.HashSet;
-import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
 
-import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.ApiVersionsResponseData;
 import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion;
-import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersionCollection;
-import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.Errors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.kroxylicious.proxy.ApiVersionsService;
-import io.kroxylicious.proxy.filter.FilterContext;
-
 public class ApiVersionsServiceImpl {
 
-    private record ApiVersions(ApiVersionCollection upstream, ApiVersionCollection intersected) {}
-
     private static final Logger LOGGER = LoggerFactory.getLogger(ApiVersionsServiceImpl.class);
-    private ApiVersions apiVersions = null;
 
-    public void updateVersions(String channel, ApiVersionsResponseData upstreamApiVersions) {
-        var upstream = upstreamApiVersions.duplicate().apiKeys();
-        intersectApiVersions(channel, upstreamApiVersions);
-        var intersected = upstreamApiVersions.duplicate().apiKeys();
-        this.apiVersions = new ApiVersions(upstream, intersected);
+    public void updateVersions(String channel, ApiVersionsResponseData apiVersionsResponse) {
+        intersectApiVersions(channel, apiVersionsResponse);
     }
 
     private static void intersectApiVersions(String channel, ApiVersionsResponseData resp) {
@@ -85,51 +69,4 @@ public class ApiVersionsServiceImpl {
         }
     }
 
-    public CompletionStage<Optional<ApiVersionsService.ApiVersionRanges>> getApiVersionRanges(ApiKeys keys, FilterContext context) {
-        return getVersions(context).thenApply(versions -> {
-            ApiVersion upstream = versions.upstream.find(keys.id);
-            ApiVersion intersected = versions.intersected.find(keys.id);
-            if (upstream == null || intersected == null) {
-                return Optional.empty();
-            }
-            else {
-                return Optional.of(new ApiVersionsService.ApiVersionRanges(upstream, intersected));
-            }
-        });
-    }
-
-    private CompletionStage<ApiVersions> getVersions(FilterContext context) {
-        if (apiVersions != null) {
-            return CompletableFuture.completedFuture(apiVersions);
-        }
-
-        // KIP-511 when the client receives an unsupported version for the ApiVersionResponse, it fails back to version 0
-        // Use the same algorithm as https://github.com/apache/kafka/blob/159d25a7df25975694e2e0eb18a8feb125f7c39e/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java#L957-L977
-        var data = new ApiVersionsRequestData();
-        var header = new RequestHeaderData().setRequestApiVersion(data.highestSupportedVersion());
-        return context.<ApiVersionsResponseData> sendRequest(header, data)
-                .thenCompose(response -> {
-                    if (response.errorCode() != Errors.NONE.code()) {
-                        if (header.requestApiVersion() == 0 || response.errorCode() != Errors.UNSUPPORTED_VERSION.code()) {
-                            throw new IllegalStateException("Received error " + Errors.forCode(response.errorCode()) +
-                                    " when making an ApiVersionsRequest with correlation id " + header.correlationId() + ".");
-                        }
-
-                        short maxApiVersion = 0;
-                        if (!response.apiKeys().isEmpty()) {
-                            ApiVersion apiVersion = response.apiKeys().find(ApiKeys.API_VERSIONS.id);
-                            if (apiVersion != null) {
-                                maxApiVersion = apiVersion.maxVersion();
-                            }
-                        }
-                        return context.sendRequest(header.setRequestApiVersion(maxApiVersion), data);
-                    }
-                    return CompletableFuture.completedStage(response);
-                })
-                .thenApply(response -> {
-                    updateVersions(context.channelDescriptor(), response);
-                    return apiVersions;
-                });
-    }
-
 }
diff --git a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
index 238fcf7e..ab5b19ef 100644
--- a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
+++ b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
@@ -6,7 +6,6 @@
 package io.kroxylicious.proxy.internal;
 
 import java.util.Objects;
-import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.TimeUnit;
@@ -29,7 +28,6 @@ import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 
-import io.kroxylicious.proxy.ApiVersionsService;
 import io.kroxylicious.proxy.filter.Filter;
 import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.filter.FilterContext;
@@ -66,21 +64,18 @@ public class FilterHandler extends ChannelDuplexHandler {
     private final String sniHostname;
     private final VirtualCluster virtualCluster;
     private final Channel inboundChannel;
-    private final ApiVersionsServiceImpl apiVersionService;
     private CompletableFuture<Void> writeFuture = CompletableFuture.completedFuture(null);
     private CompletableFuture<Void> readFuture = CompletableFuture.completedFuture(null);
     private ChannelHandlerContext ctx;
     private PromiseFactory promiseFactory;
 
-    public FilterHandler(FilterAndInvoker filterAndInvoker, long timeoutMs, String sniHostname, VirtualCluster virtualCluster, Channel inboundChannel,
-                         ApiVersionsServiceImpl apiVersionService) {
+    public FilterHandler(FilterAndInvoker filterAndInvoker, long timeoutMs, String sniHostname, VirtualCluster virtualCluster, Channel inboundChannel) {
         this.filter = Objects.requireNonNull(filterAndInvoker).filter();
         this.invoker = filterAndInvoker.invoker();
         this.timeoutMs = Assertions.requireStrictlyPositive(timeoutMs, "timeout");
         this.sniHostname = sniHostname;
         this.virtualCluster = virtualCluster;
         this.inboundChannel = inboundChannel;
-        this.apiVersionService = apiVersionService;
     }
 
     String filterDescriptor() {
@@ -168,7 +163,7 @@ public class FilterHandler extends ChannelDuplexHandler {
     }
 
     private CompletableFuture<Void> readDecodedResponse(DecodedResponseFrame<?> decodedFrame) {
-        var filterContext = new InternalFilterContext(decodedFrame, apiVersionService);
+        var filterContext = new InternalFilterContext(decodedFrame);
 
         final var future = dispatchDecodedResponseFrame(decodedFrame, filterContext);
         boolean defer = !future.isDone();
@@ -201,7 +196,7 @@ public class FilterHandler extends ChannelDuplexHandler {
     }
 
     private CompletableFuture<Void> writeDecodedRequest(DecodedRequestFrame<?> decodedFrame, ChannelPromise promise) {
-        var filterContext = new InternalFilterContext(decodedFrame, apiVersionService);
+        var filterContext = new InternalFilterContext(decodedFrame);
         final var future = dispatchDecodedRequest(decodedFrame, filterContext);
         boolean defer = !future.isDone();
         if (defer) {
@@ -420,14 +415,12 @@ public class FilterHandler extends ChannelDuplexHandler {
         });
     }
 
-    private class InternalFilterContext implements FilterContext, ApiVersionsService {
+    private class InternalFilterContext implements FilterContext {
 
         private final DecodedFrame<?, ?> decodedFrame;
-        private final ApiVersionsServiceImpl apiVersionService;
 
-        InternalFilterContext(DecodedFrame<?, ?> decodedFrame, ApiVersionsServiceImpl apiVersionService) {
+        InternalFilterContext(DecodedFrame<?, ?> decodedFrame) {
             this.decodedFrame = decodedFrame;
-            this.apiVersionService = apiVersionService;
         }
 
         @Override
@@ -452,11 +445,6 @@ public class FilterHandler extends ChannelDuplexHandler {
             return virtualCluster.getClusterName();
         }
 
-        @Override
-        public ApiVersionsService getApiVersionsService() {
-            return this;
-        }
-
         @Override
         public RequestFilterResultBuilder requestFilterResultBuilder() {
             return new RequestFilterResultBuilderImpl();
@@ -524,11 +512,6 @@ public class FilterHandler extends ChannelDuplexHandler {
             return filterPromise.minimalCompletionStage();
         }
 
-        @Override
-        public CompletionStage<Optional<ApiVersionRanges>> getApiVersionRanges(ApiKeys keys) {
-            return apiVersionService.getApiVersionRanges(keys, this);
-        }
-
     }
 
 }
diff --git a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
index e37aebdb..f29a697d 100644
--- a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
+++ b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
@@ -71,7 +71,6 @@ public class KafkaProxyFrontendHandler
 
     private final boolean logNetwork;
     private final boolean logFrames;
-    private final ApiVersionsServiceImpl apiVersionService;
     private final VirtualCluster virtualCluster;
 
     private ChannelHandlerContext outboundCtx;
@@ -136,14 +135,12 @@ public class KafkaProxyFrontendHandler
 
     KafkaProxyFrontendHandler(NetFilter filter,
                               SaslDecodePredicate dp,
-                              VirtualCluster virtualCluster,
-                              ApiVersionsServiceImpl apiVersionService) {
+                              VirtualCluster virtualCluster) {
         this.filter = filter;
         this.dp = dp;
         this.virtualCluster = virtualCluster;
         this.logNetwork = virtualCluster.isLogNetwork();
         this.logFrames = virtualCluster.isLogFrames();
-        this.apiVersionService = apiVersionService;
     }
 
     private IllegalStateException illegalState(String msg) {
@@ -345,7 +342,7 @@ public class KafkaProxyFrontendHandler
     private void addFiltersToPipeline(List<FilterAndInvoker> filters, ChannelPipeline pipeline, Channel inboundChannel) {
         for (var filter : filters) {
             // TODO configurable timeout
-            pipeline.addFirst(filter.toString(), new FilterHandler(filter, 20000, sniHostname, virtualCluster, inboundChannel, apiVersionService));
+            pipeline.addFirst(filter.toString(), new FilterHandler(filter, 20000, sniHostname, virtualCluster, inboundChannel));
         }
     }
 
diff --git a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
index 2ceca331..477b7797 100644
--- a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
+++ b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
@@ -188,7 +188,7 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
 
         ApiVersionsServiceImpl apiVersionService = new ApiVersionsServiceImpl();
         final NetFilter netFilter = new InitalizerNetFilter(dp, apiVersionService, ch, binding, pfr, filterChainFactory, endpointReconciler);
-        var frontendHandler = new KafkaProxyFrontendHandler(netFilter, dp, virtualCluster, apiVersionService);
+        var frontendHandler = new KafkaProxyFrontendHandler(netFilter, dp, virtualCluster);
 
         pipeline.addLast("netHandler", frontendHandler);
 


Test patch:
diff --git a/kroxylicious-integration-tests/pom.xml b/kroxylicious-integration-tests/pom.xml
index ea139852..8638d720 100644
--- a/kroxylicious-integration-tests/pom.xml
+++ b/kroxylicious-integration-tests/pom.xml
@@ -151,17 +151,6 @@
             <artifactId>testcontainers</artifactId>
             <scope>test</scope>
         </dependency>
-        <!-- the next two are used by the Netty Leak Detection test -->
-        <dependency>
-            <groupId>org.apache.logging.log4j</groupId>
-            <artifactId>log4j-api</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.logging.log4j</groupId>
-            <artifactId>log4j-core</artifactId>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>org.awaitility</groupId>
             <artifactId>awaitility</artifactId>
diff --git a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/FilterIT.java b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/FilterIT.java
index 151c0e4c..4efa1710 100644
--- a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/FilterIT.java
+++ b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/FilterIT.java
@@ -36,13 +36,8 @@ import org.apache.kafka.common.message.ListTransactionsResponseData;
 import org.apache.kafka.common.message.MetadataRequestData;
 import org.apache.kafka.common.message.MetadataResponseData;
 import org.apache.kafka.common.message.ProduceRequestData;
-import org.apache.kafka.common.protocol.ApiMessage;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.serialization.Serdes;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.core.LoggerContext;
-import org.apache.logging.log4j.core.config.Configuration;
-import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -53,7 +48,6 @@ import org.junit.jupiter.params.provider.MethodSource;
 import io.github.nettyplus.leakdetector.junit.NettyLeakDetectorExtension;
 
 import io.kroxylicious.proxy.config.FilterDefinitionBuilder;
-import io.kroxylicious.proxy.filter.ApiVersionsMarkingFilterFactory;
 import io.kroxylicious.proxy.filter.ForwardingStyle;
 import io.kroxylicious.proxy.filter.RejectingCreateTopicFilter;
 import io.kroxylicious.proxy.filter.RejectingCreateTopicFilterFactory;
@@ -64,14 +58,11 @@ import io.kroxylicious.proxy.filter.simpletransform.ProduceRequestTransformation
 import io.kroxylicious.test.Request;
 import io.kroxylicious.test.Response;
 import io.kroxylicious.test.ResponsePayload;
-import io.kroxylicious.test.tester.MockServerKroxyliciousTester;
 import io.kroxylicious.testing.kafka.api.KafkaCluster;
 import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
 import io.kroxylicious.testing.kafka.junit5ext.Topic;
 
 import static io.kroxylicious.UnknownTaggedFields.unknownTaggedFieldsToStrings;
-import static io.kroxylicious.proxy.filter.ApiVersionsMarkingFilter.INTERSECTED_API_VERSION_RANGE_TAG;
-import static io.kroxylicious.proxy.filter.ApiVersionsMarkingFilter.UPSTREAM_API_VERSION_RANGE_TAG;
 import static io.kroxylicious.proxy.filter.RequestResponseMarkingFilter.FILTER_NAME_TAG;
 import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.proxy;
 import static io.kroxylicious.test.tester.KroxyliciousTesters.kroxyliciousTester;
@@ -101,13 +92,6 @@ class FilterIT {
     private static final String PLAINTEXT = "Hello, world!";
     private static final FilterDefinitionBuilder REJECTING_CREATE_TOPIC_FILTER = new FilterDefinitionBuilder(RejectingCreateTopicFilterFactory.class.getName());
 
-    @BeforeAll
-    public static void beforeAll() {
-
-        final LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
-        final Configuration config = ctx.getConfiguration();
-    }
-
     @Test
     void reversibleEncryption() {
         // The precise details of the cipher don't matter
@@ -360,25 +344,6 @@ class FilterIT {
                 .withMessage(RejectingCreateTopicFilter.ERROR_MESSAGE);
     }
 
-    @Test
-    void testApiVersionsAvailableToFilter() {
-        try (MockServerKroxyliciousTester tester = mockKafkaKroxyliciousTester((mockBootstrap) -> proxy(mockBootstrap)
-                .addToFilters(new FilterDefinitionBuilder(ApiVersionsMarkingFilterFactory.class.getName()).build()));
-                var kafkaClient = tester.simpleTestClient()) {
-            ApiVersionsResponseData apiVersionsResponseData = new ApiVersionsResponseData();
-            short kroxyliciousLatestVersion = METADATA.latestVersion();
-            int upstreamLatestVersion = kroxyliciousLatestVersion + 1;
-            apiVersionsResponseData.apiKeys().add(new ApiVersionsResponseData.ApiVersion().setApiKey(METADATA.id).setMinVersion(METADATA.oldestVersion()).setMaxVersion(
-                    (short) upstreamLatestVersion));
-            tester.addMockResponseForApiKey(new ResponsePayload(API_VERSIONS, API_VERSIONS.latestVersion(), apiVersionsResponseData));
-            tester.addMockResponseForApiKey(new ResponsePayload(METADATA, METADATA.latestVersion(), new MetadataResponseData()));
-            kafkaClient.getSync(new Request(METADATA, METADATA.latestVersion(), "client", new MetadataRequestData()));
-            ApiMessage message = tester.getOnlyRequestForApiKey(METADATA).message();
-            assertThat(unknownTaggedFieldsToStrings(message, INTERSECTED_API_VERSION_RANGE_TAG)).containsExactly("0-" + kroxyliciousLatestVersion);
-            assertThat(unknownTaggedFieldsToStrings(message, UPSTREAM_API_VERSION_RANGE_TAG)).containsExactly("0-" + upstreamLatestVersion);
-        }
-    }
-
     @Test
     void shouldModifyProduceMessage(KafkaCluster cluster, Topic topic1, Topic topic2) throws Exception {
 
diff --git a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/ApiVersionsMarkingFilter.java b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/ApiVersionsMarkingFilter.java
deleted file mode 100644
index 122f8472..00000000
--- a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/ApiVersionsMarkingFilter.java
+++ /dev/null
@@ -1,44 +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.filter;
-
-import java.util.concurrent.CompletionStage;
-
-import org.apache.kafka.common.message.ApiVersionsResponseData;
-import org.apache.kafka.common.message.RequestHeaderData;
-import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.ApiMessage;
-import org.apache.kafka.common.protocol.types.RawTaggedField;
-
-import io.kroxylicious.proxy.ApiVersionsService;
-
-import static java.nio.charset.StandardCharsets.UTF_8;
-
-public class ApiVersionsMarkingFilter implements RequestFilter {
-
-    public static final int INTERSECTED_API_VERSION_RANGE_TAG = 89;
-    public static final int UPSTREAM_API_VERSION_RANGE_TAG = 90;
-
-    @Override
-    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage request, FilterContext context) {
-        return context.getApiVersionsService().getApiVersionRanges(apiKey).thenCompose(apiVersionRanges -> {
-            if (apiVersionRanges.isEmpty()) {
-                return context.requestFilterResultBuilder().withCloseConnection().completed();
-            }
-            ApiVersionsService.ApiVersionRanges versionRanges = apiVersionRanges.get();
-            ApiVersionsResponseData.ApiVersion intersected = versionRanges.intersected();
-            request.unknownTaggedFields()
-                    .add(new RawTaggedField(INTERSECTED_API_VERSION_RANGE_TAG, (intersected.minVersion() + "-" + intersected.maxVersion()).getBytes(
-                            UTF_8)));
-            ApiVersionsResponseData.ApiVersion upstream = versionRanges.upstream();
-            request.unknownTaggedFields().add(new RawTaggedField(UPSTREAM_API_VERSION_RANGE_TAG, (upstream.minVersion() + "-" + upstream.maxVersion()).getBytes(
-                    UTF_8)));
-            return context.forwardRequest(header, request);
-        });
-    }
-
-}
diff --git a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/ApiVersionsMarkingFilterFactory.java b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/ApiVersionsMarkingFilterFactory.java
deleted file mode 100644
index 31625c22..00000000
--- a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/ApiVersionsMarkingFilterFactory.java
+++ /dev/null
@@ -1,27 +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.filter;
-
-import io.kroxylicious.proxy.plugin.Plugin;
-
-import edu.umd.cs.findbugs.annotations.NonNull;
-
-@Plugin(configType = Void.class)
-public class ApiVersionsMarkingFilterFactory implements FilterFactory<Void, Void> {
-
-    @Override
-    public Void initialize(FilterFactoryContext context, Void config) {
-        return null;
-    }
-
-    @NonNull
-    @Override
-    public ApiVersionsMarkingFilter createFilter(FilterFactoryContext context, Void configuration) {
-        return new ApiVersionsMarkingFilter();
-    }
-
-}
diff --git a/kroxylicious-integration-tests/src/test/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterFactory b/kroxylicious-integration-tests/src/test/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterFactory
index a549ffa6..b36b767d 100644
--- a/kroxylicious-integration-tests/src/test/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterFactory
+++ b/kroxylicious-integration-tests/src/test/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterFactory
@@ -1,5 +1,4 @@
 io.kroxylicious.proxy.filter.FixedClientIdFilterFactory
-io.kroxylicious.proxy.filter.ApiVersionsMarkingFilterFactory
 io.kroxylicious.proxy.filter.RequestResponseMarkingFilterFactory
 io.kroxylicious.proxy.filter.OutOfBandSendFilterFactory
 io.kroxylicious.proxy.filter.RejectingCreateTopicFilterFactory
diff --git a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImplTest.java b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImplTest.java
index 7096e5cd..533340b2 100644
--- a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImplTest.java
+++ b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImplTest.java
@@ -6,78 +6,14 @@
 
 package io.kroxylicious.proxy.internal;
 
-import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
-
 import org.apache.kafka.common.message.ApiVersionsResponseData;
-import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.Errors;
 import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-
-import io.kroxylicious.proxy.ApiVersionsService;
-import io.kroxylicious.proxy.filter.FilterContext;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.argThat;
 
 class ApiVersionsServiceImplTest {
 
-    @Test
-    void testGetVersionRanges_UsesCachedResultFromIntersectionMutation() {
-        ApiVersionsServiceImpl apiVersionsService = new ApiVersionsServiceImpl();
-        ApiVersionsResponseData upstreamApiVersions = createApiVersionsWith(ApiKeys.METADATA.id, (short) (ApiKeys.METADATA.oldestVersion() + 1),
-                (short) (ApiKeys.METADATA.latestVersion() + 1));
-        apiVersionsService.updateVersions("channel", upstreamApiVersions);
-        ApiVersionsService.ApiVersionRanges range = apiVersionsService.getApiVersionRanges(ApiKeys.METADATA, Mockito.mock(FilterContext.class)).toCompletableFuture()
-                .getNow(Optional.empty()).orElse(null);
-        assertThat(range).isNotNull();
-        assertThat(range.upstream().minVersion()).isEqualTo((short) (ApiKeys.METADATA.oldestVersion() + 1));
-        assertThat(range.upstream().maxVersion()).isEqualTo((short) (ApiKeys.METADATA.latestVersion() + 1));
-        assertThat(range.intersected().minVersion()).isEqualTo((short) (ApiKeys.METADATA.oldestVersion() + 1));
-        assertThat(range.intersected().maxVersion()).isEqualTo(ApiKeys.METADATA.latestVersion());
-    }
-
-    @Test
-    void testGetVersionRanges_UsesFilterContextToObtainVersionsIfRequired() {
-        ApiVersionsServiceImpl apiVersionsService = new ApiVersionsServiceImpl();
-        ApiVersionsResponseData upstreamApiVersions = createApiVersionsWith(ApiKeys.METADATA.id, (short) (ApiKeys.METADATA.oldestVersion() - 1),
-                (short) (ApiKeys.METADATA.latestVersion() - 1));
-        FilterContext filterContext = Mockito.mock(FilterContext.class);
-        Mockito.when(filterContext.sendRequest(any(RequestHeaderData.class), any()))
-                .thenReturn(CompletableFuture.completedFuture(upstreamApiVersions));
-        ApiVersionsService.ApiVersionRanges range = apiVersionsService.getApiVersionRanges(ApiKeys.METADATA, filterContext).toCompletableFuture()
-                .getNow(Optional.empty()).orElse(null);
-        assertThat(range).isNotNull();
-        assertThat(range.upstream().minVersion()).isEqualTo((short) (ApiKeys.METADATA.oldestVersion() - 1));
-        assertThat(range.upstream().maxVersion()).isEqualTo((short) (ApiKeys.METADATA.latestVersion() - 1));
-        assertThat(range.intersected().minVersion()).isEqualTo(ApiKeys.METADATA.oldestVersion());
-        assertThat(range.intersected().maxVersion()).isEqualTo((short) (ApiKeys.METADATA.latestVersion() - 1));
-    }
-
-    @Test
-    void testGetVersionRanges_SupportsFallbackToApiResponseV0() {
-        ApiVersionsServiceImpl apiVersionsService = new ApiVersionsServiceImpl();
-        FilterContext filterContext = Mockito.mock(FilterContext.class);
-        ApiVersionsResponseData unsupportedVersion = new ApiVersionsResponseData()
-                .setErrorCode(Errors.UNSUPPORTED_VERSION.code());
-        Mockito.when(filterContext.sendRequest(argThat(header -> header != null && header.requestApiVersion() > 0), any()))
-                .thenReturn(CompletableFuture.completedFuture(unsupportedVersion));
-        ApiVersionsResponseData upstreamApiVersions = createApiVersionsWith(ApiKeys.METADATA.id, (short) (ApiKeys.METADATA.oldestVersion() - 1),
-                (short) (ApiKeys.METADATA.latestVersion() - 1));
-        Mockito.when(filterContext.sendRequest(argThat(header -> header != null && header.requestApiVersion() == 0), any()))
-                .thenReturn(CompletableFuture.completedFuture(upstreamApiVersions));
-        ApiVersionsService.ApiVersionRanges range = apiVersionsService.getApiVersionRanges(ApiKeys.METADATA, filterContext).toCompletableFuture()
-                .getNow(Optional.empty()).orElse(null);
-        assertThat(range).isNotNull();
-        assertThat(range.upstream().minVersion()).isEqualTo((short) (ApiKeys.METADATA.oldestVersion() - 1));
-        assertThat(range.upstream().maxVersion()).isEqualTo((short) (ApiKeys.METADATA.latestVersion() - 1));
-        assertThat(range.intersected().minVersion()).isEqualTo(ApiKeys.METADATA.oldestVersion());
-        assertThat(range.intersected().maxVersion()).isEqualTo((short) (ApiKeys.METADATA.latestVersion() - 1));
-    }
-
     @Test
     void testIntersection_UpstreamMatchesApiKeys() {
         ApiVersionsServiceImpl apiVersionsService = new ApiVersionsServiceImpl();
diff --git a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
index 02f0de57..f825c322 100644
--- a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
+++ b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
@@ -81,7 +81,6 @@ public abstract class FilterHarness {
         var testVirtualCluster = new VirtualCluster("TestVirtualCluster", targetCluster, mock(ClusterNetworkAddressConfigProvider.class), Optional.empty(),
                 false, false);
         var inboundChannel = new EmbeddedChannel();
-        var apiVersionService = new ApiVersionsServiceImpl();
         var channelProcessors = Stream.<ChannelHandler> of(new InternalRequestTracker(), new CorrelationIdIssuer());
 
         var filterHandlers = Arrays.stream(filters)
@@ -90,7 +89,7 @@ public abstract class FilterHarness {
                     return d2;
                 })) // reverses order
                 .stream()
-                .map(f -> new FilterHandler(getOnlyElement(FilterAndInvoker.build(f)), timeoutMs, null, testVirtualCluster, inboundChannel, apiVersionService))
+                .map(f -> new FilterHandler(getOnlyElement(FilterAndInvoker.build(f)), timeoutMs, null, testVirtualCluster, inboundChannel))
                 .map(ChannelHandler.class::cast);
         var handlers = Stream.concat(channelProcessors, filterHandlers);
 
diff --git a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java
index 00b4898a..771d2206 100644
--- a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java
+++ b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java
@@ -225,7 +225,7 @@ class KafkaProxyFrontendHandlerTest {
     }
 
     KafkaProxyFrontendHandler handler(NetFilter filter, SaslDecodePredicate dp, VirtualCluster virtualCluster) {
-        return new KafkaProxyFrontendHandler(filter, dp, virtualCluster, new ApiVersionsServiceImpl()) {
+        return new KafkaProxyFrontendHandler(filter, dp, virtualCluster) {
             @Override
             ChannelFuture initConnection(String remoteHost, int remotePort, Bootstrap b) {
                 // This is ugly... basically the EmbeddedChannel doesn't seem to handle the case