Repository: kroxylicious/kroxylicious (java)

Issue:
Rewrite broker address information carried by ShareFetchResponse and ShareAckResponse  (KIP-932)
**Is your feature request related to a problem? Please describe.**

**Describe the solution you'd like**

[KIP-923](https://cwiki.apache.org/confluence/display/KAFKA/KIP-932%3A+Queues+for+Kafka#KIP932:QueuesforKafka-Fetchingandacknowledgingrecords) (Queues for Apache Kafka) is currently partially delivered in Kafka 3.9. It is marked as "unstable" so it is disabled by default.

KIP-932 introduces several new RPC.  Two of those RPCs `ShareFetchResponse` and `ShareAckResponse` carry node host/port information.  In order to allow users of the proxy (and Filter authors) the freedom to experiment with KIP-932, the Kroxylicious Core  needs to rewrite the broker addresses in the new RPCs.  If this is not done, the Kafka Client will fail.

`BrokerAddressFilter` needs to intercept the two new RPCs and rewrite the broker addresses contained within the response.
It should be able to to this using the established pattern.   We should add some supporting tests.


**Describe alternatives you've considered**

**Additional context**
Add any other context or screenshots about the feature request here.


Pull request description:
Fix #1612: KIP-932 Rewrite broker address information carried by ShareFetchResponse and ShareAckResponse [Kafka Unstable]
### Type of change

_Select the type of your PR_

- Enhancement / new feature

### Description

Kafka 3.9 introduces the new RPC that carry broker host/port information.  The Kafka feature is marked as unstable and only partially delivered by Kafka 3.9, but we said we want to implement it early to allow Filter Authors to experiment with the new Kafka features, if they wish.

### Additional Context

_Why are you making this pull request?_

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [x] Write tests
- [ ] Update documentation
- [x] Make sure all unit/integration tests pass
- [ ] Make sure all Sonarcloud warnings are addressed or are justifiably ignored.
- [ ] If applicable to the change, [trigger the system test suite](../blob/main/DEV_GUIDE.md#jenkins-pipeline-for-system-tests).  Make sure tests pass.
- [ ] If applicable to the change, [trigger the performance test suite](../blob/main/PERFORMANCE.md#jenkins-pipeline-for-performance). Ensure that any degradations to performance numbers are understood and justified.
- [ ] Ensure the PR references relevant issue(s) so they are closed on merging.
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).

> **_NOTE:_**  You must be a member of `@kroxylicious/developers` to trigger the system test and performance test suites.  If you are not part of this group, comment on the PR requesting a trigger, tagging `@kroxylicious/developers`.


Patch:
diff --git a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java
index 570cb535..038654aa 100644
--- a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java
+++ b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java
@@ -22,6 +22,8 @@ import org.apache.kafka.common.message.MetadataResponseData;
 import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseBroker;
 import org.apache.kafka.common.message.ProduceResponseData;
 import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.message.ShareAcknowledgeResponseData;
+import org.apache.kafka.common.message.ShareFetchResponseData;
 import org.apache.kafka.common.protocol.ApiMessage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -33,6 +35,8 @@ import io.kroxylicious.proxy.filter.FindCoordinatorResponseFilter;
 import io.kroxylicious.proxy.filter.MetadataResponseFilter;
 import io.kroxylicious.proxy.filter.ProduceResponseFilter;
 import io.kroxylicious.proxy.filter.ResponseFilterResult;
+import io.kroxylicious.proxy.filter.ShareAcknowledgeResponseFilter;
+import io.kroxylicious.proxy.filter.ShareFetchResponseFilter;
 import io.kroxylicious.proxy.internal.net.EndpointReconciler;
 import io.kroxylicious.proxy.model.VirtualCluster;
 import io.kroxylicious.proxy.service.HostPort;
@@ -42,7 +46,7 @@ import io.kroxylicious.proxy.service.HostPort;
  * is responsible for updating the virtual cluster's cache of upstream broker endpoints.
  */
 public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordinatorResponseFilter, DescribeClusterResponseFilter,
-        ProduceResponseFilter, FetchResponseFilter {
+        ProduceResponseFilter, FetchResponseFilter, ShareFetchResponseFilter, ShareAcknowledgeResponseFilter {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(BrokerAddressFilter.class);
 
@@ -129,6 +133,32 @@ public class BrokerAddressFilter implements MetadataResponseFilter, FindCoordina
         return context.forwardResponse(header, response);
     }
 
+    @Override
+    public CompletionStage<ResponseFilterResult> onShareAcknowledgeResponse(short apiVersion, ResponseHeaderData header, ShareAcknowledgeResponseData response,
+                                                                            FilterContext context) {
+        // KIP-932
+        if (response.nodeEndpoints() != null) {
+            response.nodeEndpoints()
+                    .forEach(ne -> apply(context, ne, ShareAcknowledgeResponseData.NodeEndpoint::nodeId,
+                            ShareAcknowledgeResponseData.NodeEndpoint::host, ShareAcknowledgeResponseData.NodeEndpoint::port,
+                            ShareAcknowledgeResponseData.NodeEndpoint::setHost, ShareAcknowledgeResponseData.NodeEndpoint::setPort));
+        }
+        return context.forwardResponse(header, response);
+    }
+
+    @Override
+    public CompletionStage<ResponseFilterResult> onShareFetchResponse(short apiVersion, ResponseHeaderData header, ShareFetchResponseData response,
+                                                                      FilterContext context) {
+        // KIP-932
+        if (response.nodeEndpoints() != null) {
+            response.nodeEndpoints()
+                    .forEach(ne -> apply(context, ne, ShareFetchResponseData.NodeEndpoint::nodeId,
+                            ShareFetchResponseData.NodeEndpoint::host, ShareFetchResponseData.NodeEndpoint::port,
+                            ShareFetchResponseData.NodeEndpoint::setHost, ShareFetchResponseData.NodeEndpoint::setPort));
+        }
+        return context.forwardResponse(header, response);
+    }
+
     private <T> void apply(FilterContext context, T broker, Function<T, Integer> nodeIdGetter, Function<T, String> hostGetter, ToIntFunction<T> portGetter,
                            BiConsumer<T, String> hostSetter,
                            ObjIntConsumer<T> portSetter) {


Test patch:
diff --git a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
index bb6baf4e..be199df6 100644
--- a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
+++ b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
@@ -43,6 +43,8 @@ import static org.apache.kafka.common.protocol.ApiKeys.CONTROLLED_SHUTDOWN;
 import static org.apache.kafka.common.protocol.ApiKeys.DESCRIBE_CLUSTER;
 import static org.apache.kafka.common.protocol.ApiKeys.FIND_COORDINATOR;
 import static org.apache.kafka.common.protocol.ApiKeys.METADATA;
+import static org.apache.kafka.common.protocol.ApiKeys.SHARE_ACKNOWLEDGE;
+import static org.apache.kafka.common.protocol.ApiKeys.SHARE_FETCH;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 @ExtendWith(NettyLeakDetectorExtension.class)
@@ -56,7 +58,7 @@ public class ProxyRpcTest {
      * API_VERSIONS is not proxied, kroxylicious can respond to this itself
      * FIND_COORDINATOR, METADATA, DESCRIBE_CLUSTER, kroxylicious takes charge of rewriting these responses itself.
      */
-    private static final Set<ApiKeys> SKIPPED_API_KEYS = Set.of(API_VERSIONS, FIND_COORDINATOR, METADATA, DESCRIBE_CLUSTER);
+    private static final Set<ApiKeys> SKIPPED_API_KEYS = Set.of(API_VERSIONS, FIND_COORDINATOR, METADATA, DESCRIBE_CLUSTER, SHARE_ACKNOWLEDGE, SHARE_FETCH);
 
     @BeforeAll
     public static void beforeAll() {
diff --git a/kroxylicious-runtime/src/test/resources/io/kroxylicious/proxy/internal/filter/ShareAcknowledge.test.yaml b/kroxylicious-runtime/src/test/resources/io/kroxylicious/proxy/internal/filter/ShareAcknowledge.test.yaml
new file mode 100644
index 00000000..9fcd78e1
--- /dev/null
+++ b/kroxylicious-runtime/src/test/resources/io/kroxylicious/proxy/internal/filter/ShareAcknowledge.test.yaml
@@ -0,0 +1,28 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: SHARE_ACKNOWLEDGE
+  description: Node endpoints rewritten
+  version: 0
+  response:
+    payload:
+      errorCode: 0
+      errorMessage: ""
+      responses: []
+      nodeEndpoints:
+        - nodeId: 0
+          host: upstreamz
+          port: 9199
+          rack: a
+      throttleTimeMs: 0
+    diff:
+      - op: replace
+        path: "/nodeEndpoints/0/host"
+        value: downstream
+      - op: replace
+        path: "/nodeEndpoints/0/port"
+        value: 19199
+  disabled: false
diff --git a/kroxylicious-runtime/src/test/resources/io/kroxylicious/proxy/internal/filter/ShareFetch.test.yaml b/kroxylicious-runtime/src/test/resources/io/kroxylicious/proxy/internal/filter/ShareFetch.test.yaml
new file mode 100644
index 00000000..054b6e4c
--- /dev/null
+++ b/kroxylicious-runtime/src/test/resources/io/kroxylicious/proxy/internal/filter/ShareFetch.test.yaml
@@ -0,0 +1,28 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: SHARE_FETCH
+  description: Node endpoints rewritten
+  version: 0
+  response:
+    payload:
+      errorCode: 0
+      errorMessage: ""
+      responses: []
+      nodeEndpoints:
+        - nodeId: 0
+          host: upstreamz
+          port: 9199
+          rack: a
+      throttleTimeMs: 0
+    diff:
+      - op: replace
+        path: "/nodeEndpoints/0/host"
+        value: downstream
+      - op: replace
+        path: "/nodeEndpoints/0/port"
+        value: 19199
+  disabled: false