KAFKA-13935 Fix static usages of IBP in KRaft mode

core/src/main/scala/kafka/server/BrokerFeatures.scala

@@ -73,7 +73,7 @@ object BrokerFeatures extends Logging {
   def createDefault(): BrokerFeatures = {
     new BrokerFeatures(Features.supportedFeatures(
       java.util.Collections.singletonMap(MetadataVersion.FEATURE_NAME,
-        new SupportedVersionRange(MetadataVersion.IBP_3_0_IV0.featureLevel(), MetadataVersion.latest().featureLevel()))))
+        new SupportedVersionRange(MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel(), MetadataVersion.latest().featureLevel()))))
   }
 
   def createEmpty(): BrokerFeatures = {

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -1790,38 +1790,13 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
   // We keep the user-provided String as `MetadataVersion.fromVersionString` can choose a slightly different version (eg if `0.10.0`
   // is passed, `0.10.0-IV0` may be picked)
   val interBrokerProtocolVersionString = getString(KafkaConfig.InterBrokerProtocolVersionProp)
-  val interBrokerProtocolVersion = MetadataVersion.fromVersionString(interBrokerProtocolVersionString)
-
-  val fetchRequestVersion: Short =
-    if (interBrokerProtocolVersion.isAtLeast(IBP_3_1_IV0)) 13
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_2_7_IV1)) 12
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_2_3_IV1)) 11
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_2_1_IV2)) 10
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_2_0_IV1)) 8
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_1_1_IV0)) 7
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_0_11_0_IV1)) 5
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_0_11_0_IV0)) 4
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_0_10_1_IV1)) 3
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_0_10_0_IV0)) 2
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_0_9_0)) 1
-    else 0
-
-  val offsetForLeaderEpochRequestVersion: Short =
-    if (interBrokerProtocolVersion.isAtLeast(IBP_2_8_IV0)) 4
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_2_3_IV1)) 3
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_2_1_IV1)) 2
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_2_0_IV0)) 1
-    else 0
-
-  val listOffsetRequestVersion: Short =
-    if (interBrokerProtocolVersion.isAtLeast(IBP_3_0_IV1)) 7
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_2_8_IV0)) 6
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_2_2_IV1)) 5
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_2_1_IV1)) 4
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_2_0_IV1)) 3
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_0_11_0_IV0)) 2
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_0_10_1_IV2)) 1
-    else 0
+  val interBrokerProtocolVersion = if (processRoles.isEmpty) {
+    MetadataVersion.fromVersionString(interBrokerProtocolVersionString)
+  } else {
+    // In KRaft mode, we pin this value to the minimum KRaft-supported version. This prevents inadvertent usage of
+    // the static IBP config in broker components running in KRaft mode
+    MetadataVersion.MINIMUM_KRAFT_VERSION
+  }
 
   /** ********* Controlled shutdown configuration ***********/
   val controlledShutdownMaxRetries = getInt(KafkaConfig.ControlledShutdownMaxRetriesProp)

core/src/main/scala/kafka/server/KafkaRaftServer.scala

