Repository: kroxylicious/kroxylicious
Pull request #633 (2023-09-27 13:44:10)

FetchResponseTransformationFilter improvements
### Type of change

- Bugfix

### Description

* Fix exception handling path in the case the out-of-band request failed
* Improve future handling by making use of chaining
* Add tests

### Additional Context

I need to change FetchResponseTransformationFilter as part of #621, the absence of tests was causing a quality gate
issue.  Addressing in a separate PR is best practice.

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [x] Write tests
- [x] 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
- [x] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=633)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=CODE_SMELL) [0 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=CODE_SMELL)

[![100.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/100-16px.png '100.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_coverage&view=list) [100.0% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_duplicated_lines_density&view=list)


sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=633)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=CODE_SMELL) [0 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=CODE_SMELL)

[![100.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/100-16px.png '100.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_coverage&view=list) [100.0% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_duplicated_lines_density&view=list)


sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=633)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=CODE_SMELL) [0 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=CODE_SMELL)

[![100.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/100-16px.png '100.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_coverage&view=list) [100.0% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_duplicated_lines_density&view=list)


sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=633)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=CODE_SMELL) [0 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=633&resolved=false&types=CODE_SMELL)

[![100.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/100-16px.png '100.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_coverage&view=list) [100.0% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=633&metric=new_duplicated_lines_density&view=list)



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 98fef49e..785a1bdc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,7 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 
 ## 0.3.0
 
+* [#633](https://github.com/kroxylicious/kroxylicious/pull/633): Address missing exception handling in FetchResponseTransformationFilter (and add unit tests)
 * [#537](https://github.com/kroxylicious/kroxylicious/issues/537): Computation stages chained to the CompletionStage return by #sendRequest using the default executor async methods now run on the Netty Event Loop.
 * [#612](https://github.com/kroxylicious/kroxylicious/pull/612): [Breaking] Allow filter authors to declare when their filter requires configuration. Note this includes a backwards incompatible change to the contract of the `Contributor`. `getInstance` will now throw exceptions rather than returning `null` to mean there was a problem or this contributor does not know about the requested type.
 * [#608](https://github.com/kroxylicious/kroxylicious/pull/608): Improve the contributor API to allow it to express more properties about the configuration. This release deprecates `Contributor.getConfigType` in favour of `Contributor.getConfigDefinition`. It also removes the proliferation of ContributorManager classes by providing a single type which can handle all Contributors.
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java
index 5a7c081e..60df943e 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilter.java
@@ -9,7 +9,6 @@ import java.lang.reflect.InvocationTargetException;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.stream.Collectors;
 
@@ -37,7 +36,7 @@ import io.kroxylicious.proxy.filter.ResponseFilterResult;
 import io.kroxylicious.proxy.internal.util.MemoryRecordsHelper;
 
 /**
- * An filter for modifying the key/value/header/topic of {@link ApiKeys#FETCH} responses.
+ * A filter for modifying the key/value/header/topic of {@link ApiKeys#FETCH} responses.
  */
 public class FetchResponseTransformationFilter implements FetchResponseFilter {
 
@@ -86,14 +85,13 @@ public class FetchResponseTransformationFilter implements FetchResponseFilter {
                 .collect(Collectors.toList());
         if (!requestTopics.isEmpty()) {
             LOGGER.debug("Fetch response contains {} unknown topic ids, lookup via Metadata request: {}", requestTopics.size(), requestTopics);
-            var future = new CompletableFuture<ResponseFilterResult>();
-            // TODO Can't necessarily use HIGHEST_SUPPORTED_VERSION, must use highest supported version
-            context.<MetadataResponseData> sendRequest(MetadataRequestData.HIGHEST_SUPPORTED_VERSION,
-                    new MetadataRequestData()
-                            .setTopics(requestTopics))
-                    .thenAccept(metadataResponse -> {
-                        Map<Uuid, String> uidToName = metadataResponse.topics().stream().collect(Collectors.toMap(MetadataResponseData.MetadataResponseTopic::topicId,
-                                MetadataResponseData.MetadataResponseTopic::name));
+            // Version 12 required for topic id support.
+            var metadataRequest = new MetadataRequestData().setTopics(requestTopics);
+            return context.<MetadataResponseData> sendRequest(MetadataRequestData.HIGHEST_SUPPORTED_VERSION, metadataRequest)
+                    .thenCompose(metadataResponse -> {
+                        Map<Uuid, String> uidToName = metadataResponse.topics().stream()
+                                .collect(Collectors.toMap(MetadataResponseData.MetadataResponseTopic::topicId,
+                                        MetadataResponseData.MetadataResponseTopic::name));
                         LOGGER.debug("Metadata response yields {}, updating original Fetch response", uidToName);
                         for (var fetchableTopicResponse : fetchResponse.responses()) {
                             fetchableTopicResponse.setTopic(uidToName.get(fetchableTopicResponse.topicId()));
@@ -101,10 +99,8 @@ public class FetchResponseTransformationFilter implements FetchResponseFilter {
                         applyTransformation(context, fetchResponse);
                         LOGGER.debug("Forwarding original Fetch response");
 
-                        var value = context.responseFilterResultBuilder().forward(header, fetchResponse).build();
-                        future.complete(value);
+                        return context.responseFilterResultBuilder().forward(header, fetchResponse).completed();
                     });
-            return future;
         }
         else {
             applyTransformation(context, fetchResponse);
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ProduceRequestTransformationFilter.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ProduceRequestTransformationFilter.java
index 57deee10..fdbb17c8 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ProduceRequestTransformationFilter.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/ProduceRequestTransformationFilter.java
@@ -31,7 +31,7 @@ import io.kroxylicious.proxy.filter.RequestFilterResult;
 import io.kroxylicious.proxy.internal.util.MemoryRecordsHelper;
 
 /**
- * An filter for modifying the key/value/header/topic of {@link ApiKeys#PRODUCE} requests.
+ * A filter for modifying the key/value/header/topic of {@link ApiKeys#PRODUCE} requests.
  */
 public class ProduceRequestTransformationFilter implements ProduceRequestFilter {
 


Test patch:
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilterTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilterTest.java
new file mode 100644
index 00000000..0e5e56a9
--- /dev/null
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/FetchResponseTransformationFilterTest.java
@@ -0,0 +1,218 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Collection;
+import java.util.Locale;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.FetchResponseData;
+import org.apache.kafka.common.message.FetchResponseData.FetchableTopicResponse;
+import org.apache.kafka.common.message.FetchResponseData.PartitionData;
+import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.MemoryRecordsBuilder;
+import org.apache.kafka.common.record.Record;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.record.Records;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.stubbing.Answer;
+
+import io.kroxylicious.proxy.filter.FilterContext;
+import io.kroxylicious.proxy.filter.ResponseFilterResult;
+import io.kroxylicious.proxy.filter.ResponseFilterResultBuilder;
+import io.kroxylicious.proxy.filter.filterresultbuilder.CloseOrTerminalStage;
+import io.kroxylicious.proxy.internal.filter.FetchResponseTransformationFilter.FetchResponseTransformationConfig;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyShort;
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class FetchResponseTransformationFilterTest {
+
+    private static final String TOPIC_NAME = "mytopic";
+    private static final Uuid TOPIC_ID = Uuid.randomUuid();
+    private static final String ORIGINAL_RECORD_VALUE = "lowercasevalue";
+    private static final String EXPECTED_TRANSFORMED_RECORD_VALUE = ORIGINAL_RECORD_VALUE.toUpperCase(Locale.ROOT);
+    private static final String RECORD_KEY = "key";
+    private FetchResponseTransformationFilter filter;
+    @Mock(strictness = Mock.Strictness.LENIENT)
+    FilterContext context;
+
+    @Mock(strictness = Mock.Strictness.LENIENT)
+    private ResponseFilterResult responseFilterResult;
+
+    @Mock(strictness = Mock.Strictness.LENIENT, extraInterfaces = CloseOrTerminalStage.class)
+    private ResponseFilterResultBuilder responseFilterResultBuilder;
+
+    @Captor
+    private ArgumentCaptor<Integer> bufferInitialCapacity;
+
+    @Captor
+    private ArgumentCaptor<ResponseHeaderData> responseHeaderDataCaptor;
+
+    @Captor
+    private ArgumentCaptor<ApiMessage> apiMessageCaptor;
+
+    @BeforeEach
+    @SuppressWarnings("unchecked")
+    void setUp() {
+        filter = new FetchResponseTransformationFilter(new FetchResponseTransformationConfig(ProduceRequestTransformationFilter.UpperCasing.class.getName()));
+
+        when(context.forwardResponse(responseHeaderDataCaptor.capture(), apiMessageCaptor.capture())).thenAnswer(
+                invocation -> CompletableFuture.completedStage(responseFilterResult));
+        when(context.responseFilterResultBuilder()).thenReturn(responseFilterResultBuilder);
+
+        when(responseFilterResult.message()).thenAnswer(invocation -> apiMessageCaptor.getValue());
+        when(responseFilterResult.header()).thenAnswer(invocation -> responseHeaderDataCaptor.getValue());
+
+        when(responseFilterResultBuilder.forward(responseHeaderDataCaptor.capture(), apiMessageCaptor.capture()))
+                .thenReturn((CloseOrTerminalStage<ResponseFilterResult>) responseFilterResultBuilder);
+        when(((CloseOrTerminalStage<ResponseFilterResult>) responseFilterResultBuilder).completed()).thenReturn(CompletableFuture.completedStage(responseFilterResult));
+
+        when(context.createByteBufferOutputStream(bufferInitialCapacity.capture())).thenAnswer(
+                (Answer<ByteBufferOutputStream>) invocation -> {
+                    Object[] args = invocation.getArguments();
+                    Integer size = (Integer) args[0];
+                    return new ByteBufferOutputStream(size);
+                });
+    }
+
+    @Test
+    void filterHandlesPreV13ResponseBasedOnTopicNames() throws Exception {
+
+        var fetchResponse = new FetchResponseData();
+        fetchResponse.responses().add(createFetchableTopicResponseWithOneRecord(RECORD_KEY, ORIGINAL_RECORD_VALUE).setTopic(TOPIC_NAME)); // Version 12
+
+        var stage = filter.onFetchResponse(fetchResponse.apiKey(), new ResponseHeaderData(), fetchResponse, context);
+        assertThat(stage).isCompleted();
+
+        var filteredResponse = (FetchResponseData) stage.toCompletableFuture().get().message();
+        var filteredRecords = responseToRecordStream(filteredResponse).toList();
+        assertThat(filteredRecords)
+                .withFailMessage("unexpected number of records in the filter response")
+                .hasSize(1);
+
+        var filteredRecord = filteredRecords.get(0);
+        assertThat(decodeUtf8Value(filteredRecord))
+                .withFailMessage("expected record value to have been transformed")
+                .isEqualTo(EXPECTED_TRANSFORMED_RECORD_VALUE);
+    }
+
+    @Test
+    void filterHandlesV13OrHigherResponseBasedOnTopicIds() throws Exception {
+
+        var fetchResponse = new FetchResponseData();
+        fetchResponse.responses().add(createFetchableTopicResponseWithOneRecord(RECORD_KEY, ORIGINAL_RECORD_VALUE).setTopicId(TOPIC_ID));
+
+        var metadataResponse = new MetadataResponseData();
+        metadataResponse.topics().add(new MetadataResponseData.MetadataResponseTopic().setTopicId(TOPIC_ID).setName(TOPIC_NAME));
+
+        when(context.sendRequest(anyShort(), isA(MetadataRequestData.class)))
+                .thenReturn(CompletableFuture.completedStage(metadataResponse));
+
+        var stage = filter.onFetchResponse(fetchResponse.apiKey(), new ResponseHeaderData(), fetchResponse, context);
+        assertThat(stage).isCompleted();
+
+        var filteredResponse = (FetchResponseData) stage.toCompletableFuture().get().message();
+
+        // verify that the response now has the topic name
+        assertThat(filteredResponse.responses())
+                .withFailMessage("expected same number of topics in the response")
+                .hasSameSizeAs(fetchResponse.responses())
+                .withFailMessage("expected topic response to have been augmented with topic name")
+                .anyMatch(ftr -> Objects.equals(ftr.topic(), TOPIC_NAME))
+                .withFailMessage("expected topic response to still have the topic id")
+                .anyMatch(ftr -> Objects.equals(ftr.topicId(), TOPIC_ID));
+
+        var filteredRecords = responseToRecordStream(filteredResponse).toList();
+        assertThat(filteredRecords)
+                .withFailMessage("unexpected number of records in the filter response")
+                .hasSize(1);
+
+        var filteredRecord = filteredRecords.get(0);
+        assertThat(decodeUtf8Value(filteredRecord))
+                .withFailMessage("expected record value to have been transformed")
+                .isEqualTo(EXPECTED_TRANSFORMED_RECORD_VALUE);
+    }
+
+    @Test
+    void filterHandlesMetadataRequestError() throws Exception {
+
+        var fetchResponse = new FetchResponseData();
+        // Version 13 switched to topic id rather than topic names.
+        fetchResponse.responses().add(createFetchableTopicResponseWithOneRecord(RECORD_KEY, ORIGINAL_RECORD_VALUE).setTopicId(TOPIC_ID));
+
+        var metadataResponse = new MetadataResponseData();
+        metadataResponse.topics().add(new MetadataResponseData.MetadataResponseTopic().setTopicId(TOPIC_ID).setName(TOPIC_NAME));
+
+        when(context.sendRequest(anyShort(), isA(MetadataRequestData.class)))
+                .thenReturn(CompletableFuture.failedStage(new IllegalStateException("out-of-band request exception")));
+
+        var stage = filter.onFetchResponse(fetchResponse.apiKey(), new ResponseHeaderData(), fetchResponse, context);
+        assertThat(stage)
+                .withFailMessage("out-of-band request exception")
+                .isCompletedExceptionally();
+    }
+
+    private Stream<Record> responseToRecordStream(FetchResponseData filteredResponse) {
+        return Stream.of(filteredResponse.responses())
+                .flatMap(Collection::stream)
+                .map(FetchableTopicResponse::partitions)
+                .flatMap(Collection::stream)
+                .map(PartitionData::records)
+                .map(Records.class::cast)
+                .map(Records::records)
+                .map(Iterable::spliterator)
+                .flatMap(si -> StreamSupport.stream(si, false));
+    }
+
+    @NotNull
+    private static FetchableTopicResponse createFetchableTopicResponseWithOneRecord(String key, String value) {
+        var fetchableTopicResponse = new FetchableTopicResponse();
+        var partitionData1 = new PartitionData();
+        partitionData1.setRecords(buildOneRecord(key, value));
+        fetchableTopicResponse.partitions().add(partitionData1);
+        return fetchableTopicResponse;
+    }
+
+    private static MemoryRecords buildOneRecord(String key, String value) {
+        ByteBuffer buffer = ByteBuffer.allocate(1024);
+        try (MemoryRecordsBuilder builder = MemoryRecords.builder(buffer, RecordBatch.CURRENT_MAGIC_VALUE,
+                CompressionType.NONE, TimestampType.CREATE_TIME, 0L, System.currentTimeMillis())) {
+            builder.append(0L, key.getBytes(), value.getBytes());
+            return builder.build();
+        }
+    }
+
+    @NotNull
+    private String decodeUtf8Value(Record record) {
+        return StandardCharsets.UTF_8.decode(record.value()).toString();
+    }
+
+}