KAFKA-19648; Cluster metadata bootstrapping with kraft checkpoint

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

@@ -24,7 +24,7 @@ import org.apache.kafka.common.internals.Topic
 import org.apache.kafka.common.utils.{AppInfoParser, Time, Utils}
 import org.apache.kafka.common.{KafkaException, Uuid}
 import org.apache.kafka.metadata.KafkaConfigSchema
-import org.apache.kafka.metadata.bootstrap.{BootstrapDirectory, BootstrapMetadata}
+import org.apache.kafka.metadata.bootstrap.BootstrapMetadata
 import org.apache.kafka.metadata.properties.MetaPropertiesEnsemble.VerificationFlag.{REQUIRE_AT_LEAST_ONE_VALID, REQUIRE_METADATA_LOG_DIR}
 import org.apache.kafka.metadata.properties.{MetaProperties, MetaPropertiesEnsemble}
 import org.apache.kafka.raft.QuorumConfig
@@ -181,8 +181,7 @@ object KafkaRaftServer {
     }
 
     // Load the BootstrapMetadata.
-    val bootstrapDirectory = new BootstrapDirectory(config.metadataLogDir)
-    val bootstrapMetadata = bootstrapDirectory.read()
+    val bootstrapMetadata = BootstrapMetadata.fromDirectory(java.nio.file.Path.of(config.metadataLogDir))
     (metaPropsEnsemble, bootstrapMetadata)
   }
 

core/src/main/scala/kafka/tools/StorageTool.scala

@@ -123,6 +123,7 @@ object StorageTool extends Logging {
     val formatter = new Formatter().
       setPrintStream(printStream).
       setNodeId(config.nodeId).
+      setWriteBootstrapSnapshot(config.processRoles.contains(ProcessRole.ControllerRole)).
       setClusterId(namespace.getString("cluster_id")).
       setUnstableFeatureVersionsEnabled(config.unstableFeatureVersionsEnabled).
       setIgnoreFormatted(namespace.getBoolean("ignore_formatted")).

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

@@ -21,12 +21,13 @@ import java.nio.file.Files
 import java.util.{Optional, Properties}
 import org.apache.kafka.common.{KafkaException, Uuid}
 import org.apache.kafka.common.utils.Utils
-import org.apache.kafka.metadata.bootstrap.{BootstrapDirectory, BootstrapMetadata}
+import org.apache.kafka.metadata.bootstrap.{BootstrapMetadata, BootstrapTestUtils}
 import org.apache.kafka.metadata.properties.{MetaProperties, MetaPropertiesEnsemble, MetaPropertiesVersion, PropertiesUtils}
-import org.apache.kafka.raft.{KRaftConfigs, MetadataLogConfig, QuorumConfig}
+import org.apache.kafka.metadata.storage.Formatter
 import org.apache.kafka.network.SocketServerConfigs
-import org.apache.kafka.server.config.ServerLogConfigs
+import org.apache.kafka.raft.{KRaftConfigs, MetadataLogConfig, QuorumConfig}
 import org.apache.kafka.server.common.MetadataVersion
+import org.apache.kafka.server.config.ServerLogConfigs
 import org.apache.kafka.storage.internals.log.UnifiedLog
 import org.apache.kafka.test.TestUtils
 import org.junit.jupiter.api.Assertions._
@@ -109,8 +110,13 @@ class KafkaRaftServerTest {
   }
 
   private def writeBootstrapMetadata(logDir: File, metadataVersion: MetadataVersion): Unit = {
-    val bootstrapDirectory = new BootstrapDirectory(logDir.toString)
-    bootstrapDirectory.writeBinaryFile(BootstrapMetadata.fromVersion(metadataVersion, "test"))
+    Formatter.writeBoostrapSnapshot(
+      logDir.toString,
+      BootstrapMetadata.fromVersion(metadataVersion, "test"),
+      Optional.empty(),
+      0.toShort,
+      "CONTROLLER"
+    )
   }
 
   @Test
@@ -272,20 +278,32 @@ class KafkaRaftServerTest {
       setDirectoryId(Uuid.fromString("4jm0e-YRYeB6CCKBvwoS8w")).
       build()
 
-    val configProperties = new Properties
-    configProperties.put(KRaftConfigs.PROCESS_ROLES_CONFIG, "broker,controller")
-    configProperties.put(KRaftConfigs.NODE_ID_CONFIG, nodeId.toString)
-    configProperties.put(SocketServerConfigs.LISTENERS_CONFIG, "PLAINTEXT://127.0.0.1:9092,SSL://127.0.0.1:9093")
-    configProperties.put(QuorumConfig.QUORUM_VOTERS_CONFIG, s"$nodeId@localhost:9093")
-    configProperties.put(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "SSL")
+    val logDir = TestUtils.tempDirectory()
+    try {
+      writeMetaProperties(logDir, metaProperties)
+      writeBootstrapMetadata(logDir, MetadataVersion.IBP_3_3_IV3)
+
+      val configProperties = new Properties
+      configProperties.put(KRaftConfigs.PROCESS_ROLES_CONFIG, "broker,controller")
+      configProperties.put(KRaftConfigs.NODE_ID_CONFIG, nodeId.toString)
+      configProperties.put(SocketServerConfigs.LISTENERS_CONFIG, "PLAINTEXT://127.0.0.1:9092,SSL://127.0.0.1:9093")
+      configProperties.put(QuorumConfig.QUORUM_VOTERS_CONFIG, s"$nodeId@localhost:9093")
+      configProperties.put(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "SSL")
+      configProperties.put(ServerLogConfigs.LOG_DIR_CONFIG, logDir.getAbsolutePath)
+      val config = KafkaConfig.fromProps(configProperties)
 
-    val (metaPropertiesEnsemble, bootstrapMetadata) =
-      invokeLoadMetaProperties(metaProperties, configProperties, Some(MetadataVersion.IBP_3_3_IV3))
+      val (metaPropertiesEnsemble, _) =
+        KafkaRaftServer.initializeLogDirs(config, MetaPropertiesEnsemble.LOG, "")
 
-    assertEquals(metaProperties, metaPropertiesEnsemble.logDirProps().values().iterator().next())
-    assertTrue(metaPropertiesEnsemble.errorLogDirs().isEmpty)
-    assertTrue(metaPropertiesEnsemble.emptyLogDirs().isEmpty)
-    assertEquals(bootstrapMetadata.metadataVersion(), MetadataVersion.IBP_3_3_IV3)
+      assertEquals(metaProperties, metaPropertiesEnsemble.logDirProps().values().iterator().next())
+      assertTrue(metaPropertiesEnsemble.errorLogDirs().isEmpty)
+      assertTrue(metaPropertiesEnsemble.emptyLogDirs().isEmpty)
+
+      val bootstrapMetadata = BootstrapTestUtils.readBootstrapMetadata(logDir.getAbsolutePath)
+      assertEquals(MetadataVersion.IBP_3_3_IV3, bootstrapMetadata.metadataVersion())
+    } finally {
+      Utils.delete(logDir)
+    }
   }
 
   @Test

core/src/test/scala/unit/kafka/tools/StorageToolTest.scala

@@ -27,12 +27,12 @@ import kafka.utils.TestUtils
 import net.sourceforge.argparse4j.inf.ArgumentParserException
 import org.apache.kafka.common.metadata.UserScramCredentialRecord
 import org.apache.kafka.common.utils.Utils
-import org.apache.kafka.server.common.{Feature, MetadataVersion}
-import org.apache.kafka.metadata.bootstrap.BootstrapDirectory
+import org.apache.kafka.metadata.bootstrap.BootstrapTestUtils
 import org.apache.kafka.metadata.properties.{MetaPropertiesEnsemble, PropertiesUtils}
 import org.apache.kafka.metadata.storage.FormatterException
 import org.apache.kafka.network.SocketServerConfigs
 import org.apache.kafka.raft.{KRaftConfigs, MetadataLogConfig, QuorumConfig}
+import org.apache.kafka.server.common.{Feature, MetadataVersion}
 import org.apache.kafka.server.config.{ServerConfigs, ServerLogConfigs}
 import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows, assertTrue}
 import org.junit.jupiter.api.{Test, Timeout}
