Fail fast if a client sends bytes for an apiVersion below the oldest

To workaround a librdkafka issue, the proxy (following the behaviour of
upstream Kafka) sets the min version for Produce RPCs to 0, despite only
being able to decode v3+.

If a client did send us v0-v2 messages then we will probably fail to
decode the message, and the kafka message read will throw some unhelpful
message, instead we can fail earlier with a better message logged.

Note: there was some confusion over the meaning of the terms used by
ApiKeys:
- Supported means it can be decoded, so it encompasses the full
  range from oldest to latest unstable version.
- Deprecated versions are supported versions that are eligible for
  deletion in future. If there are deprecated versions, then the
  deprecated version range will be a subset of the supported version
  range. Else it will be the range 0 to -1 when there are no
  deprecations.
- The term enabled is used to talk about the supported versions, but
  depending on whether the broker has enabled unstableLastVersion.
  A supported version may not be enabled due to broker settings in
  upstream apache kafka.

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

Author: robobario

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

@@ -56,7 +56,10 @@ public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
                 }
             }
             catch (Exception e) {
-                log().error("{}: Error in decoder", ctx, e);
+                log().atError()
+                        .setMessage("{}: Error in decoder: " + e.getMessage())
+                        .addArgument(ctx)
+                        .setCause(log().isDebugEnabled() ? e : null).log();
                 throw e;
             }
         }

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

