KAFKA-18659: librdkafka compressed produce fails unless api versions returns produce v0

clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java

@@ -149,6 +149,15 @@ public enum ApiKeys {
     private static final Map<Integer, ApiKeys> ID_TO_TYPE = Arrays.stream(ApiKeys.values())
         .collect(Collectors.toMap(key -> (int) key.id, Function.identity()));
 
+    // Versions 0-2 were removed in Apache Kafka 4.0, version 3 is the new baseline. Due to a bug in librdkafka,
+    // these versions have to be included in the api versions response (see KAFKA-18659). In order to achieve that,
+    // we keep such versions in the protocol definition files, but override `oldestVersion` to return the correct value.
+    // We also adjust `toApiVersion` to return `0` for produce in the broker listener.
+    // An alternative approach would be to remove versions `0-2` from the protocol definition files and only override the
+    // behavior in this file - the main downside is that it would no longer be possible to send requests with produce v0-v2,
+    // which would make testing significantly harder (it would probably have to be a ducktape test).
+    public static final short PRODUCE_OLDEST_VERSION = 3;
+
     /** the permanent and immutable id of an API - this can't change ever */
     public final short id;
 
@@ -227,6 +236,9 @@ public short latestVersion(boolean enableUnstableLastVersion) {
     }
 
     public short oldestVersion() {
+        // See #PRODUCE_OLDEST_VERSION for details of why we do this
+        if (this == PRODUCE)
+            return PRODUCE_OLDEST_VERSION;
         return messageType.lowestSupportedVersion();
     }
 
@@ -264,8 +276,11 @@ public boolean hasValidVersion() {
         return oldestVersion() <= latestVersion();
     }
 
-    public Optional<ApiVersionsResponseData.ApiVersion> toApiVersion(boolean enableUnstableLastVersion) {
-        short oldestVersion = oldestVersion();
+    public Optional<ApiVersionsResponseData.ApiVersion> toApiVersion(boolean enableUnstableLastVersion,
+                                                                     Optional<ApiMessageType.ListenerType> listenerType) {
+        // see `PRODUCE_OLDEST_VERSION` for details on why we do this
+        short oldestVersion = (this == PRODUCE && listenerType.map(l -> l == ApiMessageType.ListenerType.BROKER).orElse(false)) ?
+            messageType.lowestSupportedVersion() : oldestVersion();
         short latestVersion = latestVersion(enableUnstableLastVersion);
 
         // API is entirely disabled if latestStableVersion is smaller than oldestVersion.
@@ -299,7 +314,7 @@ static String toHtml() {
         b.append("<th>Key</th>\n");
         b.append("</tr>");
         clientApis().stream()
-            .filter(apiKey -> apiKey.toApiVersion(false).isPresent())
+            .filter(apiKey -> apiKey.toApiVersion(false, Optional.empty()).isPresent())
             .forEach(apiKey -> {
                 b.append("<tr>\n");
                 b.append("<td>");
@@ -341,10 +356,7 @@ public static EnumSet<ApiKeys> controllerApis() {
     }
 
     public static EnumSet<ApiKeys> clientApis() {
-        List<ApiKeys> apis = Arrays.stream(ApiKeys.values())
-            .filter(apiKey -> apiKey.inScope(ApiMessageType.ListenerType.BROKER))
-            .collect(Collectors.toList());
-        return EnumSet.copyOf(apis);
+        return brokerApis();
     }
 
     public static EnumSet<ApiKeys> apisForListener(ApiMessageType.ListenerType listener) {

clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java

@@ -208,26 +208,26 @@ public static String toHtml() {
             // Responses
             b.append("<b>Responses:</b><br>\n");
             Schema[] responses = key.messageType.responseSchemas();
-            for (int i = 0; i < responses.length; i++) {
-                Schema schema = responses[i];
+            for (int version = key.oldestVersion(); version < key.latestVersion(); version++) {
+                Schema schema = responses[version];
+                if (schema == null)
+                    throw new IllegalStateException("Unexpected null schema for " + key + " with version " + version);
                 // Schema
-                if (schema != null) {
-                    b.append("<div>");
-                    // Version header
-                    b.append("<pre>");
-                    b.append(key.name);
-                    b.append(" Response (Version: ");
-                    b.append(i);
-                    b.append(") => ");
-                    schemaToBnfHtml(responses[i], b, 2);
-                    b.append("</pre>");
-
-                    b.append("<p><b>Response header version:</b> ");
-                    b.append(key.responseHeaderVersion((short) i));
-                    b.append("</p>\n");
-
-                    schemaToFieldTableHtml(responses[i], b);
-                }
+                b.append("<div>");
+                // Version header
+                b.append("<pre>");
+                b.append(key.name);
+                b.append(" Response (Version: ");
+                b.append(version);
+                b.append(") => ");
+                schemaToBnfHtml(responses[version], b, 2);
+                b.append("</pre>");
+
+                b.append("<p><b>Response header version:</b> ");
+                b.append(key.responseHeaderVersion((short) version));
+                b.append("</p>\n");
+
+                schemaToFieldTableHtml(responses[version], b);
                 b.append("</div>\n");
             }
         }

clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java

@@ -204,7 +204,7 @@ public static ApiVersionCollection filterApis(
             // Skip telemetry APIs if client telemetry is disabled.
             if ((apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey == ApiKeys.PUSH_TELEMETRY) && !clientTelemetryEnabled)
                 continue;
-            apiKey.toApiVersion(enableUnstableLastVersion).ifPresent(apiKeys::add);
+            apiKey.toApiVersion(enableUnstableLastVersion, Optional.of(listenerType)).ifPresent(apiKeys::add);
         }
         return apiKeys;
     }
@@ -215,7 +215,7 @@ public static ApiVersionCollection collectApis(
     ) {
         ApiVersionCollection res = new ApiVersionCollection();
         for (ApiKeys apiKey : apiKeys) {
-            apiKey.toApiVersion(enableUnstableLastVersion).ifPresent(res::add);
+            apiKey.toApiVersion(enableUnstableLastVersion, Optional.empty()).ifPresent(res::add);
         }
         return res;
     }
@@ -238,7 +238,7 @@ public static ApiVersionCollection intersectForwardableApis(
     ) {
         ApiVersionCollection apiKeys = new ApiVersionCollection();
         for (ApiKeys apiKey : ApiKeys.apisForListener(listenerType)) {
-            final Optional<ApiVersion> brokerApiVersion = apiKey.toApiVersion(enableUnstableLastVersion);
+            final Optional<ApiVersion> brokerApiVersion = apiKey.toApiVersion(enableUnstableLastVersion, Optional.of(listenerType));
             if (brokerApiVersion.isEmpty()) {
                 // Broker does not support this API key.
                 continue;

clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java

@@ -40,6 +40,7 @@
 import static org.apache.kafka.common.requests.ProduceResponse.INVALID_OFFSET;
 
 public class ProduceRequest extends AbstractRequest {
+
     public static final short LAST_STABLE_VERSION_BEFORE_TRANSACTION_V2 = 11;
 
     public static Builder builder(ProduceRequestData data, boolean useTransactionV1Version) {
@@ -66,21 +67,10 @@ public Builder(short minVersion,
 
         @Override
         public ProduceRequest build(short version) {
-            return build(version, true);
-        }
-
-        // Visible for testing only
-        public ProduceRequest buildUnsafe(short version) {
-            return build(version, false);
-        }
-
-        private ProduceRequest build(short version, boolean validate) {
-            if (validate) {
-                // Validate the given records first
-                data.topicData().forEach(tpd ->
-                        tpd.partitionData().forEach(partitionProduceData ->
-                                ProduceRequest.validateRecords(version, partitionProduceData.records())));
-            }
+            // Validate the given records first
+            data.topicData().forEach(tpd ->
+                tpd.partitionData().forEach(partitionProduceData ->
+                    ProduceRequest.validateRecords(version, partitionProduceData.records())));
             return new ProduceRequest(data, version);
         }
 
@@ -244,4 +234,5 @@ public static ProduceRequest parse(ByteBuffer buffer, short version) {
     public static boolean isTransactionV2Requested(short version) {
         return version > LAST_STABLE_VERSION_BEFORE_TRANSACTION_V2;
     }
+
 }

clients/src/main/resources/common/message/ProduceRequest.json

@@ -18,7 +18,9 @@
   "type": "request",
   "listeners": ["broker"],
   "name": "ProduceRequest",
-  // Versions 0-2 were removed in Apache Kafka 4.0, Version 3 is the new baseline.
+  // Versions 0-2 were removed in Apache Kafka 4.0, version 3 is the new baseline. Due to a bug in librdkafka,
+  // these versions have to be included in the api versions response (see KAFKA-18659) and are included in `validVersion`.
+  // See `ApiKeys.PRODUCE_OLDEST_VERSION` for more details.
   //
   // Version 1 and 2 are the same as version 0.
   //
@@ -44,7 +46,7 @@
   // transaction V2 (KIP_890 part 2) is enabled, the produce request will also include the function for a
   // AddPartitionsToTxn call. If V2 is disabled, the client can't use produce request version higher than 11 within
   // a transaction.
-  "validVersions": "3-12",
+  "validVersions": "0-12",
   "flexibleVersions": "9+",
   "fields": [
     { "name": "TransactionalId", "type": "string", "versions": "3+", "nullableVersions": "3+", "default": "null", "entityType": "transactionalId",

clients/src/main/resources/common/message/ProduceResponse.json

@@ -17,7 +17,9 @@
   "apiKey": 0,
   "type": "response",
   "name": "ProduceResponse",
-  // Versions 0-2 were removed in Apache Kafka 4.0, Version 3 is the new baseline.
+  // Versions 0-2 were removed in Apache Kafka 4.0, version 3 is the new baseline. Due to a bug in librdkafka,
+  // these versions have to be included in the api versions response (see KAFKA-18659) and are included in `validVersion`.
+  // See `ApiKeys.PRODUCE_OLDEST_VERSION` for more details.
   //
   // Version 1 added the throttle time.
   // Version 2 added the log append time.
@@ -38,7 +40,7 @@
   // Version 11 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
   //
   // Version 12 is the same as version 10 (KIP-890).
-  "validVersions": "3-12",
+  "validVersions": "0-12",
   "flexibleVersions": "9+",
   "fields": [
     { "name": "Responses", "type": "[]TopicProduceResponse", "versions": "0+",

clients/src/test/java/org/apache/kafka/common/protocol/ApiKeysTest.java

@@ -24,6 +24,7 @@
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashSet;
+import java.util.Optional;
 import java.util.Set;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -104,7 +105,7 @@ public void testHasValidVersions() {
     public void testHtmlOnlyHaveStableApi() {
         String html = ApiKeys.toHtml();
         for (ApiKeys apiKeys : ApiKeys.clientApis()) {
-            if (apiKeys.toApiVersion(false).isPresent()) {
+            if (apiKeys.toApiVersion(false, Optional.empty()).isPresent()) {
                 assertTrue(html.contains("The_Messages_" + apiKeys.name), "Html should contain stable api: " + apiKeys.name);
             } else {
                 assertFalse(html.contains("The_Messages_" + apiKeys.name), "Html should not contain unstable api: " + apiKeys.name);

clients/src/test/java/org/apache/kafka/common/requests/ApiVersionsResponseTest.java

@@ -58,8 +58,11 @@ public void shouldHaveCorrectDefaultApiVersionsResponse(ApiMessageType.ListenerT
         for (ApiKeys key : ApiKeys.apisForListener(scope)) {
             ApiVersion version = defaultResponse.apiVersion(key.id);
             assertNotNull(version, "Could not find ApiVersion for API " + key.name);
-            assertEquals(version.minVersion(), key.oldestVersion(), "Incorrect min version for Api " + key.name);
-            assertEquals(version.maxVersion(), key.latestVersion(), "Incorrect max version for Api " + key.name);
+            if (key == ApiKeys.PRODUCE)
+                assertEquals(key.messageType.lowestSupportedVersion(), version.minVersion(), "Incorrect min version for Api " + key.name);
+            else
+                assertEquals(key.oldestVersion(), version.minVersion(), "Incorrect min version for Api " + key.name);
+            assertEquals(key.latestVersion(), version.maxVersion(), "Incorrect max version for Api " + key.name);
 
             // Check if versions less than min version are indeed set as null, i.e., deprecated.
             for (int i = 0; i < version.minVersion(); ++i) {

clients/src/test/java/org/apache/kafka/common/requests/ProduceRequestTest.java

@@ -212,8 +212,7 @@ public void testV6AndBelowCannotUseZStdCompression() {
             .setAcks((short) 1)
             .setTimeoutMs(1000);
         // Can't create ProduceRequest instance with version within [3, 7)
-        for (short version = 3; version < 7; version++) {
-
+        for (short version = ApiKeys.PRODUCE.oldestVersion(); version < 7; version++) {
             ProduceRequest.Builder requestBuilder = new ProduceRequest.Builder(version, version, produceData);
             assertThrowsForAllVersions(requestBuilder, UnsupportedCompressionTypeException.class);
         }
@@ -277,6 +276,21 @@ public void testMixedIdempotentData() {
         assertTrue(RequestTestUtils.hasIdempotentRecords(request));
     }
 
+    @Test
+    public void testBuilderOldestAndLatestAllowed() {
+        ProduceRequest.Builder builder = ProduceRequest.builder(new ProduceRequestData()
+            .setTopicData(new ProduceRequestData.TopicProduceDataCollection(Collections.singletonList(
+                new ProduceRequestData.TopicProduceData()
+                    .setName("topic")
+                    .setPartitionData(Collections.singletonList(new ProduceRequestData.PartitionProduceData()
+                    .setIndex(1)
+                    .setRecords(MemoryRecords.withRecords(Compression.NONE, simpleRecord))))).iterator()))
+            .setAcks((short) -1)
+            .setTimeoutMs(10));
+        assertEquals(ApiKeys.PRODUCE_OLDEST_VERSION, builder.oldestAllowedVersion());
+        assertEquals(ApiKeys.PRODUCE.latestVersion(), builder.latestAllowedVersion());
+    }
+
     private static <T extends Throwable> void assertThrowsForAllVersions(ProduceRequest.Builder builder,
                                                                          Class<T> expectedType) {
         IntStream.range(builder.oldestAllowedVersion(), builder.latestAllowedVersion() + 1)

core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala

@@ -116,5 +116,10 @@ abstract class AbstractApiVersionsRequestTest(cluster: ClusterInstance) {
       assertEquals(expectedApiVersion.minVersion, actualApiVersion.minVersion, s"Received unexpected min version for API key ${actualApiVersion.apiKey}.")
       assertEquals(expectedApiVersion.maxVersion, actualApiVersion.maxVersion, s"Received unexpected max version for API key ${actualApiVersion.apiKey}.")
     }
+
+    if (listenerName.equals(cluster.clientListener)) {
+      // See ApiKeys.PRODUCE_OLDEST_VERSION for details on why this is `0` (instead of `3`)
+      assertEquals(0, apiVersionsResponse.apiVersion(ApiKeys.PRODUCE.id).minVersion)
+    }
   }
 }

core/src/test/scala/unit/kafka/server/BaseRequestTest.scala

@@ -78,7 +78,7 @@ abstract class BaseRequestTest extends IntegrationTestHarness {
     new Socket("localhost", socketServer.boundPort(listenerName))
   }
 
-  private def sendRequest(socket: Socket, request: Array[Byte]): Unit = {
+  private[server] def sendRequest(socket: Socket, request: Array[Byte]): Unit = {
     val outgoing = new DataOutputStream(socket.getOutputStream)
     outgoing.writeInt(request.length)
     outgoing.write(request)

core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala

@@ -27,15 +27,18 @@ import org.apache.kafka.common.config.TopicConfig
 import org.apache.kafka.common.message.ProduceRequestData
 import org.apache.kafka.common.protocol.{ApiKeys, Errors}
 import org.apache.kafka.common.record._
-import org.apache.kafka.common.requests.{ProduceRequest, ProduceResponse}
+import org.apache.kafka.common.requests.{ProduceRequest, ProduceResponse, RequestUtils}
+import org.apache.kafka.common.utils.Utils
 import org.apache.kafka.server.metrics.KafkaYammerMetrics
 import org.apache.kafka.server.record.BrokerCompressionType
 import org.apache.kafka.storage.log.metrics.BrokerTopicMetrics
 import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
 import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.{Arguments, MethodSource}
 import org.junit.jupiter.params.provider.ValueSource
 
+import java.io.EOFException
 import java.util.concurrent.TimeUnit
 import scala.jdk.CollectionConverters._
 
@@ -254,7 +257,7 @@ class ProduceRequestTest extends BaseRequestTest {
     // Create a single-partition topic compressed with ZSTD
     val topicConfig = new Properties
     topicConfig.setProperty(TopicConfig.COMPRESSION_TYPE_CONFIG, BrokerCompressionType.ZSTD.name)
-    val partitionToLeader = createTopic(topic, topicConfig =  topicConfig)
+    val partitionToLeader = createTopic(topic, topicConfig = topicConfig)
     val leader = partitionToLeader(partition)
     val memoryRecords = MemoryRecords.withRecords(Compression.zstd().build(),
       new SimpleRecord(System.currentTimeMillis(), "key".getBytes, "value".getBytes))
@@ -283,6 +286,48 @@ class ProduceRequestTest extends BaseRequestTest {
     assertEquals(-1, partitionProduceResponse1.logAppendTimeMs)
   }
 
+  /**
+   * See `ApiKeys.PRODUCE_OLDEST_VERSION` for the details of why we need special handling for produce request v0-v2 (inclusive).
+   */
+  @Test
+  def testProduceRequestV0ToV2IsRejectedByBroker(): Unit = {
+    val topic = "topic"
+    val partition = 0
+    val partitionToLeader = createTopic(topic)
+    val leader = partitionToLeader(partition)
+    val memoryRecords = MemoryRecords.withRecords(Compression.none().build(),
+      new SimpleRecord(System.currentTimeMillis(), "key".getBytes, "value".getBytes))
+    val produceRequestData = new ProduceRequestData()
+      .setTopicData(new ProduceRequestData.TopicProduceDataCollection(Collections.singletonList(
+        new ProduceRequestData.TopicProduceData()
+          .setName("topic").setPartitionData(Collections.singletonList(
+            new ProduceRequestData.PartitionProduceData()
+              .setIndex(partition)
+              .setRecords(memoryRecords))))
+        .iterator))
+      .setAcks((-1).toShort)
+      .setTimeoutMs(3000)
+      .setTransactionalId(null)
+
+    for (i <- 0 to 2) {
+      val version = i.toShort
+      // Broker disconnects when it receives a request with an unsupported version
+      assertThrows(classOf[EOFException], () => sendProduceRequestData(leader, version, produceRequestData))
+    }
+  }
+
+  // This method avoids some of the version validation performed by the wrapper request classes, which is useful
+  // to be able to send produce requests with versions 0-2
+  private def sendProduceRequestData(leaderId: Int, version: Short, request: ProduceRequestData): ProduceResponse = {
+    val socket = connect(brokerSocketServer(leaderId), listenerName)
+    try {
+      val header = nextRequestHeader(ApiKeys.PRODUCE, version)
+      val serializedBytes = Utils.toArray(RequestUtils.serialize(header.data, header.headerVersion, request, version))
+      sendRequest(socket, serializedBytes)
+      receive[ProduceResponse](socket, ApiKeys.PRODUCE, version)
+    } finally socket.close()
+  }
+
   private def sendProduceRequest(leaderId: Int, request: ProduceRequest): ProduceResponse = {
     connectAndReceive[ProduceResponse](request, destination = brokerSocketServer(leaderId))
   }