KAFKA-18465: Remove MetadataVersions older than 3.0-IV1 (apache#18468)

Apache Kafka 4.0 will only support KRaft and 3.0-IV1 is the minimum version supported by KRaft. So, we can assume that Apache Kafka 4.0 will only communicate with brokers that are 3.0-IV1 or newer.

Note that KRaft was only marked as production-ready in 3.3, so we could go further and set the baseline to 3.3. I think we should have that discussion, but it made sense to start with the non controversial parts.

Reviewers: Jun Rao <[email protected]>, Chia-Ping Tsai <[email protected]>, David Jacot <[email protected]>

Author: ijuma

clients/src/main/java/org/apache/kafka/clients/ClientUtils.java

@@ -118,7 +118,7 @@ public static ChannelBuilder createChannelBuilder(AbstractConfig config, Time ti
         SecurityProtocol securityProtocol = SecurityProtocol.forName(config.getString(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
         String clientSaslMechanism = config.getString(SaslConfigs.SASL_MECHANISM);
         return ChannelBuilders.clientChannelBuilder(securityProtocol, JaasContext.Type.CLIENT, config, null,
-                clientSaslMechanism, time, true, logContext);
+                clientSaslMechanism, time, logContext);
     }
 
     static List<InetAddress> resolve(String host, HostResolver hostResolver) throws UnknownHostException {
@@ -275,4 +275,4 @@ public static ClusterResourceListeners configureClusterResourceListeners(List<?>
 
         return clusterResourceListeners;
     }
-}
+}

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

@@ -56,8 +56,6 @@ private ChannelBuilders() { }
      * @param listenerName the listenerName if contextType is SERVER or null otherwise
      * @param clientSaslMechanism SASL mechanism if mode is CLIENT, ignored otherwise
      * @param time the time instance
-     * @param saslHandshakeRequestEnable flag to enable Sasl handshake requests; disabled only for SASL
-     *             inter-broker connections with inter-broker protocol version < 0.10
      * @param logContext the log context instance
      *
      * @return the configured `ChannelBuilder`
@@ -70,7 +68,6 @@ public static ChannelBuilder clientChannelBuilder(
             ListenerName listenerName,
             String clientSaslMechanism,
             Time time,
-            boolean saslHandshakeRequestEnable,
             LogContext logContext) {
 
         if (securityProtocol == SecurityProtocol.SASL_PLAINTEXT || securityProtocol == SecurityProtocol.SASL_SSL) {
@@ -80,7 +77,7 @@ public static ChannelBuilder clientChannelBuilder(
                 throw new IllegalArgumentException("`clientSaslMechanism` must be non-null in client mode if `securityProtocol` is `" + securityProtocol + "`");
         }
         return create(securityProtocol, ConnectionMode.CLIENT, contextType, config, listenerName, false, clientSaslMechanism,
-                saslHandshakeRequestEnable, null, null, time, logContext, null);
+            null, null, time, logContext, null);
     }
 
     /**
@@ -106,8 +103,8 @@ public static ChannelBuilder serverChannelBuilder(ListenerName listenerName,
                                                       LogContext logContext,
                                                       Function<Short, ApiVersionsResponse> apiVersionSupplier) {
         return create(securityProtocol, ConnectionMode.SERVER, JaasContext.Type.SERVER, config, listenerName,
-                isInterBrokerListener, null, true, credentialCache,
-                tokenCache, time, logContext, apiVersionSupplier);
+                isInterBrokerListener, null, credentialCache, tokenCache, time, logContext,
+                apiVersionSupplier);
     }
 
     private static ChannelBuilder create(SecurityProtocol securityProtocol,
@@ -117,7 +114,6 @@ private static ChannelBuilder create(SecurityProtocol securityProtocol,
                                          ListenerName listenerName,
                                          boolean isInterBrokerListener,
                                          String clientSaslMechanism,
-                                         boolean saslHandshakeRequestEnable,
                                          CredentialCache credentialCache,
                                          DelegationTokenCache tokenCache,
                                          Time time,
@@ -175,7 +171,6 @@ private static ChannelBuilder create(SecurityProtocol securityProtocol,
                         listenerName,
                         isInterBrokerListener,
                         clientSaslMechanism,
-                        saslHandshakeRequestEnable,
                         credentialCache,
                         tokenCache,
                         sslClientAuthOverride,

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

@@ -85,7 +85,6 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
     private final String clientSaslMechanism;
     private final ConnectionMode connectionMode;
     private final Map<String, JaasContext> jaasContexts;
-    private final boolean handshakeRequestEnable;
     private final CredentialCache credentialCache;
     private final DelegationTokenCache tokenCache;
     private final Map<String, LoginManager> loginManagers;
@@ -108,7 +107,6 @@ public SaslChannelBuilder(ConnectionMode connectionMode,
                               ListenerName listenerName,
                               boolean isInterBrokerListener,
                               String clientSaslMechanism,
-                              boolean handshakeRequestEnable,
                               CredentialCache credentialCache,
                               DelegationTokenCache tokenCache,
                               String sslClientAuthOverride,
@@ -122,7 +120,6 @@ public SaslChannelBuilder(ConnectionMode connectionMode,
         this.securityProtocol = securityProtocol;
         this.listenerName = listenerName;
         this.isInterBrokerListener = isInterBrokerListener;
-        this.handshakeRequestEnable = handshakeRequestEnable;
         this.clientSaslMechanism = clientSaslMechanism;
         this.credentialCache = credentialCache;
         this.tokenCache = tokenCache;
@@ -295,7 +292,7 @@ protected SaslClientAuthenticator buildClientAuthenticator(Map<String, ?> config
                                                                String servicePrincipal,
                                                                TransportLayer transportLayer, Subject subject) {
         return new SaslClientAuthenticator(configs, callbackHandler, id, subject, servicePrincipal,
-                serverHost, clientSaslMechanism, handshakeRequestEnable, transportLayer, time, logContext);
+                serverHost, clientSaslMechanism, transportLayer, time, logContext);
     }
 
     // Package private for testing

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

@@ -68,14 +68,12 @@ public static class Builder extends AbstractRequest.Builder<AlterPartitionReques
          * @param data The data to be sent. Note that because the version of the
          *             request is not known at this time, it is expected that all
          *             topics have a topic id and a topic name set.
-         * @param canUseTopicIds True if version 2 and above can be used.
          */
-        public Builder(AlterPartitionRequestData data, boolean canUseTopicIds) {
+        public Builder(AlterPartitionRequestData data) {
             super(
                 ApiKeys.ALTER_PARTITION,
                 ApiKeys.ALTER_PARTITION.oldestVersion(),
-                // Version 1 is the maximum version that can be used without topic ids.
-                canUseTopicIds ? ApiKeys.ALTER_PARTITION.latestVersion() : 1
+                ApiKeys.ALTER_PARTITION.latestVersion()
             );
             this.data = data;
         }

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

@@ -31,7 +31,6 @@
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ByteBufferAccessor;
 import org.apache.kafka.common.protocol.Errors;
-import org.apache.kafka.common.record.RecordVersion;
 
 import java.nio.ByteBuffer;
 import java.util.Map;
@@ -172,35 +171,30 @@ public static ApiVersionsResponse parse(ByteBuffer buffer, short version) {
     }
 
     public static ApiVersionCollection controllerApiVersions(
-        RecordVersion minRecordVersion,
         NodeApiVersions controllerApiVersions,
         ListenerType listenerType,
         boolean enableUnstableLastVersion,
         boolean clientTelemetryEnabled
     ) {
         return intersectForwardableApis(
             listenerType,
-            minRecordVersion,
             controllerApiVersions.allSupportedApiVersions(),
             enableUnstableLastVersion,
             clientTelemetryEnabled);
     }
 
     public static ApiVersionCollection brokerApiVersions(
-        RecordVersion minRecordVersion,
         ListenerType listenerType,
         boolean enableUnstableLastVersion,
         boolean clientTelemetryEnabled
     ) {
         return filterApis(
-            minRecordVersion,
             listenerType,
             enableUnstableLastVersion,
             clientTelemetryEnabled);
     }
 
     public static ApiVersionCollection filterApis(
-        RecordVersion minRecordVersion,
         ApiMessageType.ListenerType listenerType,
         boolean enableUnstableLastVersion,
         boolean clientTelemetryEnabled
@@ -210,10 +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;
-
-            if (apiKey.minRequiredInterBrokerMagic <= minRecordVersion.value) {
-                apiKey.toApiVersion(enableUnstableLastVersion).ifPresent(apiKeys::add);
-            }
+            apiKey.toApiVersion(enableUnstableLastVersion).ifPresent(apiKeys::add);
         }
         return apiKeys;
     }
@@ -234,50 +225,46 @@ public static ApiVersionCollection collectApis(
      * known range and that of another set.
      *
      * @param listenerType the listener type which constrains the set of exposed APIs
-     * @param minRecordVersion min inter broker magic
      * @param activeControllerApiVersions controller ApiVersions
      * @param enableUnstableLastVersion whether unstable versions should be advertised or not
      * @param clientTelemetryEnabled whether client telemetry is enabled or not
      * @return commonly agreed ApiVersion collection
      */
     public static ApiVersionCollection intersectForwardableApis(
         final ApiMessageType.ListenerType listenerType,
-        final RecordVersion minRecordVersion,
         final Map<ApiKeys, ApiVersion> activeControllerApiVersions,
         boolean enableUnstableLastVersion,
         boolean clientTelemetryEnabled
     ) {
         ApiVersionCollection apiKeys = new ApiVersionCollection();
         for (ApiKeys apiKey : ApiKeys.apisForListener(listenerType)) {
-            if (apiKey.minRequiredInterBrokerMagic <= minRecordVersion.value) {
-                final Optional<ApiVersion> brokerApiVersion = apiKey.toApiVersion(enableUnstableLastVersion);
-                if (brokerApiVersion.isEmpty()) {
-                    // Broker does not support this API key.
-                    continue;
-                }
+            final Optional<ApiVersion> brokerApiVersion = apiKey.toApiVersion(enableUnstableLastVersion);
+            if (brokerApiVersion.isEmpty()) {
+                // Broker does not support this API key.
+                continue;
+            }
 
-                // Skip telemetry APIs if client telemetry is disabled.
-                if ((apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey == ApiKeys.PUSH_TELEMETRY) && !clientTelemetryEnabled)
-                    continue;
+            // Skip telemetry APIs if client telemetry is disabled.
+            if ((apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey == ApiKeys.PUSH_TELEMETRY) && !clientTelemetryEnabled)
+                continue;
 
-                final ApiVersion finalApiVersion;
-                if (!apiKey.forwardable) {
-                    finalApiVersion = brokerApiVersion.get();
+            final ApiVersion finalApiVersion;
+            if (!apiKey.forwardable) {
+                finalApiVersion = brokerApiVersion.get();
+            } else {
+                Optional<ApiVersion> intersectVersion = intersect(
+                    brokerApiVersion.get(),
+                    activeControllerApiVersions.getOrDefault(apiKey, null)
+                );
+                if (intersectVersion.isPresent()) {
+                    finalApiVersion = intersectVersion.get();
                 } else {
-                    Optional<ApiVersion> intersectVersion = intersect(
-                        brokerApiVersion.get(),
-                        activeControllerApiVersions.getOrDefault(apiKey, null)
-                    );
-                    if (intersectVersion.isPresent()) {
-                        finalApiVersion = intersectVersion.get();
-                    } else {
-                        // Controller doesn't support this API key, or there is no intersection.
-                        continue;
-                    }
+                    // Controller doesn't support this API key, or there is no intersection.
+                    continue;
                 }
-
-                apiKeys.add(finalApiVersion.duplicate());
             }
+
+            apiKeys.add(finalApiVersion.duplicate());
         }
         return apiKeys;
     }

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

@@ -65,11 +65,12 @@ public static Builder forConsumer(OffsetForLeaderTopicCollection epochsByPartiti
             return new Builder((short) 3, ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion(), data);
         }
 
-        public static Builder forFollower(short version, OffsetForLeaderTopicCollection epochsByPartition, int replicaId) {
+        public static Builder forFollower(OffsetForLeaderTopicCollection epochsByPartition, int replicaId) {
             OffsetForLeaderEpochRequestData data = new OffsetForLeaderEpochRequestData();
             data.setReplicaId(replicaId);
             data.setTopics(epochsByPartition);
-            return new Builder(version, version, data);
+            // If we introduce new versions, we should gate them behind the appropriate metadata version
+            return new Builder((short) 4, (short) 4, data);
         }
 
         @Override

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

@@ -110,8 +110,8 @@ public Builder(WriteTxnMarkersRequestData data) {
             this.data = data;
         }
 
-        public Builder(short version, final List<TxnMarkerEntry> markers) {
-            super(ApiKeys.WRITE_TXN_MARKERS, version);
+        public Builder(final List<TxnMarkerEntry> markers) {
+            super(ApiKeys.WRITE_TXN_MARKERS, (short) 1); // if we add new versions, gate them behind metadata version
             List<WritableTxnMarker> dataMarkers = new ArrayList<>();
             for (TxnMarkerEntry marker : markers) {
                 final Map<String, WritableTxnMarkerTopic> topicMap = new HashMap<>();

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

@@ -177,7 +177,6 @@ public SaslClientAuthenticator(Map<String, ?> configs,
                                    String servicePrincipal,
                                    String host,
                                    String mechanism,
-                                   boolean handshakeRequestEnable,
                                    TransportLayer transportLayer,
                                    Time time,
                                    LogContext logContext) {
@@ -196,7 +195,7 @@ public SaslClientAuthenticator(Map<String, ?> configs,
         this.reauthInfo = new ReauthInfo();
 
         try {
-            setSaslState(handshakeRequestEnable ? SaslState.SEND_APIVERSIONS_REQUEST : SaslState.INITIAL);
+            setSaslState(SaslState.SEND_APIVERSIONS_REQUEST);
 
             // determine client principal from subject for Kerberos to use as authorization id for the SaslClient.
             // For other mechanisms, the authenticated principal (username for PLAIN and SCRAM) is used as

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

@@ -161,7 +161,6 @@
 import org.apache.kafka.common.quota.ClientQuotaEntity;
 import org.apache.kafka.common.quota.ClientQuotaFilter;
 import org.apache.kafka.common.quota.ClientQuotaFilterComponent;
-import org.apache.kafka.common.record.RecordVersion;
 import org.apache.kafka.common.requests.AddRaftVoterRequest;
 import org.apache.kafka.common.requests.AddRaftVoterResponse;
 import org.apache.kafka.common.requests.AlterClientQuotasResponse;
@@ -777,7 +776,7 @@ private static ApiVersionsResponse prepareApiVersionsResponseForDescribeFeatures
         if (error == Errors.NONE) {
             return new ApiVersionsResponse.Builder().
                 setApiVersions(ApiVersionsResponse.filterApis(
-                    RecordVersion.current(), ApiMessageType.ListenerType.ZK_BROKER, false, false)).
+                    ApiMessageType.ListenerType.ZK_BROKER, false, false)).
                 setSupportedFeatures(
                     convertSupportedFeaturesMap(defaultFeatureMetadata().supportedFeatures())).
                 setFinalizedFeatures(

clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java

@@ -166,7 +166,7 @@ public void testClientChannelBuilderWithBrokerConfigs() throws Exception {
     private SaslChannelBuilder createGssapiChannelBuilder(Map<String, JaasContext> jaasContexts, GSSManager gssManager) {
         SaslChannelBuilder channelBuilder = new SaslChannelBuilder(ConnectionMode.SERVER, jaasContexts,
             SecurityProtocol.SASL_PLAINTEXT, new ListenerName("GSSAPI"), false, "GSSAPI",
-            true, null, null, null, Time.SYSTEM, new LogContext(), defaultApiVersionsSupplier()) {
+            null, null, null, Time.SYSTEM, new LogContext(), defaultApiVersionsSupplier()) {
 
             @Override
             protected GSSManager gssManager() {
@@ -205,7 +205,7 @@ private SaslChannelBuilder createChannelBuilder(SecurityProtocol securityProtoco
         JaasContext jaasContext = new JaasContext("jaasContext", JaasContext.Type.SERVER, jaasConfig, null);
         Map<String, JaasContext> jaasContexts = Collections.singletonMap(saslMechanism, jaasContext);
         return new SaslChannelBuilder(ConnectionMode.CLIENT, jaasContexts, securityProtocol, new ListenerName(saslMechanism),
-                false, saslMechanism, true, null,
+                false, saslMechanism, null,
                 null, null, Time.SYSTEM, new LogContext(), defaultApiVersionsSupplier());
     }
 

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

@@ -61,7 +61,7 @@ public void testBuildAlterPartitionRequest(short version) {
 
         request.topics().add(topicData);
 
-        AlterPartitionRequest.Builder builder = new AlterPartitionRequest.Builder(request, version > 1);
+        AlterPartitionRequest.Builder builder = new AlterPartitionRequest.Builder(request);
         AlterPartitionRequest alterPartitionRequest = builder.build(version);
         assertEquals(1, alterPartitionRequest.data().topics().size());
         assertEquals(1, alterPartitionRequest.data().topics().get(0).partitions().size());