KAFKA-17011: Fix a bug preventing features from supporting v0

clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java

@@ -42,7 +42,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
-import java.util.function.Supplier;
+import java.util.function.Function;
 
 public class ChannelBuilders {
     private static final Logger log = LoggerFactory.getLogger(ChannelBuilders.class);
@@ -104,7 +104,7 @@ public static ChannelBuilder serverChannelBuilder(ListenerName listenerName,
                                                       DelegationTokenCache tokenCache,
                                                       Time time,
                                                       LogContext logContext,
-                                                      Supplier<ApiVersionsResponse> apiVersionSupplier) {
+                                                      Function<Short, ApiVersionsResponse> apiVersionSupplier) {
         return create(securityProtocol, Mode.SERVER, JaasContext.Type.SERVER, config, listenerName,
                 isInterBrokerListener, null, true, credentialCache,
                 tokenCache, time, logContext, apiVersionSupplier);
@@ -122,7 +122,7 @@ private static ChannelBuilder create(SecurityProtocol securityProtocol,
                                          DelegationTokenCache tokenCache,
                                          Time time,
                                          LogContext logContext,
-                                         Supplier<ApiVersionsResponse> apiVersionSupplier) {
+                                         Function<Short, ApiVersionsResponse> apiVersionSupplier) {
         Map<String, Object> configs = channelBuilderConfigs(config, listenerName);
 
         ChannelBuilder channelBuilder;

clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java

@@ -69,6 +69,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.function.Supplier;
 
 import javax.security.auth.Subject;
@@ -88,7 +89,7 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
     private final DelegationTokenCache tokenCache;
     private final Map<String, LoginManager> loginManagers;
     private final Map<String, Subject> subjects;
-    private final Supplier<ApiVersionsResponse> apiVersionSupplier;
+    private final Function<Short, ApiVersionsResponse> apiVersionSupplier;
     private final String sslClientAuthOverride;
     private final Map<String, AuthenticateCallbackHandler> saslCallbackHandlers;
     private final Map<String, Long> connectionsMaxReauthMsByMechanism;
@@ -112,7 +113,7 @@ public SaslChannelBuilder(Mode mode,
                               String sslClientAuthOverride,
                               Time time,
                               LogContext logContext,
-                              Supplier<ApiVersionsResponse> apiVersionSupplier) {
+                              Function<Short, ApiVersionsResponse> apiVersionSupplier) {
         this.mode = mode;
         this.jaasContexts = jaasContexts;
         this.loginManagers = new HashMap<>(jaasContexts.size());

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

@@ -35,6 +35,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 
@@ -49,6 +50,71 @@ public class ApiVersionsResponse extends AbstractResponse {
 
     private final ApiVersionsResponseData data;
 
+    public static class Builder {
+        private Errors error = Errors.NONE;
+        private int throttleTimeMs = 0;
+        private ApiVersionCollection apiVersions = null;
+        private Features<SupportedVersionRange> supportedFeatures = null;
+        private Map<String, Short> finalizedFeatures = null;
+        private long finalizedFeaturesEpoch = 0;
+        private boolean zkMigrationEnabled = false;
+        private boolean suppressFeatureLevel0 = false;
+
+        public Builder setError(Errors error) {
+            this.error = error;
+            return this;
+        }
+
+        public Builder setThrottleTimeMs(int throttleTimeMs) {
+            this.throttleTimeMs = throttleTimeMs;
+            return this;
+        }
+
+        public Builder setApiVersions(ApiVersionCollection apiVersions) {
+            this.apiVersions = apiVersions;
+            return this;
+        }
+
+        public Builder setSupportedFeatures(Features<SupportedVersionRange> supportedFeatures) {
+            this.supportedFeatures = supportedFeatures;
+            return this;
+        }
+
+        public Builder setFinalizedFeatures(Map<String, Short> finalizedFeatures) {
+            this.finalizedFeatures = finalizedFeatures;
+            return this;
+        }
+
+        public Builder setFinalizedFeaturesEpoch(long finalizedFeaturesEpoch) {
+            this.finalizedFeaturesEpoch = finalizedFeaturesEpoch;
+            return this;
+        }
+
+        public Builder setZkMigrationEnabled(boolean zkMigrationEnabled) {
+            this.zkMigrationEnabled = zkMigrationEnabled;
+            return this;
+        }
+
+        public Builder setSuppressFeatureLevel0(boolean suppressFeatureLevel0) {
+            this.suppressFeatureLevel0 = suppressFeatureLevel0;
+            return this;
+        }
+
+        public ApiVersionsResponse build() {
+            final ApiVersionsResponseData data = new ApiVersionsResponseData();
+            data.setErrorCode(error.code());
+            data.setApiKeys(Objects.requireNonNull(apiVersions));
+            data.setThrottleTimeMs(throttleTimeMs);
+            data.setSupportedFeatures(
+                maybeFilterSupportedFeatureKeys(Objects.requireNonNull(supportedFeatures), suppressFeatureLevel0));
+            data.setFinalizedFeatures(
+                createFinalizedFeatureKeys(Objects.requireNonNull(finalizedFeatures)));
+            data.setFinalizedFeaturesEpoch(finalizedFeaturesEpoch);
+            data.setZkMigrationReady(zkMigrationEnabled);
+            return new ApiVersionsResponse(data);
+        }
+    }
+
     public ApiVersionsResponse(ApiVersionsResponseData data) {
         super(ApiKeys.API_VERSIONS);
         this.data = data;
@@ -105,65 +171,32 @@ public static ApiVersionsResponse parse(ByteBuffer buffer, short version) {
         }
     }
 
-    public static ApiVersionsResponse createApiVersionsResponse(
-        int throttleTimeMs,
+    public static ApiVersionCollection controllerApiVersions(
         RecordVersion minRecordVersion,
-        Features<SupportedVersionRange> latestSupportedFeatures,
-        Map<String, Short> finalizedFeatures,
-        long finalizedFeaturesEpoch,
         NodeApiVersions controllerApiVersions,
         ListenerType listenerType,
         boolean enableUnstableLastVersion,
-        boolean zkMigrationEnabled,
         boolean clientTelemetryEnabled
     ) {
-        ApiVersionCollection apiKeys;
-        if (controllerApiVersions != null) {
-            apiKeys = intersectForwardableApis(
-                listenerType,
-                minRecordVersion,
-                controllerApiVersions.allSupportedApiVersions(),
-                enableUnstableLastVersion,
-                clientTelemetryEnabled
-            );
-        } else {
-            apiKeys = filterApis(
-                minRecordVersion,
-                listenerType,
-                enableUnstableLastVersion,
-                clientTelemetryEnabled
-            );
-        }
-
-        return createApiVersionsResponse(
-            throttleTimeMs,
-            apiKeys,
-            latestSupportedFeatures,
-            finalizedFeatures,
-            finalizedFeaturesEpoch,
-            zkMigrationEnabled
-        );
+        return intersectForwardableApis(
+            listenerType,
+            minRecordVersion,
+            controllerApiVersions.allSupportedApiVersions(),
+            enableUnstableLastVersion,
+            clientTelemetryEnabled);
     }
 
-    public static ApiVersionsResponse createApiVersionsResponse(
-        int throttleTimeMs,
-        ApiVersionCollection apiVersions,
-        Features<SupportedVersionRange> latestSupportedFeatures,
-        Map<String, Short> finalizedFeatures,
-        long finalizedFeaturesEpoch,
-        boolean zkMigrationEnabled
+    public static ApiVersionCollection brokerApiVersions(
+        RecordVersion minRecordVersion,
+        ListenerType listenerType,
+        boolean enableUnstableLastVersion,
+        boolean clientTelemetryEnabled
     ) {
-        return new ApiVersionsResponse(
-            createApiVersionsResponseData(
-                throttleTimeMs,
-                Errors.NONE,
-                apiVersions,
-                latestSupportedFeatures,
-                finalizedFeatures,
-                finalizedFeaturesEpoch,
-                zkMigrationEnabled
-            )
-        );
+        return filterApis(
+            minRecordVersion,
+            listenerType,
+            enableUnstableLastVersion,
+            clientTelemetryEnabled);
     }
 
     public static ApiVersionCollection filterApis(
@@ -249,37 +282,24 @@ public static ApiVersionCollection intersectForwardableApis(
         return apiKeys;
     }
 
-    private static ApiVersionsResponseData createApiVersionsResponseData(
-        final int throttleTimeMs,
-        final Errors error,
-        final ApiVersionCollection apiKeys,
-        final Features<SupportedVersionRange> latestSupportedFeatures,
-        final Map<String, Short> finalizedFeatures,
-        final long finalizedFeaturesEpoch,
-        final boolean zkMigrationEnabled
+    private static SupportedFeatureKeyCollection maybeFilterSupportedFeatureKeys(
+        Features<SupportedVersionRange> latestSupportedFeatures,
+        boolean suppressV0
     ) {
-        final ApiVersionsResponseData data = new ApiVersionsResponseData();
-        data.setThrottleTimeMs(throttleTimeMs);
-        data.setErrorCode(error.code());
-        data.setApiKeys(apiKeys);
-        data.setSupportedFeatures(createSupportedFeatureKeys(latestSupportedFeatures));
-        data.setFinalizedFeatures(createFinalizedFeatureKeys(finalizedFeatures));
-        data.setFinalizedFeaturesEpoch(finalizedFeaturesEpoch);
-        data.setZkMigrationReady(zkMigrationEnabled);
-
-        return data;
-    }
-
-    private static SupportedFeatureKeyCollection createSupportedFeatureKeys(
-        Features<SupportedVersionRange> latestSupportedFeatures) {
         SupportedFeatureKeyCollection converted = new SupportedFeatureKeyCollection();
         for (Map.Entry<String, SupportedVersionRange> feature : latestSupportedFeatures.features().entrySet()) {
-            final SupportedFeatureKey key = new SupportedFeatureKey();
             final SupportedVersionRange versionRange = feature.getValue();
-            key.setName(feature.getKey());
-            key.setMinVersion(versionRange.min());
-            key.setMaxVersion(versionRange.max());
-            converted.add(key);
+            // Some older clients will have deserialization problems if a feature's
+            // minimum supported level is 0. Therefore, when preparing ApiVersionResponse
+            // at versions less than 4, we must leave those features out. See KAFKA-17011
+            // for details.
+            if (!(suppressV0 && versionRange.min() == 0)) {
+                final SupportedFeatureKey key = new SupportedFeatureKey();
+                key.setName(feature.getKey());
+                key.setMinVersion(versionRange.min());
+                key.setMaxVersion(versionRange.max());
+                converted.add(key);
+            }
         }
 
         return converted;

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

@@ -23,6 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
+import java.util.Iterator;
 
 public class BrokerRegistrationRequest extends AbstractRequest {
 
@@ -45,7 +46,21 @@ public short oldestAllowedVersion() {
 
         @Override
         public BrokerRegistrationRequest build(short version) {
-            return new BrokerRegistrationRequest(data, version);
+            if (version < 4) {
+                // Workaround for KAFKA-17011: for BrokerRegistrationRequest versions older than 4,
+                // exclude support version ranges that begin with 0.
+                BrokerRegistrationRequestData newData = data.duplicate();
+                for (Iterator<BrokerRegistrationRequestData.Feature> iter = newData.features().iterator();
+                     iter.hasNext(); ) {
+                    BrokerRegistrationRequestData.Feature feature = iter.next();
+                    if (feature.minSupportedVersion() == 0) {
+                        iter.remove();
+                    }
+                }
+                return new BrokerRegistrationRequest(newData, version);
+            } else {
+                return new BrokerRegistrationRequest(data, version);
+            }
         }
 
         @Override

clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java

@@ -82,7 +82,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.function.Supplier;
+import java.util.function.Function;
 
 import javax.net.ssl.SSLSession;
 import javax.security.auth.Subject;
@@ -129,7 +129,7 @@ private enum SaslState {
     private final Time time;
     private final ReauthInfo reauthInfo;
     private final ChannelMetadataRegistry metadataRegistry;
-    private final Supplier<ApiVersionsResponse> apiVersionSupplier;
+    private final Function<Short, ApiVersionsResponse> apiVersionSupplier;
 
     // Current SASL state
     private SaslState saslState = SaslState.INITIAL_REQUEST;
@@ -159,7 +159,7 @@ public SaslServerAuthenticator(Map<String, ?> configs,
                                    Map<String, Long> connectionsMaxReauthMsByMechanism,
                                    ChannelMetadataRegistry metadataRegistry,
                                    Time time,
-                                   Supplier<ApiVersionsResponse> apiVersionSupplier) {
+                                   Function<Short, ApiVersionsResponse> apiVersionSupplier) {
         this.callbackHandlers = callbackHandlers;
         this.connectionId = connectionId;
         this.subjects = subjects;
@@ -596,7 +596,7 @@ else if (!apiVersionsRequest.isValid())
         else {
             metadataRegistry.registerClientInformation(new ClientInformation(apiVersionsRequest.data().clientSoftwareName(),
                 apiVersionsRequest.data().clientSoftwareVersion()));
-            sendKafkaResponse(context, apiVersionSupplier.get());
+            sendKafkaResponse(context, apiVersionSupplier.apply(apiVersionsRequest.version()));
             setSaslState(SaslState.HANDSHAKE_REQUEST);
         }
     }

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

@@ -21,7 +21,9 @@
   // Versions 0 through 2 of ApiVersionsRequest are the same.
   //
   // Version 3 is the first flexible version and adds ClientSoftwareName and ClientSoftwareVersion.
-  "validVersions": "0-3",
+  //
+  // Version 4 fixes KAFKA-17011, which blocked SupportedFeatures.MinVersion in the response from being 0.
+  "validVersions": "0-4",
   "flexibleVersions": "3+",
   "fields": [
     { "name": "ClientSoftwareName", "type": "string", "versions": "3+",

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

@@ -27,7 +27,9 @@
   //
   // Starting from Apache Kafka 2.4 (KIP-511), ApiKeys field is populated with the supported
   // versions of the ApiVersionsRequest when an UNSUPPORTED_VERSION error is returned.
-  "validVersions": "0-3",
+  //
+  // Version 4 fixes KAFKA-17011, which blocked SupportedFeatures.MinVersion from being 0.
+  "validVersions": "0-4",
   "flexibleVersions": "3+",
   "fields": [
     { "name": "ErrorCode", "type": "int16", "versions": "0+",
@@ -45,7 +47,7 @@
       "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
     { "name":  "SupportedFeatures", "type": "[]SupportedFeatureKey", "ignorable": true,
       "versions":  "3+", "tag": 0, "taggedVersions": "3+",
-      "about": "Features supported by the broker.",
+      "about": "Features supported by the broker. Note: in v0-v3, features with MinSupportedVersion = 0 must be left out.",
       "fields":  [
         { "name": "Name", "type": "string", "versions": "3+", "mapKey": true,
           "about": "The name of the feature." },

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

@@ -18,12 +18,15 @@
 // Version 2 adds LogDirs for KIP-858
 
 // Version 3 adds the PreviousBrokerEpoch for the KIP-966
+
+// Version 4 fixes KAFKA-17011, which blocked SupportedFeatures.MinVersion in the response from being 0.
+
 {
   "apiKey":62,
   "type": "request",
   "listeners": ["controller"],
   "name": "BrokerRegistrationRequest",
-  "validVersions": "0-3",
+  "validVersions": "0-4",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
@@ -45,7 +48,7 @@
     ]
     },
     { "name": "Features", "type": "[]Feature",
-      "about": "The features on this broker", "versions": "0+", "fields": [
+      "about": "The features on this broker. Note: in v0-v3, features with MinSupportedVersion = 0 must be left out.", "versions": "0+", "fields": [
       { "name": "Name", "type": "string", "versions": "0+", "mapKey": true,
         "about": "The feature name." },
       { "name": "MinSupportedVersion", "type": "int16", "versions": "0+",