Repository: kroxylicious/kroxylicious (java)

Issue:
Make Filters aware of the upstream supported api versions
**Is your feature request related to a problem? Please describe.**
When I use the `sendRequest` method of `KrpcFilterContext` to send a message to the upstream Kafka broker, I must supply an apiVersion. Currently we do not know in the Filter if the Broker supports that apiVersion. We may be able to infer it from the apiVersion of other messages, or intercept the ApiVersions response in our Filter and refer to it later. This feels risky.

For example in the topic-encryption filter we infer that because the FetchRequest is supplying topic ids, the broker should support [MetadataRequest api 12 that can carry topic ids](https://github.com/kroxylicious/topic-encryption/blob/main/kroxylicious-filter/src/main/java/io/strimzi/kafka/topicenc/kroxylicious/FetchDecryptFilter.java#L32C1-L32C1).

**Describe the solution you'd like**
It would be nice if we could instead access the supported upstream ApiVersions through the KrpcFilterContext to check what upstream supports before we send any messages.


Pull request description:
Add ApiVersionsService to KrpcFilterContext
### Type of change

- Enhancement / new feature

### Description

Adds interface `ApiVersionsService` to `kroxylicious-api`:

```
import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion;
import org.apache.kafka.common.protocol.ApiKeys;

public interface ApiVersionsService {
    record ApiVersionRanges(ApiVersion upstream, ApiVersion intersected) {}

    CompletionStage<ApiVersionRanges> getApiVersionRanges(ApiKeys keys);
}
```
and adds `ApiVersionsService getApiVersionsService();` to `FilterContext`

This service:
1. uses the ApiVersionsFilter to record the upstream versions
2. is also capable of hijacking the connection if the user needs the upstream versions in cases where the ApiVersionsFilter has not yet intercepted them.

Why:
Filters can choose to hijack the connection to send their own extra requests to the upstream broker. To do this safely we need to know what versions of the RPC the upstream supports, so we don't send it unsupported requests.

Filters are free to implement ApiVersionsResponseFilter and intercept the response, or alternatively send an additional request to the upstream broker asking for the ApiVersions (some clients may not send an ApiVersionsRequest, therefore there is no response to intercept). We can implement this pattern for Filter Authors so it does not need to be implemented in each Filter that wants to know the upstream API versions.

The API provides the intersected versions supported by both upstream and this Kroxylicious instance and also the raw upstream versions in case they are of interest.

Closes #438

### 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 Sonar-Lint warnings are addressed or are justifiably ignored.
- [ ] Update documentation
- [ ] Reference relevant issue(s) and close them after merging
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


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
new file mode 100644
index 00000000..272c4bf1
--- /dev/null
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/ApiVersionsService.java
@@ -0,0 +1,37 @@
+/*
+ * 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;
+
+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
+     */
+    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 b96d1e90..918113b5 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,6 +14,8 @@ 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;
 
 /**
@@ -124,4 +126,11 @@ 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/src/main/java/io/kroxylicious/proxy/internal/filter/ApiVersionsFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImpl.java
similarity index 52%
rename from kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ApiVersionsFilter.java
rename to kroxylicious/src/main/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImpl.java
index ff70c7d0..8eba56b3 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ApiVersionsFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImpl.java
@@ -3,32 +3,43 @@
  *
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
-package io.kroxylicious.proxy.internal.filter;
+
+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.ResponseHeaderData;
+import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion;
+import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersionCollection;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.kroxylicious.proxy.filter.ApiVersionsResponseFilter;
+import io.kroxylicious.proxy.ApiVersionsService;
 import io.kroxylicious.proxy.filter.FilterContext;
-import io.kroxylicious.proxy.filter.ResponseFilterResult;
 
-/**
- * Changes an API_VERSIONS response so that a client sees the intersection of supported version ranges for each
- * API key. This is an intrinsic part of correctly acting as a proxy.
- */
-public class ApiVersionsFilter implements ApiVersionsResponseFilter {
+public class ApiVersionsServiceImpl {
+
+    private record ApiVersions(ApiVersionCollection upstream, ApiVersionCollection intersected) {
+    }
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(ApiVersionsFilter.class);
+    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);
+    }
 
     private static void intersectApiVersions(String channel, ApiVersionsResponseData resp) {
-        Set<ApiVersionsResponseData.ApiVersion> unknownApis = new HashSet<>();
+        Set<ApiVersion> unknownApis = new HashSet<>();
         for (var key : resp.apiKeys()) {
             short apiId = key.apiKey();
             if (ApiKeys.hasId(apiId)) {
@@ -73,10 +84,30 @@ public class ApiVersionsFilter implements ApiVersionsResponseFilter {
         }
     }
 
-    @Override
-    public CompletionStage<ResponseFilterResult> onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData data,
-                                                                       FilterContext context) {
-        intersectApiVersions(context.channelDescriptor(), data);
-        return context.forwardResponse(header, data);
+    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);
+        }
+        else {
+            return context.sendRequest(ApiKeys.API_VERSIONS.latestVersion(), new ApiVersionsRequestData()).thenApply(message -> {
+                ApiVersionsResponseData apiVersionsResponseData = (ApiVersionsResponseData) message;
+                updateVersions(context.channelDescriptor(), apiVersionsResponseData);
+                return apiVersions;
+            });
+        }
+    }
+
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
index fd136b7c..95319bae 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
@@ -6,6 +6,7 @@
 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,6 +30,7 @@ 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;
