Merge branch 'trunk' into KAFKA-16907

Author: frankvicky

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

@@ -149,6 +149,11 @@ 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,
+    // version `0` has to be included in the api versions response (see KAFKA-18659). In order to achieve that,
+    // we adjust `toApiVersion` to return `0` for the min version of `produce` in the broker listener.
+    public static final short PRODUCE_API_VERSIONS_RESPONSE_MIN_VERSION = 0;
+
     /** the permanent and immutable id of an API - this can't change ever */
     public final short id;
 
@@ -264,8 +269,30 @@ public boolean hasValidVersion() {
         return oldestVersion() <= latestVersion();
     }
 
+    /**
+     * To workaround a critical bug in librdkafka, the api versions response is inconsistent with the actual versions
+     * supported by `produce` - this method handles that. It should be called in the context of the api response protocol
+     * handling.
+     *
+     * It should not be used by code generating protocol documentation - we keep that consistent with the actual versions
+     * supported by `produce`.
+     *
+     * See `PRODUCE_API_VERSIONS_RESPONSE_MIN_VERSION` for details.
+     */
+    public Optional<ApiVersionsResponseData.ApiVersion> toApiVersionForApiResponse(boolean enableUnstableLastVersion,
+                                                                                   ApiMessageType.ListenerType listenerType) {
+        return toApiVersion(enableUnstableLastVersion, Optional.of(listenerType));
+    }
+
     public Optional<ApiVersionsResponseData.ApiVersion> toApiVersion(boolean enableUnstableLastVersion) {
-        short oldestVersion = oldestVersion();
+        return toApiVersion(enableUnstableLastVersion, Optional.empty());
+    }
+
+    private Optional<ApiVersionsResponseData.ApiVersion> toApiVersion(boolean enableUnstableLastVersion,
+                                                                     Optional<ApiMessageType.ListenerType> listenerType) {
+        // see `PRODUCE_API_VERSIONS_RESPONSE_MIN_VERSION` for details on why we do this
+        short oldestVersion = (this == PRODUCE && listenerType.map(l -> l == ApiMessageType.ListenerType.BROKER).orElse(false)) ?
+            PRODUCE_API_VERSIONS_RESPONSE_MIN_VERSION : oldestVersion();
         short latestVersion = latestVersion(enableUnstableLastVersion);
 
         // API is entirely disabled if latestStableVersion is smaller than oldestVersion.
@@ -299,7 +326,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 +368,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,18 +204,19 @@ 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.toApiVersionForApiResponse(enableUnstableLastVersion, listenerType).ifPresent(apiKeys::add);
         }
         return apiKeys;
     }
 
     public static ApiVersionCollection collectApis(
+        ApiMessageType.ListenerType listenerType,
         Set<ApiKeys> apiKeys,
         boolean enableUnstableLastVersion
     ) {
         ApiVersionCollection res = new ApiVersionCollection();
         for (ApiKeys apiKey : apiKeys) {
-            apiKey.toApiVersion(enableUnstableLastVersion).ifPresent(res::add);
+            apiKey.toApiVersionForApiResponse(enableUnstableLastVersion, listenerType).ifPresent(res::add);
         }
         return res;
     }
@@ -238,7 +239,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.toApiVersionForApiResponse(enableUnstableLastVersion, 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), but are rejected otherwise.
+  // See `ApiKeys.PRODUCE_API_VERSIONS_RESPONSE_MIN_VERSION` for more details.
   //
   // Version 1 and 2 are the same as version 0.
   //

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), but are rejected otherwise.
+  // See `ApiKeys.PRODUCE_API_VERSIONS_RESPONSE_MIN_VERSION` for more details.
   //
   // Version 1 added the throttle time.
   // Version 2 added the log append time.

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

@@ -58,19 +58,25 @@ 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(ApiKeys.PRODUCE_API_VERSIONS_RESPONSE_MIN_VERSION, 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.
+            // Check if versions less than min version are indeed set as null, i.e., removed.
             for (int i = 0; i < version.minVersion(); ++i) {
                 assertNull(key.messageType.requestSchemas()[i],
                     "Request version " + i + " for API " + version.apiKey() + " must be null");
                 assertNull(key.messageType.responseSchemas()[i],
                     "Response version " + i + " for API " + version.apiKey() + " must be null");
             }
 
+            // The min version returned in ApiResponse for Produce is not the actual min version, so adjust it
+            var minVersion = (key == ApiKeys.PRODUCE && scope == ListenerType.BROKER) ?
+                ApiKeys.PRODUCE.oldestVersion() : version.minVersion();
             // Check if versions between min and max versions are non null, i.e., valid.
-            for (int i = version.minVersion(); i <= version.maxVersion(); ++i) {
+            for (int i = minVersion; i <= version.maxVersion(); ++i) {
                 assertNotNull(key.messageType.requestSchemas()[i],
                     "Request version " + i + " for API " + version.apiKey() + " must not be null");
                 assertNotNull(key.messageType.responseSchemas()[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,22 @@ 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.oldestVersion(), 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)

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoader.java

@@ -28,23 +28,6 @@
  */
 public interface CoordinatorLoader<U> extends AutoCloseable {
 
-    /**
-     * UnknownRecordTypeException is thrown when the Deserializer encounters
-     * an unknown record type.
-     */
-    class UnknownRecordTypeException extends RuntimeException {
-        private final short unknownType;
-
-        public UnknownRecordTypeException(short unknownType) {
-            super(String.format("Found an unknown record type %d", unknownType));
-            this.unknownType = unknownType;
-        }
-
-        public short unknownType() {
-            return unknownType;
-        }
-    }
-
     /**
      * Object that is returned as part of the future from load(). Holds the partition load time and the
      * end time.

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java

@@ -76,6 +76,11 @@ public CoordinatorRecord deserialize(
 
         final ApiMessage valueMessage = apiMessageValueFor(recordType);
         final short valueVersion = readVersion(valueBuffer, "value");
+
+        if (valueVersion < valueMessage.lowestSupportedVersion() || valueVersion > valueMessage.highestSupportedVersion()) {
+            throw new UnknownRecordVersionException(recordType, valueVersion);
+        }
+
         readMessage(valueMessage, valueBuffer, valueVersion, "value");
 
         return CoordinatorRecord.record(