Repository: kroxylicious/kroxylicious (java)

Issue:
kcat consumer terminates when kroxylicious is restarted despite the -E (Do not exit on non fatal error) flag
_Please use this to only for bug reports_. For questions or when you need help, you can use the [GitHub Discussions](https://github.com/kroxylicious/kroxylicious/discussions) or use the [community Slack](https://kroxylicious.slack.com) chat.

**Describe the bug**

Testing the work of #447,  I find that whilst a kcat producer is tolerate to a kroxylicios restart, a kcat consumer is not.
It tries to reconnect, but encounters this condition which it considers to be FATAL, and so exits.


```
% ERROR: Topic my-topic [0] error: my-topic [0]: desired partition is no longer available (Local: Unknown partition)
```

A clear and concise description of what the bug is.

**To Reproduce**
Steps to reproduce the behavior:
1. Setup with `./scripts/run-with-strimzi.sh kubernetes-examples/portperbroker_plain`
2. Run the kcat example commands (those being added by #469), but include the `-E` flag to the kcat command line.  Use kcat `1.6.0` so you can send messages from the producer one by one.
3. Send messages from consumer to producer iteratively.
4. `kubectl delete pod -n kafka kroxylicious-proxy..`
5. Observe that `kcat -P` tolerates the dropped connections, and reconnects.
6. Observe the `kcat -C` quits with the error above. 

**Expected behavior**

Both `kcat -P -E` and `kcat -C -E` should reconnect and continue to allow producing and consuming of messages.

**YAML files and logs**

Attach or copy paste your kroxylicious configuration YAML (with sensitive details
like broker address redacted).
Attach or copy and paste relevant logs.

**Additional context**

The Java Kafka console producer and console consumer are resilient to kroxylicious restarts.


Pull request description:
Fix #470: Ensure that the EagerMetadataLearner passes on a client's metadata request with fidelity
### Type of change

_Select the type of your PR_

- Bugfix

### Description

If `EagerMetadataLearner` receives a Metadata request RPC, it must forward that rather than substituting an empty object.
Why: The response was confusing `kcat` and causing it to abandon the reconnection.


### 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
- [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
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index abb5b49b..8a947dbe 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
 
+* [#470](https://github.com/kroxylicious/kroxylicious/issues/470): Ensure that the EagerMetadataLearner passes on a client's metadata request with fidelity (fix for kcat -C -E)
 * [#416](https://github.com/kroxylicious/kroxylicious/issues/416): Eagerly expose broker endpoints on startup to allow existing client to reconnect (without connecting to bootstrap).
 * [#463](https://github.com/kroxylicious/kroxylicious/issues/463): deregister micrometer hooks, meters and the registry on shutdown
 * [#443](https://github.com/kroxylicious/kroxylicious/pull/443): Obtain upstream ApiVersions when proxy is not SASL offloading
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearner.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearner.java
index 1d44a150..2de82316 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearner.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearner.java
@@ -50,9 +50,12 @@ public class EagerMetadataLearner implements RequestFilter {
             final short apiVersion = determineMetadataApiVersion(header);
             // Send an out-of-band Metadata request. The response will be intercepted by the in-built BrokerAddressFilter.
             // By the time control returns to the handler, the upstream addresses will have been reconciled.
-            var unused = filterContext.<MetadataResponseData> sendRequest(apiVersion, new MetadataRequestData())
+            boolean useClientRequest = apiKey.equals(ApiKeys.METADATA) && apiVersion == header.requestApiVersion();
+            var request = useClientRequest ? (MetadataRequestData) body : new MetadataRequestData();
+
+            var unused = filterContext.<MetadataResponseData> sendRequest(apiVersion, request)
                     .thenAccept(metadataResponseData -> {
-                        if (apiKey.equals(ApiKeys.METADATA) && apiVersion == header.requestApiVersion()) {
+                        if (useClientRequest) {
                             // The client's requested matched our out-of-band request, so we may as well return the
                             // response.
                             filterContext.forwardResponse(metadataResponseData);
diff --git a/kubernetes-examples/portperbroker_plain/postinstall.sh b/kubernetes-examples/portperbroker_plain/postinstall.sh
index 18255472..2feae49f 100755
--- a/kubernetes-examples/portperbroker_plain/postinstall.sh
+++ b/kubernetes-examples/portperbroker_plain/postinstall.sh
@@ -7,7 +7,30 @@
 
 set -eo pipefail
 
+STRIMZI_KAFKA=quay.io/strimzi/kafka:0.35.0-kafka-3.4.0
+KCAT=quay.io/kroxylicious/kcat:1.7.1
+TOPIC=my-topic
+BOOTSTRAP=kroxylicious-service:9292
+
+GREEN='\033[0;32m'
+NOCOLOUR='\033[0m'
+
+echo "Here are some commands to try things out. Run the producer and consumer in separate terminals"
+
+echo -e "${GREEN}Apache Kafka Client${NOCOLOUR}"
+
+echo "To produce to kroxylicious:"
+echo "kubectl -n kafka run java-kafka-producer -ti --image=${STRIMZI_KAFKA} --rm=true --restart=Never -- bin/kafka-console-producer.sh --bootstrap-server ${BOOTSTRAP} --topic ${TOPIC}"
+echo "To consume from kroxylicious:"
+echo "kubectl -n kafka run java-kafka-consumer -ti --image=${STRIMZI_KAFKA} --rm=true --restart=Never -- bin/kafka-console-consumer.sh --bootstrap-server ${BOOTSTRAP} --topic ${TOPIC} --from-beginning"
+
+echo -e "${GREEN}Edenhill Kcat (librdkafka)${NOCOLOUR}"
+
+echo "Note: kcat 1.7 -P will send each line as a separate record, but you need to press ^D to actually get kcat to start sending - see  https://github.com/edenhill/kcat/issues/322"
+echo "quay.io/kroxylicious/kcat:1.6.0 will send each line as a record the newline."
+
 echo "To produce to kroxylicious:"
-echo "kubectl -n kafka run kafka-producer -ti --image=quay.io/strimzi/kafka:0.35.0-kafka-3.4.0 --rm=true --restart=Never -- bin/kafka-console-producer.sh --bootstrap-server kroxylicious-service:9292 --topic my-topic"
+echo "kubectl -n kafka run kcat-kafka-producer -ti --image=${KCAT} --rm=true --restart=Never -- -P -t ${TOPIC} -b ${BOOTSTRAP}"
 echo "To consume from kroxylicious:"
-echo "kubectl -n kafka run kafka-consumer -ti --image=quay.io/strimzi/kafka:0.35.0-kafka-3.4.0 --rm=true --restart=Never -- bin/kafka-console-consumer.sh --bootstrap-server kroxylicious-service:9292 --topic my-topic --from-beginning"
+echo "kubectl -n kafka run kcat-kafka-consumer -ti --image=${KCAT} --rm=true --restart=Never -- -C -t ${TOPIC} -b ${BOOTSTRAP} -o beginning"
+
diff --git a/pom.xml b/pom.xml
index dbf25086..eb55b973 100644
--- a/pom.xml
+++ b/pom.xml
@@ -33,7 +33,7 @@
         <freemarker.version>2.3.32</freemarker.version>
         <guava.version>32.1.1-jre</guava.version>
         <jackson.version>2.15.2</jackson.version>
-        <junit.version>5.9.3</junit.version>
+        <junit.version>5.10.0</junit.version>
         <kafka.version>3.5.0</kafka.version>
         <kroxy.extension.version>0.4.0</kroxy.extension.version>
         <log4j.version>2.20.0</log4j.version>


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java
index 71092738..f10f4f1e 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/MetricsIT.java
@@ -12,6 +12,9 @@ import java.util.Map;
 import java.util.Random;
 import java.util.regex.Pattern;
 
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
@@ -31,6 +34,16 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 @ExtendWith(KafkaClusterExtension.class)
 public class MetricsIT {
 
+    @BeforeEach
+    public void beforeEach() {
+        Assertions.assertThat(Metrics.globalRegistry.getMeters()).isEmpty();
+    }
+
+    @AfterEach
+    public void afterEach() throws Exception {
+        Assertions.assertThat(Metrics.globalRegistry.getMeters()).isEmpty();
+    }
+
     @Test
     public void shouldOfferPrometheusMetricsScrapeEndpoint(KafkaCluster cluster) {
         var config = proxy(cluster)
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearnerTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearnerTest.java
index e30278c2..8dfc306d 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearnerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/EagerMetadataLearnerTest.java
@@ -6,11 +6,13 @@
 
 package io.kroxylicious.proxy.internal.filter;
 
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.Stream;
 
 import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.MetadataRequestData.MetadataRequestTopic;
 import org.apache.kafka.common.message.MetadataResponseData;
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.RequestHeaderData;
@@ -35,6 +37,7 @@ import org.mockito.junit.jupiter.MockitoExtension;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
 
 import static org.mockito.ArgumentMatchers.anyShort;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isA;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -55,43 +58,52 @@ class EagerMetadataLearnerTest {
 
     public static Stream<Arguments> preludeRequests() {
         return Stream.of(
-                toArgs(new ApiVersionsRequest(new ApiVersionsRequestData(), ApiVersionsRequestData.HIGHEST_SUPPORTED_VERSION)),
-                toArgs(new SaslHandshakeRequest(new SaslHandshakeRequestData(), SaslHandshakeRequestData.HIGHEST_SUPPORTED_VERSION)),
-                toArgs(new SaslAuthenticateRequest(new SaslAuthenticateRequestData(), SaslHandshakeRequestData.HIGHEST_SUPPORTED_VERSION)));
+                toArgs("ApiVersionsRequest", new ApiVersionsRequest(new ApiVersionsRequestData(), ApiVersionsRequestData.HIGHEST_SUPPORTED_VERSION)),
+                toArgs("SaslHandshakeRequest", new SaslHandshakeRequest(new SaslHandshakeRequestData(), SaslHandshakeRequestData.HIGHEST_SUPPORTED_VERSION)),
+                toArgs("SaslAuthenticateRequest", new SaslAuthenticateRequest(new SaslAuthenticateRequestData(), SaslHandshakeRequestData.HIGHEST_SUPPORTED_VERSION)));
     }
 
-    @ParameterizedTest
+    @ParameterizedTest(name = "{0}")
     @MethodSource("preludeRequests")
-    public void forwardsRequestsOfKafkaPrelude(ApiKeys apiKey, RequestHeaderData header, ApiMessage request) {
+    public void forwardsRequestsOfKafkaPrelude(String name, ApiKeys apiKey, RequestHeaderData header, ApiMessage request) {
         learner.onRequest(apiKey, header, request, context);
         verify(context).forwardRequest(header, request);
     }
 
     public static Stream<Arguments> postPreludeRequests() {
         return Stream.of(
-                toArgs(new ProduceRequest(new ProduceRequestData(), ProduceRequestData.HIGHEST_SUPPORTED_VERSION)),
-                toArgs(new MetadataRequest(new MetadataRequestData(), MetadataRequestData.HIGHEST_SUPPORTED_VERSION)),
-                toArgs(new MetadataRequest(new MetadataRequestData(), MetadataRequestData.LOWEST_SUPPORTED_VERSION)));
+                toArgs("ProduceRequest replaced by MetadataRequest", new ProduceRequest(new ProduceRequestData(), ProduceRequestData.HIGHEST_SUPPORTED_VERSION)),
+                toArgs("MetadataRequest (highest supported)", new MetadataRequest(new MetadataRequestData(), MetadataRequestData.HIGHEST_SUPPORTED_VERSION)),
+                toArgs("MetadataRequest (lowest supported)", new MetadataRequest(new MetadataRequestData(), MetadataRequestData.LOWEST_SUPPORTED_VERSION)),
+                toArgs("MetadataRequest (payload fidelity)", new MetadataRequest(new MetadataRequestData().setTopics(List.of(new MetadataRequestTopic().setName("foo"))),
+                        MetadataRequestData.LOWEST_SUPPORTED_VERSION)));
     }
 
-    @ParameterizedTest
+    @ParameterizedTest(name = "{0}")
     @MethodSource("postPreludeRequests")
-    public void spontaneouslyEmitsMetadataRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage request) {
+    public void spontaneouslyEmitsMetadataRequest(String name, ApiKeys apiKey, RequestHeaderData header, ApiMessage request) {
         var metadataResponse = new MetadataResponseData();
         metadataResponse.brokers().add(new MetadataResponseData.MetadataResponseBroker().setNodeId(1).setHost("localhost").setPort(1234));
 
         when(context.sendRequest(anyShort(), isA(MetadataRequestData.class))).thenReturn(CompletableFuture.completedStage(metadataResponse));
         learner.onRequest(apiKey, header, request, context);
 
+        if (apiKey == ApiKeys.METADATA) {
+            // if caller's request is a metadata request, then the filter must forward it with fidelity
+            verify(context).sendRequest(eq(header.requestApiVersion()), eq(request));
+        }
+        else {
+            verify(context).sendRequest(anyShort(), isA(MetadataRequestData.class));
+        }
         verify(context, apiKey.equals(ApiKeys.METADATA) ? times(1) : never()).forwardResponse(metadataResponse);
         verify(context).closeConnection();
     }
 
-    private static Arguments toArgs(AbstractRequest request) {
+    private static Arguments toArgs(String name, AbstractRequest request) {
         var header = new RequestHeaderData().setRequestApiKey(request.apiKey().id).setRequestApiVersion(request.version());
         var apiKey = request.apiKey();
         var request1 = request.data();
-        return Arguments.of(apiKey, header, request1);
+        return Arguments.of(name, apiKey, header, request1);
     }
 
-}
\ No newline at end of file
+}