Fix kroxylicious#1612: [Kafka Unstable] Rewrite broker address information carried by ShareFetchResponse and ShareAckResponse (KIP-932) (kroxylicious#1656)

Kafka 3.9 introduces the new RPC that carry broker host/port information.  The feature is marked as unstable, but we said we want to implement it
early to allow Filter Authors to experiment with the new Kafka features, if they wish.

Signed-off-by: kwall <kwall@apache.org>

Author: k-wall

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java

@@ -43,6 +43,8 @@
 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 record Scenario(String name, ResponsePayload givenMockResponse, Request w
      * 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() {

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/filter/BrokerAddressFilter.java

@@ -22,6 +22,8 @@
 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.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 @@
  * 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 CompletionStage<ResponseFilterResult> onFetchResponse(short apiVersion, R
         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) {

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

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