@@ -185,7 +185,7 @@ object KafkaRaftServer {
     val bootstrapMetadata = if (config.originals.containsKey(KafkaConfig.InterBrokerProtocolVersionProp)) {
       BootstrapMetadata.load(Paths.get(config.metadataLogDir), config.interBrokerProtocolVersion)
     } else {
-      BootstrapMetadata.load(Paths.get(config.metadataLogDir), MetadataVersion.IBP_3_0_IV0)
+      BootstrapMetadata.load(Paths.get(config.metadataLogDir), MetadataVersion.MINIMUM_KRAFT_VERSION)
     }
 
     (metaProperties, bootstrapMetadata, offlineDirs.toSeq)

core/src/main/scala/kafka/server/RemoteLeaderEndPoint.scala

@@ -31,6 +31,7 @@ import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.{OffsetFo
 import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData.EpochEndOffset
 import org.apache.kafka.common.protocol.Errors
 import org.apache.kafka.common.requests.{FetchRequest, FetchResponse, ListOffsetsRequest, ListOffsetsResponse, OffsetsForLeaderEpochRequest, OffsetsForLeaderEpochResponse}
+import org.apache.kafka.server.common.MetadataVersion
 import org.apache.kafka.server.common.MetadataVersion.IBP_0_10_1_IV2
 
 import scala.jdk.CollectionConverters._
@@ -46,13 +47,16 @@ import scala.compat.java8.OptionConverters.RichOptionForJava8
  * @param brokerConfig Broker configuration
  * @param replicaManager A ReplicaManager
  * @param quota The quota, used when building a fetch request
+ * @param metadataVersionSupplier A supplier that returns the current MetadataVersion. This can change during
+ *                                runtime in KRaft mode.
  */
 class RemoteLeaderEndPoint(logPrefix: String,
                            blockingSender: BlockingSend,
                            private[server] val fetchSessionHandler: FetchSessionHandler, // visible for testing
                            brokerConfig: KafkaConfig,
                            replicaManager: ReplicaManager,
-                           quota: ReplicaQuota) extends LeaderEndPoint with Logging {
+                           quota: ReplicaQuota,
+                           metadataVersionSupplier: () => MetadataVersion) extends LeaderEndPoint with Logging {
 
   this.logIdent = logPrefix
 
@@ -61,7 +65,7 @@ class RemoteLeaderEndPoint(logPrefix: String,
   private val maxBytes = brokerConfig.replicaFetchResponseMaxBytes
   private val fetchSize = brokerConfig.replicaFetchMaxBytes
 
-  override val isTruncationOnFetchSupported = brokerConfig.interBrokerProtocolVersion.isTruncationOnFetchSupported
+  override def isTruncationOnFetchSupported = metadataVersionSupplier().isTruncationOnFetchSupported
 
   override def initiateClose(): Unit = blockingSender.initiateClose()
 
@@ -106,7 +110,8 @@ class RemoteLeaderEndPoint(logPrefix: String,
           .setPartitionIndex(topicPartition.partition)
           .setCurrentLeaderEpoch(currentLeaderEpoch)
           .setTimestamp(earliestOrLatest)))
-    val requestBuilder = ListOffsetsRequest.Builder.forReplica(brokerConfig.listOffsetRequestVersion, brokerConfig.brokerId)
+    val meteadataVersion = metadataVersionSupplier()
+    val requestBuilder = ListOffsetsRequest.Builder.forReplica(meteadataVersion.listOffsetRequestVersion(), brokerConfig.brokerId)
       .setTargetTimes(Collections.singletonList(topic))
 
     val clientResponse = blockingSender.sendRequest(requestBuilder)
@@ -116,7 +121,7 @@ class RemoteLeaderEndPoint(logPrefix: String,
 
     Errors.forCode(responsePartition.errorCode) match {
       case Errors.NONE =>
-        if (brokerConfig.interBrokerProtocolVersion.isAtLeast(IBP_0_10_1_IV2))
+        if (meteadataVersion.isAtLeast(IBP_0_10_1_IV2))
           responsePartition.offset
         else
           responsePartition.oldStyleOffsets.get(0)
@@ -141,7 +146,7 @@ class RemoteLeaderEndPoint(logPrefix: String,
     }
 
     val epochRequest = OffsetsForLeaderEpochRequest.Builder.forFollower(
-      brokerConfig.offsetForLeaderEpochRequestVersion, topics, brokerConfig.brokerId)
+      metadataVersionSupplier().offsetForLeaderEpochRequestVersion(), topics, brokerConfig.brokerId)
     debug(s"Sending offset for leader epoch request $epochRequest")
 
     try {
@@ -201,7 +206,12 @@ class RemoteLeaderEndPoint(logPrefix: String,
     val fetchRequestOpt = if (fetchData.sessionPartitions.isEmpty && fetchData.toForget.isEmpty) {
       None
     } else {
-      val version: Short = if (brokerConfig.fetchRequestVersion >= 13 && !fetchData.canUseTopicIds) 12 else brokerConfig.fetchRequestVersion
+      val metadataVersion = metadataVersionSupplier()
+      val version: Short = if (metadataVersion.fetchRequestVersion() >= 13 && !fetchData.canUseTopicIds) {
+        12
+      } else {
+        metadataVersion.fetchRequestVersion()
+      }
       val requestBuilder = FetchRequest.Builder
         .forReplica(version, brokerConfig.brokerId, maxWait, minBytes, fetchData.toSend)
         .setMaxBytes(maxBytes)

core/src/main/scala/kafka/server/ReplicaFetcherManager.scala

@@ -21,13 +21,15 @@ import kafka.cluster.BrokerEndPoint
 import org.apache.kafka.clients.FetchSessionHandler
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.utils.{LogContext, Time}
+import org.apache.kafka.server.common.MetadataVersion
 
 class ReplicaFetcherManager(brokerConfig: KafkaConfig,
                             protected val replicaManager: ReplicaManager,
                             metrics: Metrics,
                             time: Time,
                             threadNamePrefix: Option[String] = None,
-                            quotaManager: ReplicationQuotaManager)
+                            quotaManager: ReplicationQuotaManager,
+                            val metadataVersionSupplier: () => MetadataVersion)
       extends AbstractFetcherManager[ReplicaFetcherThread](
         name = "ReplicaFetcherManager on broker " + brokerConfig.brokerId,
         clientId = "Replica",
@@ -41,9 +43,10 @@ class ReplicaFetcherManager(brokerConfig: KafkaConfig,
     val endpoint = new BrokerBlockingSender(sourceBroker, brokerConfig, metrics, time, fetcherId,
       s"broker-${brokerConfig.brokerId}-fetcher-$fetcherId", logContext)
     val fetchSessionHandler = new FetchSessionHandler(logContext, sourceBroker.id)
-    val leader = new RemoteLeaderEndPoint(logContext.logPrefix, endpoint, fetchSessionHandler, brokerConfig, replicaManager, quotaManager)
+    val leader = new RemoteLeaderEndPoint(logContext.logPrefix, endpoint, fetchSessionHandler, brokerConfig,
+      replicaManager, quotaManager, metadataVersionSupplier)
     new ReplicaFetcherThread(threadName, leader, brokerConfig, failedPartitions, replicaManager,
-      quotaManager, logContext.logPrefix)
+      quotaManager, logContext.logPrefix, metadataVersionSupplier)
   }
 
   def shutdown(): Unit = {

core/src/main/scala/kafka/server/ReplicaFetcherThread.scala

@@ -21,14 +21,16 @@ import kafka.log.{LeaderOffsetIncremented, LogAppendInfo}
 import org.apache.kafka.common.record.MemoryRecords
 import org.apache.kafka.common.requests._
 import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.server.common.MetadataVersion
 
 class ReplicaFetcherThread(name: String,
                            leader: LeaderEndPoint,
                            brokerConfig: KafkaConfig,
                            failedPartitions: FailedPartitions,
                            replicaMgr: ReplicaManager,
                            quota: ReplicaQuota,
-                           logPrefix: String)
+                           logPrefix: String,
+                           metadataVersionSupplier: () => MetadataVersion)
   extends AbstractFetcherThread(name = name,
                                 clientId = name,
                                 leader = leader,
@@ -39,7 +41,7 @@ class ReplicaFetcherThread(name: String,
 
   this.logIdent = logPrefix
 
-  override protected val isOffsetForLeaderEpochSupported: Boolean = brokerConfig.interBrokerProtocolVersion.isOffsetForLeaderEpochSupported
+  override protected val isOffsetForLeaderEpochSupported: Boolean = metadataVersionSupplier().isOffsetForLeaderEpochSupported
 
   override protected def latestEpoch(topicPartition: TopicPartition): Option[Int] = {
     replicaMgr.localLogOrException(topicPartition).latestEpoch
@@ -135,7 +137,7 @@ class ReplicaFetcherThread(name: String,
 
   def maybeWarnIfOversizedRecords(records: MemoryRecords, topicPartition: TopicPartition): Unit = {
     // oversized messages don't cause replication to fail from fetch request version 3 (KIP-74)
-    if (brokerConfig.fetchRequestVersion <= 2 && records.sizeInBytes > 0 && records.validBytes <= 0)
+    if (metadataVersionSupplier().fetchRequestVersion() <= 2 && records.sizeInBytes > 0 && records.validBytes <= 0)
       error(s"Replication is failing due to a message that is greater than replica.fetch.max.bytes for partition $topicPartition. " +
         "This generally occurs when the max.message.bytes has been overridden to exceed this value and a suitably large " +
         "message has also been sent. To fix this problem increase replica.fetch.max.bytes in your broker config to be " +

core/src/main/scala/kafka/server/ReplicaManager.scala

@@ -310,7 +310,7 @@ class ReplicaManager(val config: KafkaConfig,
     // If inter-broker protocol (IBP) < 1.0, the controller will send LeaderAndIsrRequest V0 which does not include isNew field.
     // In this case, the broker receiving the request cannot determine whether it is safe to create a partition if a log directory has failed.
     // Thus, we choose to halt the broker on any log directory failure if IBP < 1.0
-    val haltBrokerOnFailure = config.interBrokerProtocolVersion.isLessThan(IBP_1_0_IV0)
+    val haltBrokerOnFailure = metadataCache.metadataVersion().isLessThan(IBP_1_0_IV0)
     logDirFailureHandler = new LogDirFailureHandler("LogDirFailureHandler", haltBrokerOnFailure)
     logDirFailureHandler.start()
   }
@@ -1773,7 +1773,7 @@ class ReplicaManager(val config: KafkaConfig,
    * OffsetForLeaderEpoch request.
    */
   protected def initialFetchOffset(log: UnifiedLog): Long = {
-    if (config.interBrokerProtocolVersion.isTruncationOnFetchSupported() && log.latestEpoch.nonEmpty)
+    if (metadataCache.metadataVersion().isTruncationOnFetchSupported && log.latestEpoch.nonEmpty)
       log.logEndOffset
     else
       log.highWatermark
@@ -1903,7 +1903,7 @@ class ReplicaManager(val config: KafkaConfig,
   }
 
   protected def createReplicaFetcherManager(metrics: Metrics, time: Time, threadNamePrefix: Option[String], quotaManager: ReplicationQuotaManager) = {
-    new ReplicaFetcherManager(config, this, metrics, time, threadNamePrefix, quotaManager)
+    new ReplicaFetcherManager(config, this, metrics, time, threadNamePrefix, quotaManager, () => metadataCache.metadataVersion())
   }
 
   protected def createReplicaAlterLogDirsManager(quotaManager: ReplicationQuotaManager, brokerTopicStats: BrokerTopicStats) = {

core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala

@@ -31,7 +31,6 @@ import org.apache.kafka.common.internals.Topic
 import org.apache.kafka.image.{MetadataDelta, MetadataImage, TopicDelta, TopicsImage}
 import org.apache.kafka.metadata.authorizer.ClusterMetadataAuthorizer
 import org.apache.kafka.server.authorizer.Authorizer
-import org.apache.kafka.server.common.MetadataVersion
 
 import scala.collection.mutable
 
@@ -132,10 +131,7 @@ class BrokerMetadataPublisher(conf: KafkaConfig,
       // Publish the new metadata image to the metadata cache.
       metadataCache.setImage(newImage)
 
-      val metadataVersionLogMsg = newImage.features().metadataVersion() match {
-        case MetadataVersion.UNINITIALIZED => "un-initialized metadata.version"
-        case mv: MetadataVersion => s"metadata.version ${mv.featureLevel()}"
-      }
+      val metadataVersionLogMsg = s"metadata.version ${newImage.features().metadataVersion()}"
 
       if (_firstPublish) {
         info(s"Publishing initial metadata at offset $highestOffsetAndEpoch with $metadataVersionLogMsg.")

core/src/test/java/kafka/test/ClusterConfig.java

@@ -122,8 +122,8 @@ public Optional<File> trustStoreFile() {
         return Optional.ofNullable(trustStoreFile);
     }
 
-    public Optional<MetadataVersion> metadataVersion() {
-        return Optional.ofNullable(metadataVersion);
+    public MetadataVersion metadataVersion() {
+        return metadataVersion;
     }
 
     public Properties brokerServerProperties(int brokerId) {
@@ -133,7 +133,7 @@ public Properties brokerServerProperties(int brokerId) {
     public Map<String, String> nameTags() {
         Map<String, String> tags = new LinkedHashMap<>(4);
         name().ifPresent(name -> tags.put("Name", name));
-        metadataVersion().ifPresent(mv -> tags.put("MetadataVersion", mv.toString()));
+        tags.put("MetadataVersion", metadataVersion.toString());
         tags.put("Security", securityProtocol.name());
         listenerName().ifPresent(listener -> tags.put("Listener", listener));
         return tags;
@@ -150,11 +150,12 @@ public ClusterConfig copyOf() {
     }
 
     public static Builder defaultClusterBuilder() {
-        return new Builder(Type.ZK, 1, 1, true, SecurityProtocol.PLAINTEXT);
+        return new Builder(Type.ZK, 1, 1, true, SecurityProtocol.PLAINTEXT, MetadataVersion.latest());
     }
 
-    public static Builder clusterBuilder(Type type, int brokers, int controllers, boolean autoStart, SecurityProtocol securityProtocol) {
-        return new Builder(type, brokers, controllers, autoStart, securityProtocol);
+    public static Builder clusterBuilder(Type type, int brokers, int controllers, boolean autoStart,
+                                         SecurityProtocol securityProtocol, MetadataVersion metadataVersion) {
+        return new Builder(type, brokers, controllers, autoStart, securityProtocol, metadataVersion);
     }
 
     public static class Builder {
@@ -168,12 +169,13 @@ public static class Builder {
         private File trustStoreFile;
         private MetadataVersion metadataVersion;
 
-        Builder(Type type, int brokers, int controllers, boolean autoStart, SecurityProtocol securityProtocol) {
+        Builder(Type type, int brokers, int controllers, boolean autoStart, SecurityProtocol securityProtocol, MetadataVersion metadataVersion) {
             this.type = type;
             this.brokers = brokers;
             this.controllers = controllers;
             this.autoStart = autoStart;
             this.securityProtocol = securityProtocol;
+            this.metadataVersion = metadataVersion;
         }
 
         public Builder type(Type type) {

core/src/test/java/kafka/test/ClusterTestExtensionsTest.java

@@ -25,6 +25,7 @@
 import kafka.test.annotation.ClusterTests;
 import kafka.test.annotation.Type;
 import kafka.test.junit.ClusterTestExtensions;
+import org.apache.kafka.server.common.MetadataVersion;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
@@ -109,4 +110,9 @@ public void testNoAutoStart() {
         clusterInstance.start();
         Assertions.assertNotNull(clusterInstance.anyBrokerSocketServer());
     }
+
+    @ClusterTest
+    public void testDefaults(ClusterConfig config) {
+        Assertions.assertEquals(MetadataVersion.IBP_3_3_IV2, config.metadataVersion());
+    }
 }

core/src/test/java/kafka/test/annotation/ClusterTest.java

@@ -41,6 +41,6 @@
     String name() default "";
     SecurityProtocol securityProtocol() default SecurityProtocol.PLAINTEXT;
     String listener() default "";
-    MetadataVersion metadataVersion() default MetadataVersion.UNINITIALIZED;
+    MetadataVersion metadataVersion() default MetadataVersion.IBP_3_3_IV2;
     ClusterConfigProperty[] serverProperties() default {};
 }

core/src/test/java/kafka/test/junit/ClusterTestExtensions.java

@@ -25,7 +25,6 @@
 import kafka.test.annotation.ClusterTest;
 import kafka.test.annotation.ClusterTests;
 import kafka.test.annotation.Type;
-import org.apache.kafka.server.common.MetadataVersion;
 import org.junit.jupiter.api.extension.ExtensionContext;
 import org.junit.jupiter.api.extension.TestTemplateInvocationContext;
 import org.junit.jupiter.api.extension.TestTemplateInvocationContextProvider;
@@ -180,7 +179,8 @@ private void processClusterTest(ExtensionContext context, ClusterTest annot, Clu
                 throw new IllegalStateException();
         }
 
-        ClusterConfig.Builder builder = ClusterConfig.clusterBuilder(type, brokers, controllers, autoStart, annot.securityProtocol());
+        ClusterConfig.Builder builder = ClusterConfig.clusterBuilder(type, brokers, controllers, autoStart,
+            annot.securityProtocol(), annot.metadataVersion());
         if (!annot.name().isEmpty()) {
             builder.name(annot.name());
         } else {
@@ -195,10 +195,6 @@ private void processClusterTest(ExtensionContext context, ClusterTest annot, Clu
             properties.put(property.key(), property.value());
         }
 
-        if (!annot.metadataVersion().equals(MetadataVersion.UNINITIALIZED)) {
-            builder.metadataVersion(annot.metadataVersion());
-        }
-
         ClusterConfig config = builder.build();
         config.serverProperties().putAll(properties);
         type.invocationContexts(config, testInvocations);