Repository: kroxylicious/kroxylicious (java)

Issue:
Out-of-band requests have upstream Request filters applied, but not upstream Response filters.
> I confirmed it [experimentally](https://github.com/kroxylicious/kroxylicious/compare/main...robobario:kproxy:intercept-out-of-band). Will make an issue to find the right solution separately.

_Originally posted by @robobario in https://github.com/kroxylicious/kroxylicious/pull/437#discussion_r1259145995_
 
Currently when you send an out-of-band request to the Cluster from a Filter, upstream filters are applied to the request. But when handling the response other user filters are not invoked. The FilterHandler[ forwards it on ](https://github.com/kroxylicious/kroxylicious/blob/4be5632203f131e26c25a3db38f5dd3a54aa5bbc/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java#L92)until it gets to the FilterHandler for the recipient/sender Filter, at which point it completes the future.

Pull request description:
Enable Response filters to intercept out-of-band responses
Closes #445

### Type of change

- Bugfix

### Description

Enable Response filters to intercept out-of-band responses. Previously the response to an out of band request was forwarded without invoking any of the upstream filters between the Cluster and the filter that initiated the `sendRequest`.

### Additional Context

There was an inconsistency in that out-of-band requests could be intercepted by upstream filters, but the Cluster's response was not intercepted. The javadoc for sendRequest said `The response will pass through upstream filters prior to the handler being invoked`, which can be read as saying they should be invoked.

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [x] Write tests
- [ ] Make sure all tests pass
- [ ] Make sure all Sonar-Lint warnings are addressed or are justifiably ignored.
- [ ] Update documentation
- [x] 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/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
index 7764e7d3..5bd6c12a 100644
--- a/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
+++ b/api/kroxylicious-filter-api/src/main/java/io/kroxylicious/proxy/filter/KrpcFilterContext.java
@@ -48,7 +48,7 @@ public interface KrpcFilterContext {
     /**
      * Send a message from a filter towards the broker, invoking upstream filters
      * and being informed of the response via TODO.
-     * The response will pass through upstream filters prior to the handler being invoked.
+     * The response will pass through upstream filters, invoking them, prior to the handler being invoked.
      * Response propagation will stop once the handler has completed,
      * i.e. the downstream filters will not receive the response.
      *
diff --git a/docs/custom-filters.adoc b/docs/custom-filters.adoc
index 05e0a4a8..226fd93f 100644
--- a/docs/custom-filters.adoc
+++ b/docs/custom-filters.adoc
@@ -366,7 +366,7 @@ currently do not have a mechanism for your Filter to know the supported API vers
 you use a hardcoded API version as shown, your Filter will only work with upstream brokers that support that api version
 for that RPC (see issue {github}/issues/438[#438]).
 
-The out-of-band Request/Response are eligible to be intercepted by Filters upstream of this Filter.
+The out-of-band Request/Response will be intercepted by Filters upstream of this Filter.
 
 Note you should not attempt to call `forward` methods on the `KrpcFilterContext` using the CompletionStage, they still
 must be called synchronously during the invocation of the `onXyz[Request|Response]` method.
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 8c246fec..b881a4b5 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
@@ -75,22 +75,13 @@ public class FilterHandler
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
         if (msg instanceof DecodedResponseFrame<?> decodedFrame) {
-            if (decodedFrame instanceof InternalResponseFrame<?> frame) {
-                if (frame.isRecipient(filter)) {
-                    if (LOGGER.isDebugEnabled()) {
-                        LOGGER.debug("{}: Completing {} response for request sent by this filter{}: {}",
-                                ctx.channel(), decodedFrame.apiKey(), filterDescriptor(), msg);
-                    }
-                    CompletableFuture<ApiMessage> p = frame.promise();
-                    p.complete(decodedFrame.body());
-                }
-                else {
-                    if (LOGGER.isDebugEnabled()) {
-                        LOGGER.debug("{}: Not completing {} response for request sent by another filter {}",
-                                ctx.channel(), decodedFrame.apiKey(), frame.recipient());
-                    }
-                    ctx.fireChannelRead(msg);
+            if (decodedFrame instanceof InternalResponseFrame<?> frame && frame.isRecipient(filter)) {
+                if (LOGGER.isDebugEnabled()) {
+                    LOGGER.debug("{}: Completing {} response for request sent by this filter{}: {}",
+                            ctx.channel(), decodedFrame.apiKey(), filterDescriptor(), msg);
                 }
+                CompletableFuture<ApiMessage> p = frame.promise();
+                p.complete(decodedFrame.body());
             }
             else {
                 var filterContext = new DefaultFilterContext(filter, ctx, decodedFrame, null, timeoutMs, sniHostname);


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/UnknownTaggedFields.java b/integrationtests/src/test/java/io/kroxylicious/UnknownTaggedFields.java
new file mode 100644
index 00000000..dbe25637
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/UnknownTaggedFields.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious;
+
+import java.util.stream.Stream;
+
+import org.apache.kafka.common.protocol.Message;
+import org.apache.kafka.common.protocol.types.RawTaggedField;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+public class UnknownTaggedFields {
+
+    public static Stream<String> unknownTaggedFieldsToStrings(Message message, int tag) {
+        return message.unknownTaggedFields().stream()
+                .filter(t -> t.tag() == tag)
+                .map(RawTaggedField::data)
+                .map(b -> new String(b, UTF_8));
+    }
+}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyStandaloneIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyStandaloneIT.java
index ce829174..03b89e1c 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyStandaloneIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/KroxyStandaloneIT.java
@@ -12,6 +12,7 @@ import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
 import org.apache.kafka.clients.admin.Admin;
@@ -84,7 +85,10 @@ public class KroxyStandaloneIT {
             String classpath = System.getProperty("java.class.path");
             var processBuilder = new ProcessBuilder(java, "-cp", classpath, "io.kroxylicious.proxy.Kroxylicious", "-c", configPath.toString()).inheritIO();
             Process start = processBuilder.start();
-            return start::destroy;
+            return () -> {
+                start.destroy();
+                start.onExit().get(10, TimeUnit.SECONDS);
+            };
         }
         catch (Exception e) {
             throw new RuntimeException(e);
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/OutOfBandRequestIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/OutOfBandRequestIT.java
new file mode 100644
index 00000000..f38d5f54
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/OutOfBandRequestIT.java
@@ -0,0 +1,102 @@
+/*
+ * 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.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.common.message.CreateTopicsResponseData;
+import org.apache.kafka.common.message.DescribeClusterRequestData;
+import org.apache.kafka.common.message.DescribeClusterResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.junit.jupiter.api.Test;
+
+import io.kroxylicious.proxy.config.FilterDefinition;
+import io.kroxylicious.proxy.config.FilterDefinitionBuilder;
+import io.kroxylicious.test.Request;
+import io.kroxylicious.test.Response;
+import io.kroxylicious.test.client.KafkaClient;
+import io.kroxylicious.test.tester.KroxyliciousConfigUtils;
+import io.kroxylicious.test.tester.KroxyliciousTesters;
+import io.kroxylicious.test.tester.MockServerKroxyliciousTester;
+
+import static io.kroxylicious.UnknownTaggedFields.unknownTaggedFieldsToStrings;
+import static io.kroxylicious.proxy.filter.RequestResponseMarkingFilter.FILTER_NAME_TAG;
+import static org.apache.kafka.common.protocol.ApiKeys.CREATE_TOPICS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Given filter chain A -> B -> C. If B sends an out-of-band request X, then:
+ * <ol>
+ *     <li>C can Filter and mutate X before it is sent to the Cluster</li>
+ *     <li>C can Filter and mutate the Cluster's response to X before the CompletionStage is completed</li>
+ *     <li>The CompletionStage will be completed when the Cluster's response to X flows to B. This is the end of the flow.</li>
+ *     <li>A can not filter X or the response to X, the flow was terminated.</li>
+ * </ol>
+ */
+public class OutOfBandRequestIT {
+
+    @Test
+    public void testOutOfBandMessageInterceptedByUpstreamFilters() {
+        // this filter should not intercept the out-of-band request or response, it will not either message with its name
+        FilterDefinition downstreamFilter = addAddUnknownTaggedFieldToMessagesWithApiKey("downstreamOfOutOfBandFilter", CREATE_TOPICS);
+        FilterDefinition outOfBandSender = outOfBandSender(CREATE_TOPICS, FILTER_NAME_TAG);
+        // this filter should intercept the out-of-band request and response, it will tag both messages with its name
+        FilterDefinition upstreamFilter = addAddUnknownTaggedFieldToMessagesWithApiKey("upstreamOfOutOfBandFilter", CREATE_TOPICS);
+        try (var tester = createMockTesterWithFilters(downstreamFilter, outOfBandSender,
+                upstreamFilter);
+                var client = tester.singleRequestClient()) {
+            givenMockReturnsArbitraryCreateTopicResponse(tester);
+            DescribeClusterResponseData responseData = whenDescribeCluster(client);
+            thenResponseContainsTagsAugmentedInByUpstreamFilterOnly(responseData);
+            andMessageFromOutOfBandRequestToMockHadTagAddedByUpstreamFilterOnly(tester);
+        }
+    }
+
+    private static MockServerKroxyliciousTester createMockTesterWithFilters(FilterDefinition... definitions) {
+        List<FilterDefinition> filterDefs = Arrays.stream(definitions).toList();
+        return KroxyliciousTesters.mockKafkaKroxyliciousTester(s -> KroxyliciousConfigUtils.proxy(s).addAllToFilters(filterDefs));
+    }
+
+    private static FilterDefinition outOfBandSender(ApiKeys apiKeyToSend, int tagToCollect) {
+        return new FilterDefinitionBuilder("OutOfBandSend").withConfig(Map.of("apiKeyToSend", apiKeyToSend, "tagToCollect", tagToCollect)).build();
+    }
+
+    private static FilterDefinition addAddUnknownTaggedFieldToMessagesWithApiKey(String name, ApiKeys apiKeys) {
+        return new FilterDefinitionBuilder("RequestResponseMarking").withConfig("name", name, "keysToMark", Set.of(apiKeys)).build();
+    }
+
+    private static void andMessageFromOutOfBandRequestToMockHadTagAddedByUpstreamFilterOnly(MockServerKroxyliciousTester tester) {
+        Request request = tester.onlyRequest();
+        String tags = unknownTaggedFieldsToStrings(request.message(), FILTER_NAME_TAG)
+                .collect(Collectors.joining(","));
+        assertEquals("RequestResponseMarkingFilter-upstreamOfOutOfBandFilter-request", tags);
+    }
+
+    private static void thenResponseContainsTagsAugmentedInByUpstreamFilterOnly(DescribeClusterResponseData responseData) {
+        assertEquals("filterNameTaggedFieldsFromOutOfBandResponse: RequestResponseMarkingFilter-upstreamOfOutOfBandFilter-response", responseData.errorMessage());
+    }
+
+    private static DescribeClusterResponseData whenDescribeCluster(KafkaClient client) {
+        Response response = client.getSync(
+                new Request(ApiKeys.DESCRIBE_CLUSTER, ApiKeys.DESCRIBE_CLUSTER.latestVersion(), "client", new DescribeClusterRequestData()));
+        return (DescribeClusterResponseData) response.message();
+    }
+
+    private static void givenMockReturnsArbitraryCreateTopicResponse(MockServerKroxyliciousTester tester) {
+        CreateTopicsResponseData message = new CreateTopicsResponseData();
+        CreateTopicsResponseData.CreatableTopicResult topic = new CreateTopicsResponseData.CreatableTopicResult();
+        topic.setName("mockTopic");
+        topic.setReplicationFactor((short) 3);
+        topic.setNumPartitions(3);
+        message.topics().add(topic);
+        tester.setMockResponse(new Response(CREATE_TOPICS, CREATE_TOPICS.latestVersion(), message));
+    }
+}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/Utils.java b/integrationtests/src/test/java/io/kroxylicious/proxy/Utils.java
deleted file mode 100644
index 5dda63f0..00000000
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/Utils.java
+++ /dev/null
@@ -1,20 +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 io.kroxylicious.proxy.config.ConfigParser;
-import io.kroxylicious.proxy.config.Configuration;
-
-public class Utils {
-    public static KafkaProxy startProxy(String config) throws InterruptedException {
-        Configuration proxyConfig = new ConfigParser().parseConfiguration(config);
-
-        KafkaProxy kafkaProxy = new KafkaProxy(proxyConfig);
-        kafkaProxy.startup();
-
-        return kafkaProxy;
-    }
-}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/OutOfBandSendFilter.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/OutOfBandSendFilter.java
new file mode 100644
index 00000000..11ed6e01
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/OutOfBandSendFilter.java
@@ -0,0 +1,80 @@
+/*
+ * 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.Objects;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData;
+import org.apache.kafka.common.message.DescribeClusterRequestData;
+import org.apache.kafka.common.message.DescribeClusterResponseData;
+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.Errors;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import io.kroxylicious.proxy.config.BaseConfig;
+
+import static io.kroxylicious.UnknownTaggedFields.unknownTaggedFieldsToStrings;
+
+/**
+ * Designed to work in tandem with another Filter like {@link RequestResponseMarkingFilter}. This Filter targets
+ * DescribeClusterRequest and sends an out-of-band request instead. Then, when it gets a response via the sendRequest
+ * CompletionStage, it collects the Unknown Tagged Fields for a specified tag and forwards them as an error message in
+ * a DescribeClusterResponse. This exposes what unknown tagged fields were added to the out-of-band response by other
+ * filters like {@link RequestResponseMarkingFilter}.
+ */
+public class OutOfBandSendFilter implements DescribeClusterRequestFilter {
+
+    private final OutOfBandSendFilterConfig config;
+
+    public OutOfBandSendFilter(OutOfBandSendFilterConfig config) {
+        this.config = config;
+    }
+
+    public static class OutOfBandSendFilterConfig extends BaseConfig {
+        private final ApiKeys apiKeyToSend;
+        private final int tagIdToCollect;
+
+        @JsonCreator
+        public OutOfBandSendFilterConfig(@JsonProperty(value = "apiKeyToSend", required = true) ApiKeys apiKeyToSend,
+                                         @JsonProperty(value = "tagToCollect", required = true) int tagIdToCollect) {
+            this.apiKeyToSend = apiKeyToSend;
+            this.tagIdToCollect = tagIdToCollect;
+        }
+
+    }
+
+    @Override
+    public void onDescribeClusterRequest(short apiVersion, RequestHeaderData header, DescribeClusterRequestData request, KrpcFilterContext context) {
+        ApiKeys apiKeyToSend = config.apiKeyToSend;
+        ApiMessage message = createApiMessage(apiKeyToSend);
+        context.sendRequest(apiKeyToSend.latestVersion(), message).thenAccept(apiMessage -> {
+            String values = unknownTaggedFieldsToStrings(apiMessage, config.tagIdToCollect).collect(Collectors.joining(","));
+            // WARNING: forwarding from within the completion is not generally supported yet, this only works because the test is strictly controlling
+            // requests being sent to Kroxylicious. If used in a proxy between a real Kafka client and broker, this could cause out-of-order responses to
+            // the client, causing client side errors. We use it here so that we can see the outcome of the sendRequest call with a single request to Kroxylicious.
+            context.forwardResponse(
+                    new DescribeClusterResponseData().setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code())
+                            .setErrorMessage("filterNameTaggedFieldsFromOutOfBandResponse: " + values));
+        });
+    }
+
+    private static ApiMessage createApiMessage(ApiKeys apiKeyToSend) {
+        ApiMessage message;
+        if (Objects.requireNonNull(apiKeyToSend) == ApiKeys.CREATE_TOPICS) {
+            message = new CreateTopicsRequestData();
+        }
+        else {
+            throw new IllegalArgumentException("apiKey " + apiKeyToSend + " not supported yet");
+        }
+        return message;
+    }
+}
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
new file mode 100644
index 00000000..bffd8947
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RequestResponseMarkingFilter.java
@@ -0,0 +1,76 @@
+/*
+ * 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.Set;
+
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.types.RawTaggedField;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import io.kroxylicious.proxy.config.BaseConfig;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+/**
+ * A Filter that adds an UnknownTaggedField to all request/response messages for some
+ * specified set of ApiKeys. The tags have values:
+ * <ul>
+ *     <li>Requests: RequestResponseMarkingFilter-${config.name}-request</li>
+ *     <li>Responses: RequestResponseMarkingFilter-${config.name}-response</li>
+ * </ul>
+ */
+public class RequestResponseMarkingFilter implements RequestFilter, ResponseFilter {
+
+    public static final int FILTER_NAME_TAG = 500;
+    private final String name;
+    private final Set<ApiKeys> keysToMark;
+
+    public RequestResponseMarkingFilter(RequestResponseMarkingFilterConfig config) {
+        name = config.name;
+        keysToMark = config.keysToMark;
+    }
+
+    @Override
+    public void onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        if (keysToMark.contains(apiKey)) {
+            body.unknownTaggedFields().add(createTaggedField("request"));
+        }
+        filterContext.forwardRequest(header, body);
+    }
+
+    @Override
+    public void onResponse(ApiKeys apiKey, ResponseHeaderData header, ApiMessage body, KrpcFilterContext filterContext) {
+        if (keysToMark.contains(apiKey)) {
+            body.unknownTaggedFields().add(createTaggedField("response"));
+        }
+        filterContext.forwardResponse(header, body);
+    }
+
+    private RawTaggedField createTaggedField(String type) {
+        return new RawTaggedField(FILTER_NAME_TAG, (this.getClass().getSimpleName() + "-" + name + "-" + type).getBytes(UTF_8));
+    }
+
+    public static class RequestResponseMarkingFilterConfig extends BaseConfig {
+        private final String name;
+
+        private final Set<ApiKeys> keysToMark;
+
+        @JsonCreator
+        public RequestResponseMarkingFilterConfig(@JsonProperty(value = "name", required = true) String name,
+                                                  @JsonProperty(value = "keysToMark", required = true) Set<ApiKeys> keysToMark) {
+            this.name = name;
+            this.keysToMark = keysToMark;
+        }
+    }
+
+}
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 e82eee70..b0770a1c 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,8 @@ public class TestFilterContributor extends BaseContributor<KrpcFilter> implement
 
     public static final BaseContributorBuilder<KrpcFilter> FILTERS = BaseContributor.<KrpcFilter> builder()
             .add("FixedClientId", FixedClientIdFilter.FixedClientIdFilterConfig.class, FixedClientIdFilter::new)
+            .add("RequestResponseMarking", RequestResponseMarkingFilter.RequestResponseMarkingFilterConfig.class, RequestResponseMarkingFilter::new)
+            .add("OutOfBandSend", OutOfBandSendFilter.OutOfBandSendFilterConfig.class, OutOfBandSendFilter::new)
             .add("CompositePrefixingFixedClientId", CompositePrefixingFixedClientIdFilterConfig.class, CompositePrefixingFixedClientIdFilter::new)
             .add("CreateTopicRejectFilter", CreateTopicRejectFilter::new);
 
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
index 15e28278..f6ccb617 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
@@ -8,6 +8,7 @@ package io.kroxylicious.proxy.internal;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
 
 import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.ApiVersionsResponseData;
@@ -17,13 +18,17 @@ import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.types.RawTaggedField;
 import org.junit.jupiter.api.Test;
 
 import io.kroxylicious.proxy.filter.ApiVersionsRequestFilter;
 import io.kroxylicious.proxy.filter.ApiVersionsResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
+import io.kroxylicious.proxy.frame.DecodedRequestFrame;
+import io.kroxylicious.proxy.frame.DecodedResponseFrame;
 import io.kroxylicious.proxy.future.InternalCompletionStage;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
@@ -33,6 +38,8 @@ import static org.junit.jupiter.api.Assertions.fail;
 
 public class FilterHandlerTest extends FilterHarness {
 
+    public static final int ARBITRARY_TAG = 500;
+
     @Test
     public void testForwardRequest() {
         ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> context.forwardRequest(header, request);
@@ -79,6 +86,34 @@ public class FilterHandlerTest extends FilterHarness {
         assertEquals(frame, propagated, "Expect it to be the frame that was sent");
     }
 
+    @Test
+    public void testOtherFiltersInChainCanFilterOutOfBandResponse() {
+        ApiVersionsResponseFilter recipientFilter = taggingApiVersionsResponseFilter("recipient");
+        String filterName = "other-interested-filter";
+        ApiVersionsResponseFilter filterUnderTest = taggingApiVersionsResponseFilter(filterName);
+        buildChannel(filterUnderTest);
+        CompletableFuture<Object> future = new CompletableFuture<>();
+        var frame = writeInternalResponse(new ApiVersionsResponseData(), future, recipientFilter);
+        var propagated = channel.readInbound();
+        assertEquals(frame, propagated, "Expect it to be the frame that was sent");
+        assertResponseMessageTaggedWith(filterName, (InternalResponseFrame<?>) propagated);
+        assertFalse(future.isDone());
+    }
+
+    @Test
+    public void testOtherFiltersInChainCanFilterOutOfBandRequest() {
+        ApiVersionsRequestFilter recipientFilter = taggingApiVersionsRequestFilter("recipient");
+        String filterName = "other-interested-filter";
+        ApiVersionsRequestFilter filterUnderTest = taggingApiVersionsRequestFilter(filterName);
+        buildChannel(filterUnderTest);
+        CompletableFuture<Object> future = new CompletableFuture<>();
+        var frame = writeInternalRequest(new ApiVersionsRequestData(), future, recipientFilter);
+        var propagated = channel.readOutbound();
+        assertEquals(frame, propagated, "Expect it to be the frame that was sent");
+        assertRequestMessageTaggedWith(filterName, (DecodedRequestFrame<?>) propagated);
+        assertFalse(future.isDone());
+    }
+
     @Test
     public void testShouldNotDeserializeResponse() {
         ApiVersionsResponseFilter filter = new ApiVersionsResponseFilter() {
@@ -243,4 +278,33 @@ public class FilterHandlerTest extends FilterHarness {
         assertThrows(ExecutionException.class, q::get);
     }
 
+    private static void assertResponseMessageTaggedWith(String filterName, DecodedResponseFrame<?> propagated) {
+        String tag = collectTagsToStrings(propagated.body(), ARBITRARY_TAG);
+        assertEquals(tag, filterName);
+    }
+
+    private static void assertRequestMessageTaggedWith(String filterName, DecodedRequestFrame<?> propagated) {
+        String tag = collectTagsToStrings(propagated.body(), ARBITRARY_TAG);
+        assertEquals(tag, filterName);
+    }
+
+    private static ApiVersionsResponseFilter taggingApiVersionsResponseFilter(String tag) {
+        return (apiVersion, header, response, context) -> {
+            response.unknownTaggedFields().add(new RawTaggedField(ARBITRARY_TAG, tag.getBytes(UTF_8)));
+            context.forwardResponse(header, response);
+        };
+    }
+
+    private static String collectTagsToStrings(ApiMessage body, int tag) {
+        return body.unknownTaggedFields().stream().filter(f -> f.tag() == tag)
+                .map(RawTaggedField::data).map(f -> new String(f, UTF_8)).collect(Collectors.joining(","));
+    }
+
+    private static ApiVersionsRequestFilter taggingApiVersionsRequestFilter(String tag) {
+        return (apiVersion, header, request, context) -> {
+            request.unknownTaggedFields().add(new RawTaggedField(ARBITRARY_TAG, tag.getBytes(UTF_8)));
+            context.forwardRequest(header, request);
+        };
+    }
+
 }
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 40bac954..f117e2ac 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java
@@ -96,15 +96,45 @@ public abstract class FilterHarness {
      * @param <B> The type of the response body.
      */
     protected <B extends ApiMessage> DecodedResponseFrame<B> writeInternalResponse(B data, CompletableFuture<?> future) {
+        return writeInternalResponse(data, future, filter);
+    }
+
+    /**
+     * Write a response for a filter-originated request, as if from the broker.
+     * @param data The body of the response.
+     * @param future
+     * @param recipient the Filter that wants to receive the internal response
+     * @return The frame that was written.
+     * @param <B> The type of the response body.
+     */
+    protected <B extends ApiMessage> DecodedResponseFrame<B> writeInternalResponse(B data, CompletableFuture<?> future, KrpcFilter recipient) {
         var apiKey = ApiKeys.forId(data.apiKey());
         var header = new ResponseHeaderData();
         int correlationId = 42;
         header.setCorrelationId(correlationId);
-        var frame = new InternalResponseFrame<>(filter, apiKey.latestVersion(), correlationId, header, data, future);
+        var frame = new InternalResponseFrame<>(recipient, apiKey.latestVersion(), correlationId, header, data, future);
         channel.writeInbound(frame);
         return frame;
     }
 
+    /**
+     * Write a response for a filter-originated request, as if from the broker.
+     * @param data The body of the response.
+     * @param future
+     * @param recipient the Filter that wants to receive the internal response
+     * @return The frame that was written.
+     * @param <B> The type of the response body.
+     */
+    protected <B extends ApiMessage> DecodedRequestFrame<B> writeInternalRequest(B data, CompletableFuture<?> future, KrpcFilter recipient) {
+        var apiKey = ApiKeys.forId(data.apiKey());
+        var header = new RequestHeaderData();
+        int correlationId = 42;
+        header.setCorrelationId(correlationId);
+        var frame = new InternalRequestFrame<>(apiKey.latestVersion(), correlationId, true, recipient, future, header, data);
+        channel.writeOutbound(frame);
+        return frame;
+    }
+
     /**
      * Shutdown the channel, asserting there were no further requests or responses to read.
      */