@@ -64,17 +66,20 @@ 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;
 
-    public FilterHandler(FilterAndInvoker filterAndInvoker, long timeoutMs, String sniHostname, VirtualCluster virtualCluster, Channel inboundChannel) {
+    public FilterHandler(FilterAndInvoker filterAndInvoker, long timeoutMs, String sniHostname, VirtualCluster virtualCluster, Channel inboundChannel,
+                         ApiVersionsServiceImpl apiVersionService) {
         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() {
@@ -161,7 +166,7 @@ public class FilterHandler extends ChannelDuplexHandler {
     }
 
     private CompletableFuture<Void> readDecodedResponse(DecodedResponseFrame<?> decodedFrame) {
-        var filterContext = new InternalFilterContext(decodedFrame);
+        var filterContext = new InternalFilterContext(decodedFrame, apiVersionService);
 
         final var future = dispatchDecodedResponseFrame(decodedFrame, filterContext);
         boolean defer = !future.isDone();
@@ -201,7 +206,7 @@ public class FilterHandler extends ChannelDuplexHandler {
     }
 
     private CompletableFuture<Void> writeDecodedRequest(DecodedRequestFrame<?> decodedFrame, ChannelPromise promise) {
-        var filterContext = new InternalFilterContext(decodedFrame);
+        var filterContext = new InternalFilterContext(decodedFrame, apiVersionService);
         final var future = dispatchDecodedRequest(decodedFrame, filterContext);
         boolean defer = !future.isDone();
         if (defer) {
@@ -427,12 +432,14 @@ public class FilterHandler extends ChannelDuplexHandler {
         });
     }
 
-    private class InternalFilterContext implements FilterContext {
+    private class InternalFilterContext implements FilterContext, ApiVersionsService {
 
         private final DecodedFrame<?, ?> decodedFrame;
+        private final ApiVersionsServiceImpl apiVersionService;
 
-        InternalFilterContext(DecodedFrame<?, ?> decodedFrame) {
+        InternalFilterContext(DecodedFrame<?, ?> decodedFrame, ApiVersionsServiceImpl apiVersionService) {
             this.decodedFrame = decodedFrame;
+            this.apiVersionService = apiVersionService;
         }
 
         @Override
@@ -457,6 +464,11 @@ public class FilterHandler extends ChannelDuplexHandler {
             return virtualCluster.getClusterName();
         }
 
+        @Override
+        public ApiVersionsService getApiVersionsService() {
+            return this;
+        }
+
         @Override
         public RequestFilterResultBuilder requestFilterResultBuilder() {
             return new RequestFilterResultBuilderImpl();
@@ -525,5 +537,11 @@ public class FilterHandler extends ChannelDuplexHandler {
             }, timeoutMs, TimeUnit.MILLISECONDS);
             return filterStage;
         }
+
+        @Override
+        public CompletionStage<Optional<ApiVersionRanges>> getApiVersionRanges(ApiKeys keys) {
+            return apiVersionService.getApiVersionRanges(keys, this);
+        }
+
     }
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
index 16b8e192..b81a55ca 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java
@@ -69,6 +69,7 @@ public class KafkaProxyFrontendHandler
 
     private final boolean logNetwork;
     private final boolean logFrames;
+    private final ApiVersionsServiceImpl apiVersionService;
     private final VirtualCluster virtualCluster;
 
     private ChannelHandlerContext outboundCtx;
@@ -133,12 +134,14 @@ public class KafkaProxyFrontendHandler
 
     KafkaProxyFrontendHandler(NetFilter filter,
                               SaslDecodePredicate dp,
-                              VirtualCluster virtualCluster) {
+                              VirtualCluster virtualCluster,
+                              ApiVersionsServiceImpl apiVersionService) {
         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) {
@@ -336,7 +339,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));
+            pipeline.addFirst(filter.toString(), new FilterHandler(filter, 20000, sniHostname, virtualCluster, inboundChannel, apiVersionService));
         }
     }
 
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 1558205e..f90ed9bb 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
@@ -31,7 +31,7 @@ import io.kroxylicious.proxy.config.Configuration;
 import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.internal.codec.KafkaRequestDecoder;
 import io.kroxylicious.proxy.internal.codec.KafkaResponseEncoder;
-import io.kroxylicious.proxy.internal.filter.ApiVersionsFilter;
+import io.kroxylicious.proxy.internal.filter.ApiVersionsIntersectFilter;
 import io.kroxylicious.proxy.internal.filter.BrokerAddressFilter;
 import io.kroxylicious.proxy.internal.filter.EagerMetadataLearner;
 import io.kroxylicious.proxy.internal.net.Endpoint;
@@ -170,9 +170,11 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
             pipeline.addLast(new KafkaAuthnHandler(ch, authnHandlers));
         }
 
+        ApiVersionsServiceImpl apiVersionService = new ApiVersionsServiceImpl();
         var frontendHandler = new KafkaProxyFrontendHandler(context -> {
             var filterChainFactory = new FilterChainFactory(config);
-            List<FilterAndInvoker> apiVersionFilters = dp.isAuthenticationOffloadEnabled() ? List.of() : FilterAndInvoker.build(new ApiVersionsFilter());
+            List<FilterAndInvoker> apiVersionFilters = dp.isAuthenticationOffloadEnabled() ? List.of()
+                    : FilterAndInvoker.build(new ApiVersionsIntersectFilter(apiVersionService));
             List<FilterAndInvoker> customProtocolFilters = filterChainFactory.createFilters();
             List<FilterAndInvoker> brokerAddressFilters = FilterAndInvoker.build(new BrokerAddressFilter(virtualCluster, endpointReconciler));
             var filters = new ArrayList<>(apiVersionFilters);
@@ -189,7 +191,7 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
             }
 
             context.initiateConnect(target, filters);
-        }, dp, virtualCluster);
+        }, dp, virtualCluster, apiVersionService);
 
         pipeline.addLast("netHandler", frontendHandler);
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ApiVersionsIntersectFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ApiVersionsIntersectFilter.java
new file mode 100644
index 00000000..29037d8f
--- /dev/null
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ApiVersionsIntersectFilter.java
@@ -0,0 +1,35 @@
+/*
+ * 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.filter;
+
+import java.util.concurrent.CompletionStage;
+
+import org.apache.kafka.common.message.ApiVersionsResponseData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+
+import io.kroxylicious.proxy.filter.ApiVersionsResponseFilter;
+import io.kroxylicious.proxy.filter.FilterContext;
+import io.kroxylicious.proxy.filter.ResponseFilterResult;
+import io.kroxylicious.proxy.internal.ApiVersionsServiceImpl;
+
+/**
+ * Changes an API_VERSIONS response so that a client sees the intersection of supported version ranges for each
+ * API key. This is an intrinsic part of correctly acting as a proxy.
+ */
+public class ApiVersionsIntersectFilter implements ApiVersionsResponseFilter {
+    private final ApiVersionsServiceImpl apiVersionsService;
+
+    public ApiVersionsIntersectFilter(ApiVersionsServiceImpl service) {
+        this.apiVersionsService = service;
+    }
+
+    @Override
+    public CompletionStage<ResponseFilterResult> onApiVersionsResponse(short apiVersion, ResponseHeaderData header, ApiVersionsResponseData data,
+                                                                       FilterContext context) {
+        apiVersionsService.updateVersions(context.channelDescriptor(), data);
+        return context.forwardResponse(header, data);
+    }
+}


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/FilterIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/FilterIT.java
index 5f77aeb4..fd860042 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/FilterIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/FilterIT.java
@@ -12,7 +12,6 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Stream;
@@ -33,6 +32,7 @@ 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;
@@ -61,6 +61,8 @@ import io.kroxylicious.testing.kafka.api.KafkaCluster;
 import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
 
 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;