@@ -389,15 +389,15 @@ Found problem:
   def testFormatWithReleaseVersionAndFeatureOverride(): Unit = {
     val availableDirs = Seq(TestUtils.tempDir())
     val properties = new Properties()
-    properties.putAll(defaultStaticQuorumProperties)
+    properties.putAll(defaultDynamicQuorumProperties)
     properties.setProperty("log.dirs", availableDirs.mkString(","))
     val stream = new ByteArrayOutputStream()
     assertEquals(0, runFormatCommand(stream, properties, Seq(
-      "--release-version", "3.7-IV0",
+      "--release-version", "3.7-IV0", "--standalone",
       "--feature", "share.version=1")))
 
     // Verify that the feature override is applied by checking the bootstrap metadata
-    val bootstrapMetadata = new BootstrapDirectory(availableDirs.head.toString).read
+    val bootstrapMetadata = BootstrapTestUtils.readBootstrapMetadata(availableDirs.head.toString)
 
     // Verify that the share.version feature is set to 1 as specified
     assertEquals(1.toShort, bootstrapMetadata.featureLevel("share.version"),
@@ -408,25 +408,25 @@ Found problem:
       "Failed to find release version in output: " + stream.toString())
 
     // Verify that the format command completed successfully with features
-    assertTrue(stream.toString().contains("Formatting metadata directory"),
+    assertTrue(stream.toString().contains("Formatting dynamic metadata voter directory"),
       "Failed to find formatting message in output: " + stream.toString())
   }
 
   @Test
   def testFormatWithMultipleFeatures(): Unit = {
     val availableDirs = Seq(TestUtils.tempDir())
     val properties = new Properties()
-    properties.putAll(defaultStaticQuorumProperties)
+    properties.putAll(defaultDynamicQuorumProperties)
     properties.setProperty("log.dirs", availableDirs.mkString(","))
     val stream = new ByteArrayOutputStream()
     assertEquals(0, runFormatCommand(stream, properties, Seq(
-      "--release-version", "3.8-IV0",
+      "--release-version", "3.8-IV0", "--standalone",
       "--feature", "share.version=1",
       "--feature", "transaction.version=2",
       "--feature", "group.version=1")))
 
     // Verify that all features are properly bootstrapped by checking the bootstrap metadata
-    val bootstrapMetadata = new BootstrapDirectory(availableDirs.head.toString).read
+    val bootstrapMetadata = BootstrapTestUtils.readBootstrapMetadata(availableDirs.head.toString)
 
     // Verify that all specified features are set correctly
     assertEquals(1.toShort, bootstrapMetadata.featureLevel("share.version"),
@@ -441,7 +441,7 @@ Found problem:
       "Failed to find release version in output: " + stream.toString())
 
     // Verify that the format command completed successfully with multiple features
-    assertTrue(stream.toString().contains("Formatting metadata directory"),
+    assertTrue(stream.toString().contains("Formatting dynamic metadata voter directory"),
       "Failed to find formatting message in output: " + stream.toString())
   }
 
@@ -851,11 +851,11 @@ Found problem:
   def testBootstrapScramRecords(): Unit = {
     val availableDirs = Seq(TestUtils.tempDir())
     val properties = new Properties()
-    properties.putAll(defaultStaticQuorumProperties)
+    properties.putAll(defaultDynamicQuorumProperties)
     properties.setProperty("log.dirs", availableDirs.mkString(","))
     val stream = new ByteArrayOutputStream()
     val arguments = ListBuffer[String](
-      "--release-version", "3.9-IV0",
+      "--release-version", "3.9-IV0", "--standalone",
       "--add-scram", "SCRAM-SHA-512=[name=alice,password=changeit]",
       "--add-scram", "SCRAM-SHA-512=[name=bob,password=changeit]"
     )
@@ -864,7 +864,7 @@ Found problem:
 
     // Not doing full SCRAM record validation since that's covered elsewhere.
     // Just checking that we generate the correct number of records
-    val bootstrapMetadata = new BootstrapDirectory(availableDirs.head.toString).read
+    val bootstrapMetadata = BootstrapTestUtils.readBootstrapMetadata(availableDirs.head.toString)
     val scramRecords = bootstrapMetadata.records().asScala
       .filter(apiMessageAndVersion => apiMessageAndVersion.message().isInstanceOf[UserScramCredentialRecord])
       .map(apiMessageAndVersion => apiMessageAndVersion.message().asInstanceOf[UserScramCredentialRecord])

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

@@ -127,6 +127,7 @@
 
 import org.slf4j.Logger;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.EnumSet;
 import java.util.HashMap;
@@ -1067,7 +1068,31 @@ public void handleLoadSnapshot(SnapshotReader<ApiMessageAndVersion> reader) {
 
         @Override
         public void handleLoadBootstrap(SnapshotReader<ApiMessageAndVersion> reader) {
-            reader.close();
+            appendRaftEvent(String.format("handleLoadBootstrap[snapshotId=%s]", reader.snapshotId()), () -> {
+                try {
+                    String snapshotName = Snapshots.filenameFromSnapshotId(reader.snapshotId());
+                    if (isActiveController()) {
+                        throw fatalFaultHandler.handleFault("Asked to load bootstrap snapshot " + snapshotName +
+                                ", but we are the active controller at epoch " + curClaimEpoch);
+                    }
+                    List<ApiMessageAndVersion> records = new ArrayList<>();
+                    while (reader.hasNext()) {
+                        Batch<ApiMessageAndVersion> batch = reader.next();
+                        records.addAll(batch.records());
+                    }
+                    if (!records.isEmpty()) {
+                        log.debug("Loaded {} bootstrap records from {}", records.size(), snapshotName);
+                        bootstrapMetadata = BootstrapMetadata.fromRecords(records, "bootstrap");
+                    }
+                } catch (FaultHandlerException e) {
+                    throw e;
+                } catch (Throwable e) {
+                    throw fatalFaultHandler.handleFault("Error while loading bootstrap snapshot " +
+                            reader.snapshotId(), e);
+                } finally {
+                    reader.close();
+                }
+            });
         }
 
         @Override
@@ -1450,9 +1475,11 @@ private void replay(ApiMessage message, Optional<OffsetAndEpoch> snapshotId, lon
     private volatile int curClaimEpoch;
 
     /**
-     * The bootstrap metadata to use for initialization if needed.
+     * The bootstrap mtryinetadata to use for initialization if needed.
+     * Written by handleLoadBootstrap and read during CompleteActivationEvent,
+     * both on the controller event thread.
      */
-    private final BootstrapMetadata bootstrapMetadata;
+    private BootstrapMetadata bootstrapMetadata;
 
     /**
      * The maximum number of records per batch to allow.