KAFKA-13935 Fix static usages of IBP in KRaft mode

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

@@ -1799,7 +1799,9 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
       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. See kafka-storage.sh help for details.")
+        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

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 older KRaft, 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.MINIMUM_KRAFT_VERSION)
+    // 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

@@ -111,7 +111,7 @@ class RemoteLeaderEndPoint(logPrefix: String,
           .setCurrentLeaderEpoch(currentLeaderEpoch)
           .setTimestamp(earliestOrLatest)))
     val metadataVersion = metadataVersionSupplier()
-    val requestBuilder = ListOffsetsRequest.Builder.forReplica(metadataVersion.listOffsetRequestVersion(), brokerConfig.brokerId)
+    val requestBuilder = ListOffsetsRequest.Builder.forReplica(metadataVersion.listOffsetRequestVersion, brokerConfig.brokerId)
       .setTargetTimes(Collections.singletonList(topic))
 
     val clientResponse = blockingSender.sendRequest(requestBuilder)
@@ -146,7 +146,7 @@ class RemoteLeaderEndPoint(logPrefix: String,
     }
 
     val epochRequest = OffsetsForLeaderEpochRequest.Builder.forFollower(
-      metadataVersionSupplier().offsetForLeaderEpochRequestVersion(), topics, brokerConfig.brokerId)
+      metadataVersionSupplier().offsetForLeaderEpochRequestVersion, topics, brokerConfig.brokerId)
     debug(s"Sending offset for leader epoch request $epochRequest")
 
     try {
@@ -207,10 +207,10 @@ class RemoteLeaderEndPoint(logPrefix: String,
       None
     } else {
       val metadataVersion = metadataVersionSupplier()
-      val version: Short = if (metadataVersion.fetchRequestVersion() >= 13 && !fetchData.canUseTopicIds) {
+      val version: Short = if (metadataVersion.fetchRequestVersion >= 13 && !fetchData.canUseTopicIds) {
         12
       } else {
-        metadataVersion.fetchRequestVersion()
+        metadataVersion.fetchRequestVersion
       }
       val requestBuilder = FetchRequest.Builder
         .forReplica(version, brokerConfig.brokerId, maxWait, minBytes, fetchData.toSend)

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 (e.g., 3.2, 3.3) 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)
   }

core/src/test/scala/unit/kafka/server/KafkaRaftServerTest.scala

@@ -24,6 +24,7 @@ import kafka.log.UnifiedLog
 import org.apache.kafka.common.{KafkaException, Uuid}
 import org.apache.kafka.common.utils.Utils
 import org.apache.kafka.controller.BootstrapMetadata
+import org.apache.kafka.server.common.MetadataVersion
 import org.apache.kafka.test.TestUtils
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.Test
@@ -71,12 +72,13 @@ class KafkaRaftServerTest {
 
   private def invokeLoadMetaProperties(
     metaProperties: MetaProperties,
-    configProperties: Properties
+    configProperties: Properties,
+    metadataVersion: Option[MetadataVersion] = Some(MetadataVersion.latest())
   ): (MetaProperties, BootstrapMetadata, collection.Seq[String]) = {
     val tempLogDir = TestUtils.tempDirectory()
     try {
       writeMetaProperties(tempLogDir, metaProperties)
-
+      metadataVersion.foreach(mv => writeBootstrapMetadata(tempLogDir, mv))
       configProperties.put(KafkaConfig.LogDirProp, tempLogDir.getAbsolutePath)
       val config = KafkaConfig.fromProps(configProperties)
       KafkaRaftServer.initializeLogDirs(config)
@@ -94,6 +96,11 @@ class KafkaRaftServerTest {
     checkpoint.write(metaProperties.toProperties)
   }
 
+  private def writeBootstrapMetadata(logDir: File, metadataVersion: MetadataVersion): Unit = {
+    val bootstrapMetadata = BootstrapMetadata.create(metadataVersion)
+    BootstrapMetadata.write(bootstrapMetadata, logDir.toPath)
+  }
+
   @Test
   def testStartupFailsIfMetaPropertiesMissingInSomeLogDir(): Unit = {
     val clusterId = clusterIdBase64
@@ -147,6 +154,7 @@ class KafkaRaftServerTest {
     // One log dir is online and has properly formatted `meta.properties`
     val validDir = TestUtils.tempDirectory()
     writeMetaProperties(validDir, MetaProperties(clusterId, nodeId))
+    writeBootstrapMetadata(validDir, MetadataVersion.latest())
 
     // Use a regular file as an invalid log dir to trigger an IO error
     val invalidDir = TestUtils.tempFile("blah")
@@ -215,4 +223,47 @@ class KafkaRaftServerTest {
       () => KafkaRaftServer.initializeLogDirs(config))
   }
 
+  @Test
+  def testKRaftUpdateWithIBP(): Unit = {
+    val clusterId = clusterIdBase64
+    val nodeId = 0
+    val metaProperties = MetaProperties(clusterId, nodeId)
+
+    val configProperties = new Properties
+    configProperties.put(KafkaConfig.ProcessRolesProp, "broker,controller")
+    configProperties.put(KafkaConfig.NodeIdProp, nodeId.toString)
+    configProperties.put(KafkaConfig.ListenersProp, "PLAINTEXT://127.0.0.1:9092,SSL://127.0.0.1:9093")
+    configProperties.put(KafkaConfig.QuorumVotersProp, s"$nodeId@localhost:9093")
+    configProperties.put(KafkaConfig.ControllerListenerNamesProp, "SSL")
+    configProperties.put(KafkaConfig.InterBrokerProtocolVersionProp, "3.2")
+
+    val (loadedMetaProperties, bootstrapMetadata, offlineDirs) =
+      invokeLoadMetaProperties(metaProperties, configProperties, None)
+
+    assertEquals(metaProperties, loadedMetaProperties)
+    assertEquals(Seq.empty, offlineDirs)
+    assertEquals(bootstrapMetadata.metadataVersion(), MetadataVersion.IBP_3_2_IV0)
+  }
+
+  @Test
+  def testKRaftUpdateWithoutIBP(): Unit = {
+    val clusterId = clusterIdBase64
+    val nodeId = 0
+    val metaProperties = MetaProperties(clusterId, nodeId)
+
+    val logDir = TestUtils.tempDirectory()
+    writeMetaProperties(logDir, metaProperties)
+
+    val configProperties = new Properties
+    configProperties.put(KafkaConfig.ProcessRolesProp, "broker,controller")
+    configProperties.put(KafkaConfig.NodeIdProp, nodeId.toString)
+    configProperties.put(KafkaConfig.ListenersProp, "PLAINTEXT://127.0.0.1:9092,SSL://127.0.0.1:9093")
+    configProperties.put(KafkaConfig.QuorumVotersProp, s"$nodeId@localhost:9093")
+    configProperties.put(KafkaConfig.ControllerListenerNamesProp, "SSL")
+    configProperties.put(KafkaConfig.LogDirProp, logDir.getAbsolutePath)
+
+    val config = KafkaConfig.fromProps(configProperties)
+    assertEquals("Cannot upgrade from KRaft version prior to 3.3 without first setting inter.broker.protocol.version on each broker.",
+      assertThrows(classOf[KafkaException], () => KafkaRaftServer.initializeLogDirs(config)).getMessage)
+  }
 }

metadata/src/main/java/org/apache/kafka/controller/BootstrapMetadata.java

@@ -39,6 +39,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
+import java.util.function.Supplier;
 import java.util.stream.Stream;
 
 
@@ -144,18 +145,20 @@ public static BootstrapMetadata create(MetadataVersion metadataVersion, List<Api
     /**
      * Load a bootstrap snapshot into a read-only bootstrap metadata object and return it.
      *
-     * @param bootstrapDir      The directory from which to read the snapshot file.
-     * @param fallbackVersion   The metadata.version to boostrap if upgrading from KRaft
-     * @return                  The read-only bootstrap metadata
+     * @param bootstrapDir              The directory from which to read the snapshot file.
+     * @param fallbackVersionSupplier   A function that returns the metadata.version to use when upgrading from an older KRaft
+     * @return                          The read-only bootstrap metadata
      * @throws Exception
      */
-    public static BootstrapMetadata load(Path bootstrapDir, MetadataVersion fallbackVersion) throws Exception {
+    public static BootstrapMetadata load(Path bootstrapDir, Supplier<MetadataVersion> fallbackVersionSupplier) throws Exception {
         final Path bootstrapPath = bootstrapDir.resolve(BOOTSTRAP_FILE);
 
         if (!Files.exists(bootstrapPath)) {
+            // Upgrade scenario from KRaft prior to 3.3 (i.e., no bootstrap metadata present)
+            MetadataVersion fallbackVersion = fallbackVersionSupplier.get();
             if (fallbackVersion.isKRaftSupported()) {
                 log.debug("Missing bootstrap file, this appears to be a KRaft cluster older than 3.3. Setting metadata.version to {}.",
-                        fallbackVersion.featureLevel());
+                    fallbackVersion.featureLevel());
                 return BootstrapMetadata.create(fallbackVersion);
             } else {
                 throw new Exception(String.format("Could not set fallback bootstrap metadata with non-KRaft metadata version of %s", fallbackVersion));

metadata/src/test/java/org/apache/kafka/controller/BootstrapMetadataTest.java

@@ -39,16 +39,16 @@ public void testWriteAndReadBootstrapFile() throws Exception {
 
         assertTrue(Files.exists(tmpDir.resolve(BootstrapMetadata.BOOTSTRAP_FILE)));
 
-        BootstrapMetadata newMetadata = BootstrapMetadata.load(tmpDir, MetadataVersion.MINIMUM_KRAFT_VERSION);
+        BootstrapMetadata newMetadata = BootstrapMetadata.load(tmpDir, () -> MetadataVersion.MINIMUM_KRAFT_VERSION);
         assertEquals(metadata, newMetadata);
     }
 
     @Test
     public void testNoBootstrapFile() throws Exception {
         Path tmpDir = Files.createTempDirectory("BootstrapMetadataTest");
-        BootstrapMetadata metadata = BootstrapMetadata.load(tmpDir, MetadataVersion.MINIMUM_KRAFT_VERSION);
+        BootstrapMetadata metadata = BootstrapMetadata.load(tmpDir, () -> MetadataVersion.MINIMUM_KRAFT_VERSION);
         assertEquals(MetadataVersion.MINIMUM_KRAFT_VERSION, metadata.metadataVersion());
-        metadata = BootstrapMetadata.load(tmpDir, MetadataVersion.IBP_3_2_IV0);
+        metadata = BootstrapMetadata.load(tmpDir, () -> MetadataVersion.IBP_3_2_IV0);
         assertEquals(MetadataVersion.IBP_3_2_IV0, metadata.metadataVersion());
     }
 
@@ -65,7 +65,7 @@ public void testExistingBootstrapFile() throws Exception {
     public void testEmptyBootstrapFile() throws Exception {
         Path tmpDir = Files.createTempDirectory("BootstrapMetadataTest");
         Files.createFile(tmpDir.resolve(BootstrapMetadata.BOOTSTRAP_FILE));
-        assertThrows(Exception.class, () -> BootstrapMetadata.load(tmpDir, MetadataVersion.MINIMUM_KRAFT_VERSION),
+        assertThrows(Exception.class, () -> BootstrapMetadata.load(tmpDir, () -> MetadataVersion.MINIMUM_KRAFT_VERSION),
             "Should fail to load if no metadata.version is set");
     }
 
@@ -77,7 +77,7 @@ public void testGarbageBootstrapFile() throws Exception {
         byte[] data = new byte[100];
         random.nextBytes(data);
         Files.write(tmpDir.resolve(BootstrapMetadata.BOOTSTRAP_FILE), data, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
-        assertThrows(Exception.class, () -> BootstrapMetadata.load(tmpDir, MetadataVersion.MINIMUM_KRAFT_VERSION),
+        assertThrows(Exception.class, () -> BootstrapMetadata.load(tmpDir, () -> MetadataVersion.MINIMUM_KRAFT_VERSION),
             "Should fail on invalid data");
     }
 }