Test KafkaClient ensures frame entirely read by header/body decode

We want to use KafkaClient to declare what response apiVersion we expect
the proxy to respond with. The kafka-client code we depend on to read
the messages will sometimes be backwards compatible, for example if we
try to read a v1 or v2 ApiVersions response message, then it will
successfully read the v0 portion of the message, leaving some additional
fields like throttleTimeMs unread.

By KIP-511 the contract is we will send a v0 response, not a v0
compatible response, so I think it's good for the test client to require
that all bytes are read at the response apiVersion we declare we expect.

Signed-off-by: Robert Young <[email protected]>

Author: robobario

kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/codec/KafkaResponseDecoder.java

@@ -67,6 +67,9 @@ else if (LOGGER.isDebugEnabled()) {
             ApiMessage body = BodyDecoder.decodeResponse(apiKey, apiVersion, accessor);
             log().trace("{}: Body: {}", ctx, body);
             frame = new DecodedResponseFrame<>(apiVersion, correlationId, header, body);
+            if (in.readableBytes() != 0) {
+                throw new RuntimeException("Unread bytes remaining in frame, potentially response api version differs from expectation");
+            }
             correlation.responseFuture().complete(new SequencedResponse(frame, i++));
             return frame;
         }

kroxylicious-integration-test-support/src/test/java/io/kroxylicious/test/client/KafkaClientTest.java

@@ -103,6 +103,21 @@ void testClientCanHandleResponseApiVersionDifferentFromRequestApiVersion() {
         }
     }
 
+    @Test
+    void unreadBytesAfterFrameDecodeThrowsException() {
+        ApiVersionsResponseData message = new ApiVersionsResponseData();
+        message.setErrorCode(Errors.UNSUPPORTED_VERSION.code());
+        message.setThrottleTimeMs(22);
+        ResponsePayload v0Payload = new ResponsePayload(ApiKeys.API_VERSIONS, (short) 1, message);
+        try (var mockServer = MockServer.startOnRandomPort(v0Payload);
+                var kafkaClient = new KafkaClient("localhost", mockServer.port())) {
+            CompletableFuture<Response> future = kafkaClient.get(new Request(ApiKeys.API_VERSIONS, (short) 3, "client", new ApiVersionsRequestData(), (short) 0));
+            // fails to decode v0 response because client encodes v1, which is backwards compatible but emits additional bytes. We want to be as sure as we can be that it was a v0 response.
+            assertThat(future).failsWithin(10, TimeUnit.SECONDS).withThrowableThat()
+                    .withMessageContaining("Unread bytes remaining in frame, potentially response api version differs from expectation");
+        }
+    }
+
     @Test
     void unexpectedResponseFormatTriggersFailure() {
         ApiVersionsResponseData message = new ApiVersionsResponseData();