Hoist the special casing out of BodyDecoder and into KafkaResponseDecoder

Signed-off-by: Sam Barker <sam@quadrocket.co.uk>

Author: SamBarker

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaResponseDecoder.java

@@ -22,11 +22,13 @@
 import io.kroxylicious.proxy.frame.OpaqueResponseFrame;
 import io.kroxylicious.proxy.internal.InternalResponseFrame;
 
+import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
 
 public class KafkaResponseDecoder extends KafkaMessageDecoder {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaResponseDecoder.class);
+    public static final short EMERGENCY_API_VERSION = (short) 0;
 
     private final CorrelationManager correlationManager;
 
@@ -70,14 +72,14 @@ else if (LOGGER.isDebugEnabled()) {
             log().trace("{}: Header version: {}", ctx, headerVersion);
             ResponseHeaderData header = readHeader(headerVersion, accessor);
             log().trace("{}: Header: {}", ctx, header);
-            ApiMessage body = BodyDecoder.decodeResponse(apiKey, apiVersion, accessor);
+            ApiMessageVersion body = decodeBody(apiKey, apiVersion, accessor);
             log().trace("{}: Body: {}", ctx, body);
             Filter recipient = correlation.recipient();
             if (recipient == null) {
-                frame = new DecodedResponseFrame<>(apiVersion, correlationId, header, body);
+                frame = new DecodedResponseFrame<>(body.apiVersion(), correlationId, header, body.apiMessage());
             }
             else {
-                frame = new InternalResponseFrame<>(recipient, apiVersion, correlationId, header, body, correlation.promise());
+                frame = new InternalResponseFrame<>(recipient, body.apiVersion(), correlationId, header, body.apiMessage(), correlation.promise());
             }
         }
         else {
@@ -87,6 +89,26 @@ else if (LOGGER.isDebugEnabled()) {
         return frame;
     }
 
+    @NonNull
+    private static ApiMessageVersion decodeBody(ApiKeys apiKey, short apiVersion, ByteBufAccessorImpl accessor) {
+        int prev = accessor.readerIndex();
+        try {
+            return new ApiMessageVersion(BodyDecoder.decodeResponse(apiKey, apiVersion, accessor), apiVersion);
+        }
+        catch (RuntimeException e) {
+            // KIP-511 when the client receives an unsupported version for the ApiVersionResponse, it fails back to version 0
+            // Use the same algorithm as
+            // https://github.com/apache/kafka/blob/a41c10fd49841381b5207c184a385622094ed440/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java#L90-L106
+            accessor.readerIndex(prev);
+            if (ApiKeys.API_VERSIONS.equals(apiKey) && apiVersion != 0) {
+                return new ApiMessageVersion(BodyDecoder.decodeResponse(apiKey, EMERGENCY_API_VERSION, accessor), EMERGENCY_API_VERSION);
+            }
+            else {
+                throw e;
+            }
+        }
+    }
+
     private OpaqueFrame opaqueFrame(short apiKeyId, short apiVersion, ByteBuf in, int correlationId, int length) {
         return new OpaqueResponseFrame(apiKeyId, apiVersion, in.readSlice(length).retain(), correlationId, length);
     }
@@ -95,4 +117,5 @@ private ResponseHeaderData readHeader(short headerVersion, Readable accessor) {
         return new ResponseHeaderData(accessor, headerVersion);
     }
 
+    record ApiMessageVersion(ApiMessage apiMessage, short apiVersion) {}
 }

kroxylicious-runtime/src/main/templates/BodyDecoder.ftl

@@ -70,27 +70,7 @@ public class BodyDecoder {
         return switch (apiKey) {
 <#list messageSpecs as messageSpec>
     <#if messageSpec.type?lower_case == 'response'>
-        <#if messageSpec.name == 'ApiVersionsResponse'>
-            case ${retrieveApiKey(messageSpec)} -> {
-                // KIP-511 when the client receives an unsupported version for the ApiVersionResponse, it fails back to version 0
-                // Use the same algorithm as https://github.com/apache/kafka/blob/a41c10fd49841381b5207c184a385622094ed440/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java#L90-L106
-                int prev = accessor.readerIndex();
-                try {
-                    yield new ${messageSpec.name}Data(accessor, apiVersion);
-                }
-                catch (RuntimeException e) {
-                    accessor.readerIndex(prev);
-                    if (apiVersion != 0) {
-                        yield new ${messageSpec.name}Data(accessor, (short) 0);
-                    }
-                    else {
-                        throw e;
-                    }
-                }
-            }
-        <#else>
             case ${retrieveApiKey(messageSpec)} -> new ${messageSpec.name}Data(accessor, apiVersion);
-        </#if>
     </#if>
 </#list>