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,24 @@ 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 {
+    if (originals.containsKey(KafkaConfig.InterBrokerProtocolVersionProp)) {
+      // A user-supplied IBP was given
+      val configuredVersion = MetadataVersion.fromVersionString(interBrokerProtocolVersionString)
+      if (!configuredVersion.isKRaftSupported) {
+        throw new ConfigException(s"A non-KRaft version ${interBrokerProtocolVersionString} given for ${KafkaConfig.InterBrokerProtocolVersionProp}")
+      } else {
+        warn(s"${KafkaConfig.InterBrokerProtocolVersionProp} is deprecated in KRaft mode as of 3.3 and will only " +
+          s"be read when first upgrading from a KRaft prior to 3.3. See kafka-storage.sh help for details on setting " +
+          s"the metadata version for a new KRaft cluster.")
+      }
+    }
+    // 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

@@ -35,6 +35,7 @@ import org.apache.kafka.server.metrics.KafkaYammerMetrics
 
 import java.nio.file.Paths
 import scala.collection.Seq
+import scala.compat.java8.FunctionConverters.asJavaSupplier
 import scala.jdk.CollectionConverters._
 
 /**
@@ -180,13 +181,16 @@ object KafkaRaftServer {
           "If you intend to create a new broker, you should remove all data in your data directories (log.dirs).")
     }
 
-    // Load the bootstrap metadata file or, in the case of an upgrade from KRaft preview, bootstrap the
-    // metadata.version corresponding to a user-configured IBP.
-    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)
+    // Load the bootstrap metadata file. In the case of an upgrade from older KRaft where there is no bootstrap metadata,
+    // read the IBP from config in order to bootstrap the equivalent metadata version.
+    def getUserDefinedIBPVersionOrThrow(): MetadataVersion = {
+      if (config.originals.containsKey(KafkaConfig.InterBrokerProtocolVersionProp)) {
+        MetadataVersion.fromVersionString(config.interBrokerProtocolVersionString)
+      } else {
+        throw new KafkaException(s"Cannot upgrade from KRaft version prior to 3.3 without first setting ${KafkaConfig.InterBrokerProtocolVersionProp} on each broker.")
+      }
     }
+    val bootstrapMetadata = BootstrapMetadata.load(Paths.get(config.metadataLogDir), asJavaSupplier(() => getUserDefinedIBPVersionOrThrow()))
 
     (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 metadataVersion = metadataVersionSupplier()
+    val requestBuilder = ListOffsetsRequest.Builder.forReplica(metadataVersion.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 (metadataVersion.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,
+                            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/main/scala/kafka/tools/StorageTool.scala

@@ -49,7 +49,7 @@ object StorageTool extends Logging {
           val clusterId = namespace.getString("cluster_id")
           val metadataVersion = getMetadataVersion(namespace)
           if (!metadataVersion.isKRaftSupported) {
-            throw new TerseFailure(s"Must specify a metadata version of at least 1.")
+            throw new TerseFailure(s"Must specify a valid KRaft metadata version of at least 3.0.")
           }
           val metaProperties = buildMetadataProperties(clusterId, config.get)
           val ignoreFormatted = namespace.getBoolean("ignore_formatted")
@@ -99,7 +99,7 @@ object StorageTool extends Logging {
       action(storeTrue())
     formatParser.addArgument("--release-version", "-r").
       action(store()).
-      help(s"A release version to use for the initial metadata.version. The default is (${MetadataVersion.latest().version()})")
+      help(s"A KRaft release version to use for the initial metadata version. The minimum is 3.0, the default is (${MetadataVersion.latest().version()})")
 
     parser.parseArgsOrFail(args)
   }