@@ -73,6 +75,7 @@ import static org.apache.kafka.clients.producer.ProducerConfig.CLIENT_ID_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG;
 import static org.apache.kafka.common.protocol.ApiKeys.API_VERSIONS;
 import static org.apache.kafka.common.protocol.ApiKeys.CREATE_TOPICS;
+import static org.apache.kafka.common.protocol.ApiKeys.FETCH;
 import static org.apache.kafka.common.protocol.ApiKeys.LIST_GROUPS;
 import static org.apache.kafka.common.protocol.ApiKeys.LIST_TRANSACTIONS;
 import static org.apache.kafka.common.protocol.ApiKeys.METADATA;
@@ -268,6 +271,10 @@ class FilterIT {
         try (var tester = mockKafkaKroxyliciousTester((mockBootstrap) -> proxy(mockBootstrap).addToFilters(markingFilter));
                 var kafkaClient = tester.simpleTestClient()) {
             tester.addMockResponseForApiKey(new ResponsePayload(LIST_TRANSACTIONS, LIST_TRANSACTIONS.latestVersion(), new ListTransactionsResponseData()));
+            ApiVersionsResponseData apiVersionsResponseData = new ApiVersionsResponseData();
+            apiVersionsResponseData.apiKeys()
+                    .add(new ApiVersionsResponseData.ApiVersion().setApiKey(FETCH.id).setMaxVersion(FETCH.latestVersion()).setMinVersion(FETCH.oldestVersion()));
+            tester.addMockResponseForApiKey(new ResponsePayload(API_VERSIONS, API_VERSIONS.latestVersion(), apiVersionsResponseData));
 
             // In the ASYNCHRONOUS_REQUEST_TO_BROKER case, the filter will send an async list_group
             // request to the broker and defer the forward of the list transaction response until the list groups
@@ -333,6 +340,25 @@ class FilterIT {
         }
     }
 
+    @Test
+    void testApiVersionsAvailableToFilter() {
+        try (MockServerKroxyliciousTester tester = mockKafkaKroxyliciousTester((mockBootstrap) -> proxy(mockBootstrap)
+                .addToFilters(new FilterDefinitionBuilder("ApiVersionsMarkingFilter").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, Admin admin) throws Exception {
         admin.createTopics(List.of(
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/ApiVersionsMarkingFilter.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/ApiVersionsMarkingFilter.java
new file mode 100644
index 00000000..a699e878
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/ApiVersionsMarkingFilter.java
@@ -0,0 +1,42 @@
+/*
+ * 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/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
index 7fdc8707..b7947ba4 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
@@ -58,7 +58,6 @@ public class RequestResponseMarkingFilter implements RequestFilter, ResponseFilt
         if (!(direction.contains(Direction.REQUEST) && keysToMark.contains(apiKey))) {
             return context.forwardRequest(header, body);
         }
-
         return forwardingStyle.apply(context, body)
                 .thenApply(request -> applyTaggedField(request, Direction.REQUEST, name))
                 .thenCompose(taggedRequest -> context.forwardRequest(header, taggedRequest));
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java
index 820884be..8990e343 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java
@@ -12,6 +12,7 @@ public class TestFilterContributor extends BaseContributor<Filter> implements Fi
 
     public static final BaseContributorBuilder<Filter> FILTERS = BaseContributor.<Filter> builder()
             .add("FixedClientId", FixedClientIdFilter.FixedClientIdFilterConfig.class, FixedClientIdFilter::new)
+            .add("ApiVersionsMarkingFilter", ApiVersionsMarkingFilter::new)
             .add("RequestResponseMarking", RequestResponseMarkingFilter.RequestResponseMarkingFilterConfig.class, RequestResponseMarkingFilter::new)
             .add("OutOfBandSend", OutOfBandSendFilter.OutOfBandSendFilterConfig.class, OutOfBandSendFilter::new)
             .add("CompositePrefixingFixedClientId", CompositePrefixingFixedClientIdFilterConfig.class, CompositePrefixingFixedClientIdFilter::new)
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImplTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImplTest.java
new file mode 100644
index 00000000..2276c611
--- /dev/null
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/ApiVersionsServiceImplTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.kafka.common.message.ApiVersionsResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+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.anyShort;
+
+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(anyShort(), 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();
+        ApiVersionsResponseData upstreamApiVersions = createApiVersionsWith(ApiKeys.METADATA.id, ApiKeys.METADATA.oldestVersion(),
+                ApiKeys.METADATA.latestVersion());
+        apiVersionsService.updateVersions("channel", upstreamApiVersions);
+        assertThatApiVersionsContainsExactly(upstreamApiVersions, ApiKeys.METADATA, ApiKeys.METADATA.oldestVersion(), ApiKeys.METADATA.latestVersion());
+    }
+
+    @Test
+    void testIntersection_UpstreamMinVersionLessThanApiKeys() {
+        ApiVersionsServiceImpl apiVersionsService = new ApiVersionsServiceImpl();
+        ApiVersionsResponseData upstreamApiVersions = createApiVersionsWith(ApiKeys.METADATA.id, (short) (ApiKeys.METADATA.oldestVersion() - 1),
+                ApiKeys.METADATA.latestVersion());
+        apiVersionsService.updateVersions("channel", upstreamApiVersions);
+        assertThatApiVersionsContainsExactly(upstreamApiVersions, ApiKeys.METADATA, ApiKeys.METADATA.oldestVersion(), ApiKeys.METADATA.latestVersion());
+    }
+
+    @Test
+    void testIntersection_UpstreamMinVersionGreaterThanApiKeys() {
+        ApiVersionsServiceImpl apiVersionsService = new ApiVersionsServiceImpl();
+        ApiVersionsResponseData upstreamApiVersions = createApiVersionsWith(ApiKeys.METADATA.id, (short) (ApiKeys.METADATA.oldestVersion() + 1),
+                ApiKeys.METADATA.latestVersion());
+        apiVersionsService.updateVersions("channel", upstreamApiVersions);
+        assertThatApiVersionsContainsExactly(upstreamApiVersions, ApiKeys.METADATA, (short) (ApiKeys.METADATA.oldestVersion() + 1), ApiKeys.METADATA.latestVersion());
+    }
+
+    @Test
+    void testIntersection_UpstreamMaxVersionLessThanApiKeys() {
+        ApiVersionsServiceImpl apiVersionsService = new ApiVersionsServiceImpl();
+        ApiVersionsResponseData upstreamApiVersions = createApiVersionsWith(ApiKeys.METADATA.id, ApiKeys.METADATA.oldestVersion(),
+                (short) (ApiKeys.METADATA.latestVersion() - 1));
+        apiVersionsService.updateVersions("channel", upstreamApiVersions);
+        assertThatApiVersionsContainsExactly(upstreamApiVersions, ApiKeys.METADATA, ApiKeys.METADATA.oldestVersion(), (short) (ApiKeys.METADATA.latestVersion() - 1));
+    }
+
+    @Test
+    void testIntersection_UpstreamMaxVersionGreaterThanApiKeys() {
+        ApiVersionsServiceImpl apiVersionsService = new ApiVersionsServiceImpl();
+        ApiVersionsResponseData upstreamApiVersions = createApiVersionsWith(ApiKeys.METADATA.id, ApiKeys.METADATA.oldestVersion(),
+                (short) (ApiKeys.METADATA.latestVersion() + 1));
+        apiVersionsService.updateVersions("channel", upstreamApiVersions);
+        assertThatApiVersionsContainsExactly(upstreamApiVersions, ApiKeys.METADATA, ApiKeys.METADATA.oldestVersion(), ApiKeys.METADATA.latestVersion());
+    }
+
+    @Test
+    void testIntersection_DiscardsUnknownUpstreamApiKeys() {
+        ApiVersionsServiceImpl apiVersionsService = new ApiVersionsServiceImpl();
+        ApiVersionsResponseData upstreamApiVersions = createApiVersionsWith((short) 678, ApiKeys.METADATA.oldestVersion(),
+                (short) (ApiKeys.METADATA.latestVersion() + 1));
+        apiVersionsService.updateVersions("channel", upstreamApiVersions);
+        assertThat(upstreamApiVersions.apiKeys()).isEmpty();
+    }
+
+    private static void assertThatApiVersionsContainsExactly(ApiVersionsResponseData upstreamApiVersions, ApiKeys apiKeys, short minVersion, short maxVersion) {
+        assertThat(upstreamApiVersions.apiKeys()).satisfies(apiVersions -> {
+            assertThat(apiVersions).hasSize(1).first().satisfies(apiVersion -> {
+                assertThat(apiVersion.apiKey()).isEqualTo(apiKeys.id);
+                assertThat(apiVersion.minVersion()).isEqualTo(minVersion);
+                assertThat(apiVersion.maxVersion()).isEqualTo(maxVersion);
+            });
+        });
+    }
+
+    private static ApiVersionsResponseData createApiVersionsWith(short api, short minVersion, short maxVersion) {
+        ApiVersionsResponseData upstreamApiVersions = new ApiVersionsResponseData();
+        upstreamApiVersions.apiKeys().add(new ApiVersionsResponseData.ApiVersion().setApiKey(api).setMinVersion(minVersion).setMaxVersion(
+                maxVersion));
+        return upstreamApiVersions;
+    }
+
+}
\ No newline at end of file
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
index 481e7baa..201c012b 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
@@ -60,7 +60,7 @@ public abstract class FilterHarness {
         this.filter = filter;
         filterHandler = new FilterHandler(getOnlyElement(FilterAndInvoker.build(filter)), timeoutMs, null,
                 new VirtualCluster("TestVirtualCluster", mock(TargetCluster.class), mock(ClusterNetworkAddressConfigProvider.class), Optional.empty(), false, false),
-                new EmbeddedChannel());
+                new EmbeddedChannel(), new ApiVersionsServiceImpl());
         channel = new EmbeddedChannel(filterHandler);
     }
 
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java
index 4fea691d..e88c33db 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java
@@ -182,7 +182,7 @@ class KafkaProxyFrontendHandlerTest {
     }
 
     KafkaProxyFrontendHandler handler(NetFilter filter, SaslDecodePredicate dp, VirtualCluster virtualCluster) {
-        return new KafkaProxyFrontendHandler(filter, dp, virtualCluster) {
+        return new KafkaProxyFrontendHandler(filter, dp, virtualCluster, new ApiVersionsServiceImpl()) {
             @Override
             ChannelFuture initConnection(String remoteHost, int remotePort, Bootstrap b) {
                 // This is ugly... basically the EmbeddedChannel doesn't seem to handle the case
diff --git a/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
index 74f8e820..c252ba78 100644
--- a/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
+++ b/performance-tests/src/main/java/io/kroxylicious/benchmarks/InvokerDispatchBenchmark.java
@@ -33,6 +33,7 @@ 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;
@@ -178,6 +179,11 @@ public class InvokerDispatchBenchmark {
             return null;
         }
 
+        @Override
+        public ApiVersionsService getApiVersionsService() {
+            return null;
+        }
+
         @Override
         public <T extends ApiMessage> CompletionStage<T> sendRequest(short apiVersion, ApiMessage request) {
             return null;