@@ -5,6 +5,7 @@
  */
 package io.kroxylicious.proxy.internal.codec;
 
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -70,16 +71,21 @@ protected Frame decodeHeaderAndBody(ChannelHandlerContext ctx, ByteBuf in, final
         final RequestFrame frame;
         if (decodeRequest) {
             short highestProxyVersion = apiVersionsService.latestVersion(apiKey);
+            short oldestProxyVersion = apiKey.oldestVersion();
             boolean clientAheadOfProxy = apiVersion > highestProxyVersion;
+            boolean proxyAheadOfClient = apiVersion < oldestProxyVersion;
             if (clientAheadOfProxy) {
                 if (apiKey == ApiKeys.API_VERSIONS) {
                     return createV0ApiVersionRequestFrame(ctx, correlationId);
                 }
                 else {
-                    log().error("{}: apiVersion {} for {} ahead of proxy maximum: {}", ctx, apiVersion, apiKey, highestProxyVersion);
-                    throw new IllegalStateException("client apiVersion " + apiVersion + " ahead of proxy maximum " + highestProxyVersion + " for api key: " + apiKey);
+                    throw new UnsupportedVersionException(
+                            "client apiVersion " + apiVersion + " ahead of proxy maximum " + highestProxyVersion + " for api key: " + apiKey);
                 }
             }
+            else if (proxyAheadOfClient) {
+                throw new UnsupportedVersionException("client apiVersion " + apiVersion + " below proxy minimum " + oldestProxyVersion + " for api key: " + apiKey);
+            }
             DecodedBufer result = decodeRequest(ctx, in, headerVersion, sof);
             ApiMessage body = BodyDecoder.decodeRequest(apiKey, apiVersion, result.accessor());
             if (log().isTraceEnabled()) {

kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoderTest.java

@@ -15,6 +15,7 @@
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.RequestHeaderData;
@@ -125,7 +126,7 @@ void decodeUnknownApiVersions() {
 
     // after ApiVersions negotiation we should never encounter a request from the client for an api version unknown to the proxy
     @Test
-    void throwsOnUnsupportedVersionOfNonApiVersionsRequests() {
+    void throwsOnUnsupportedVersion_NonApiVersionsRequestsAheadOfProxyMaximum() {
         EmbeddedChannel embeddedChannel = newEmbeddedChannel(new ApiVersionsServiceImpl(), RequestDecoderTest.DECODE_EVERYTHING);
         short maxSupportedVersion = ApiKeys.METADATA.latestVersion(true);
         short unsupportedVersion = (short) (maxSupportedVersion + 1);
@@ -140,10 +141,30 @@ void throwsOnUnsupportedVersionOfNonApiVersionsRequests() {
         accessor.writeInt(messageSize);
         header.write(accessor, cache, requestHeaderVersion);
         accessor.writeByteArray(arbitraryBodyBytes);
-        assertThatThrownBy(() -> embeddedChannel.writeInbound(buffer)).isInstanceOf(DecoderException.class).cause().isInstanceOf(IllegalStateException.class)
+        assertThatThrownBy(() -> embeddedChannel.writeInbound(buffer)).isInstanceOf(DecoderException.class).cause().isInstanceOf(UnsupportedVersionException.class)
                 .hasMessage("client apiVersion %d ahead of proxy maximum %d for api key: METADATA", unsupportedVersion, maxSupportedVersion);
     }
 
+    @Test
+    void throwsOnUnsupportedVersionRequestBelowProxyMinimum() {
+        EmbeddedChannel embeddedChannel = newEmbeddedChannel(new ApiVersionsServiceImpl(), RequestDecoderTest.DECODE_EVERYTHING);
+        short minSupportedVersion = ApiKeys.PRODUCE.oldestVersion();
+        short unsupportedVersion = (short) (minSupportedVersion - 1);
+        RequestHeaderData header = latestVersionHeaderWithAllFields(ApiKeys.PRODUCE, unsupportedVersion);
+        byte[] arbitraryBodyBytes = new byte[]{ 1, 2, 3, 4 };
+        ObjectSerializationCache cache = new ObjectSerializationCache();
+        short requestHeaderVersion = ApiKeys.PRODUCE.requestHeaderVersion(ApiKeys.PRODUCE.latestVersion());
+        int headerSize = header.size(cache, requestHeaderVersion);
+        int messageSize = headerSize + arbitraryBodyBytes.length;
+        ByteBuf buffer = Unpooled.buffer();
+        ByteBufAccessorImpl accessor = new ByteBufAccessorImpl(buffer);
+        accessor.writeInt(messageSize);
+        header.write(accessor, cache, requestHeaderVersion);
+        accessor.writeByteArray(arbitraryBodyBytes);
+        assertThatThrownBy(() -> embeddedChannel.writeInbound(buffer)).isInstanceOf(DecoderException.class).cause().isInstanceOf(UnsupportedVersionException.class)
+                .hasMessage("client apiVersion %d below proxy minimum %d for api key: PRODUCE", unsupportedVersion, minSupportedVersion);
+    }
+
     private static Stream<Arguments> produceRequestCases() {
         return Stream.concat(produceRequestsDoNotRequireResponse((short) 0, false), produceRequestsDoNotRequireResponse((short) 1, true));
     }

kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderMockitoTest.java

@@ -9,6 +9,7 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.kafka.common.protocol.ApiKeys;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -57,6 +58,7 @@ void shouldRewindBufferIfNotEnoughBytesAvailable() {
 
         // Then
         verify(frameBuffer).readerIndex(initialIndex);
+
     }
 
     @SuppressWarnings("DataFlowIssue")
@@ -69,6 +71,7 @@ void shouldThrowOnPartialIfReadingPartialFrame() {
         when(frameBuffer.readInt()).thenReturn(10);
         when(frameBuffer.readableBytes()).thenReturn(20);
         when(frameBuffer.readSlice(anyInt())).thenReturn(frameSlice);
+        when(frameSlice.readShort()).thenReturn(ApiKeys.API_VERSIONS.id, (short) 0);
 
         List<Object> messageReceiver = new ArrayList<>();
 
@@ -77,6 +80,5 @@ void shouldThrowOnPartialIfReadingPartialFrame() {
         assertThatThrownBy(() -> kafkaRequestDecoder.decode(null, frameBuffer, messageReceiver))
                 .isInstanceOf(IllegalStateException.class)
                 .hasMessageContaining("decodeHeaderAndBody did not read all of the buffer");
-
     }
 }