KAFKA-19648: KIP-1170 Unify cluster metadata bootstrapping

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

@@ -593,6 +593,7 @@ class DumpLogSegmentsTest {
     )
 
     val lastContainedLogTimestamp = 10000
+    val emptyOptional: Optional[java.util.List[ApiMessageAndVersion]] = Optional.empty()
 
     Using.resource(
       new RecordsSnapshotWriter.Builder()
@@ -601,7 +602,7 @@ class DumpLogSegmentsTest {
         .setRawSnapshotWriter(metadataLog.createNewSnapshot(new OffsetAndEpoch(0, 0)).get)
         .setKraftVersion(KRaftVersion.KRAFT_VERSION_1)
         .setVoterSet(Optional.of(VoterSetTest.voterSet(VoterSetTest.voterMap(IntStream.of(1, 2, 3), true))))
-        .build(MetadataRecordSerde.INSTANCE)
+        .build(MetadataRecordSerde.INSTANCE, emptyOptional)
     ) { snapshotWriter =>
       snapshotWriter.append(metadataRecords)
       snapshotWriter.freeze()

metadata/src/main/java/org/apache/kafka/metadata/storage/Formatter.java

@@ -437,14 +437,11 @@ void doFormat(BootstrapMetadata bootstrapMetadata) throws Exception {
                     directoryTypes.get(writeLogDir).description(), writeLogDir,
                     MetadataVersion.FEATURE_NAME, releaseVersion);
                 Files.createDirectories(Paths.get(writeLogDir));
-                BootstrapDirectory bootstrapDirectory = new BootstrapDirectory(writeLogDir);
-                bootstrapDirectory.writeBinaryFile(bootstrapMetadata);
-                if (directoryTypes.get(writeLogDir).isDynamicMetadataDirectory()) {
-                    writeDynamicQuorumSnapshot(writeLogDir,
-                        initialControllers.get(),
-                        featureLevels.get(KRaftVersion.FEATURE_NAME),
-                        controllerListenerName);
-                }
+                writeBoostrapSnapshot(writeLogDir,
+                    bootstrapMetadata,
+                    initialControllers.get(),
+                    featureLevels.get(KRaftVersion.FEATURE_NAME),
+                    controllerListenerName);
             });
             copier.setWriteErrorHandler((errorLogDir, e) -> {
                 throw new FormatterException("Error while writing meta.properties file " +
@@ -492,8 +489,9 @@ static DirectoryType calculate(
         }
     }
 
-    static void writeDynamicQuorumSnapshot(
+    static void writeBoostrapSnapshot(
         String writeLogDir,
+        BootstrapMetadata bootstrapMetadata,
         DynamicVoters initialControllers,
         short kraftVersion,
         String controllerListenerName
@@ -511,7 +509,7 @@ static void writeDynamicQuorumSnapshot(
                 Snapshots.BOOTSTRAP_SNAPSHOT_ID)).
             setKraftVersion(KRaftVersion.fromFeatureLevel(kraftVersion)).
             setVoterSet(Optional.of(voterSet));
-        try (RecordsSnapshotWriter<ApiMessageAndVersion> writer = builder.build(new MetadataRecordSerde())) {
+        try (RecordsSnapshotWriter<ApiMessageAndVersion> writer = builder.build(new MetadataRecordSerde(), Optional.of(bootstrapMetadata.records()))) {
             writer.freeze();
         }
     }

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

@@ -729,7 +729,7 @@ public Optional<SnapshotWriter<ApiMessageAndVersion>> createSnapshot(
                 .setLastContainedLogTimestamp(lastContainedLogTimestamp)
                 .setTime(new MockTime())
                 .setRawSnapshotWriter(createNewSnapshot(snapshotId))
-                .build(new MetadataRecordSerde())
+                .build(new MetadataRecordSerde(), Optional.empty())
         );
     }
 

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

@@ -3726,7 +3726,7 @@ public Optional<SnapshotWriter<T>> createSnapshot(
                 .setRawSnapshotWriter(wrappedWriter)
                 .setKraftVersion(partitionState.kraftVersionAtOffset(lastContainedLogOffset))
                 .setVoterSet(partitionState.voterSetAtOffset(lastContainedLogOffset))
-                .build(serde);
+                .build(serde, Optional.empty());
         });
     }
 

raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotWriter.java

@@ -191,7 +191,7 @@ public Builder setVoterSet(Optional<VoterSet> voterSet) {
             return this;
         }
 
-        public <T> RecordsSnapshotWriter<T> build(RecordSerde<T> serde) {
+        public <T> RecordsSnapshotWriter<T> build(RecordSerde<T> serde, Optional<List<T>> bootstrapRecords) {
             if (rawSnapshotWriter.isEmpty()) {
                 throw new IllegalStateException("Builder::build called without a RawSnapshotWriter");
             } else if (rawSnapshotWriter.get().sizeInBytes() != 0) {
@@ -213,6 +213,8 @@ public <T> RecordsSnapshotWriter<T> build(RecordSerde<T> serde) {
                 serde
             );
 
+            bootstrapRecords.ifPresent(writer::append);
+
             writer.accumulator.appendControlMessages((baseOffset, epoch, compression, buffer) -> {
                 long now = time.milliseconds();
                 try (MemoryRecordsBuilder builder = new MemoryRecordsBuilder(

raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java

@@ -2224,7 +2224,7 @@ private static SnapshotWriter<String> snapshotWriter(RaftClientTestContext conte
         return new RecordsSnapshotWriter.Builder()
             .setTime(context.time)
             .setRawSnapshotWriter(snapshot)
-            .build(new StringSerde());
+            .build(new StringSerde(), Optional.empty());
     }
 
     private static final class MemorySnapshotWriter implements RawSnapshotWriter {

raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java

@@ -268,7 +268,7 @@ Builder withEmptySnapshot(OffsetAndEpoch snapshotId) {
                     .setTime(time)
                     .setKraftVersion(KRaftVersion.KRAFT_VERSION_0)
                     .setRawSnapshotWriter(log.createNewSnapshotUnchecked(snapshotId).get())
-                    .build(SERDE)
+                    .build(SERDE, Optional.empty())
             ) {
                 snapshot.freeze();
             }
@@ -363,7 +363,7 @@ Builder withBootstrapSnapshot(Optional<VoterSet> voters) {
                     .setKraftVersion(kraftVersion)
                     .setVoterSet(voters);
 
-                try (RecordsSnapshotWriter<String> writer = builder.build(SERDE)) {
+                try (RecordsSnapshotWriter<String> writer = builder.build(SERDE, Optional.empty())) {
                     writer.freeze();
                 }
             } else {

raft/src/test/java/org/apache/kafka/raft/internals/KRaftControlRecordStateMachineTest.java

@@ -172,7 +172,7 @@ void testUpdateWithEmptySnapshot() {
         // Create a snapshot that doesn't have any kraft.version or voter set control records
         RecordsSnapshotWriter.Builder builder = new RecordsSnapshotWriter.Builder()
             .setRawSnapshotWriter(log.createNewSnapshotUnchecked(new OffsetAndEpoch(10, epoch)).get());
-        try (RecordsSnapshotWriter<?> writer = builder.build(STRING_SERDE)) {
+        try (RecordsSnapshotWriter<?> writer = builder.build(STRING_SERDE, Optional.empty())) {
             writer.freeze();
         }
         log.truncateToLatestSnapshot();
@@ -234,7 +234,7 @@ void testUpdateWithSnapshot() {
             .setRawSnapshotWriter(log.createNewSnapshotUnchecked(new OffsetAndEpoch(10, epoch)).get())
             .setKraftVersion(kraftVersion)
             .setVoterSet(Optional.of(voterSet));
-        try (RecordsSnapshotWriter<?> writer = builder.build(STRING_SERDE)) {
+        try (RecordsSnapshotWriter<?> writer = builder.build(STRING_SERDE, Optional.empty())) {
             writer.freeze();
         }
         log.truncateToLatestSnapshot();
@@ -272,7 +272,7 @@ void testUpdateWithSnapshotAndLogOverride() {
             .setRawSnapshotWriter(log.createNewSnapshotUnchecked(snapshotId).get())
             .setKraftVersion(kraftVersion)
             .setVoterSet(Optional.of(snapshotVoterSet));
-        try (RecordsSnapshotWriter<?> writer = builder.build(STRING_SERDE)) {
+        try (RecordsSnapshotWriter<?> writer = builder.build(STRING_SERDE, Optional.empty())) {
             writer.freeze();
         }
         log.truncateToLatestSnapshot();

raft/src/test/java/org/apache/kafka/raft/internals/RecordsIteratorTest.java

@@ -171,7 +171,7 @@ public void testControlRecordIterationWithKraftVersion0() {
             .setRawSnapshotWriter(
                 new MockRawSnapshotWriter(new OffsetAndEpoch(100, 10), buffer::set)
             );
-        try (RecordsSnapshotWriter<String> snapshot = builder.build(STRING_SERDE)) {
+        try (RecordsSnapshotWriter<String> snapshot = builder.build(STRING_SERDE, Optional.empty())) {
             snapshot.append(List.of("a", "b", "c"));
             snapshot.append(List.of("d", "e", "f"));
             snapshot.append(List.of("g", "h", "i"));
@@ -221,7 +221,7 @@ public void testControlRecordIterationWithKraftVersion1() {
             .setRawSnapshotWriter(
                 new MockRawSnapshotWriter(new OffsetAndEpoch(100, 10), buffer::set)
             );
-        try (RecordsSnapshotWriter<String> snapshot = builder.build(STRING_SERDE)) {
+        try (RecordsSnapshotWriter<String> snapshot = builder.build(STRING_SERDE, Optional.empty())) {
             snapshot.append(List.of("a", "b", "c"));
             snapshot.append(List.of("d", "e", "f"));
             snapshot.append(List.of("g", "h", "i"));

raft/src/test/java/org/apache/kafka/snapshot/RecordsSnapshotWriterTest.java

@@ -60,7 +60,7 @@ void testBuilderKRaftVersion0() {
             .setRawSnapshotWriter(
                 new MockRawSnapshotWriter(snapshotId, buffer::set)
             );
-        try (RecordsSnapshotWriter<String> snapshot = builder.build(STRING_SERDE)) {
+        try (RecordsSnapshotWriter<String> snapshot = builder.build(STRING_SERDE, Optional.empty())) {
             snapshot.freeze();
         }
 
@@ -114,7 +114,7 @@ void testBuilderKRaftVersion0WithVoterSet() {
                 new MockRawSnapshotWriter(snapshotId, buffer::set)
             );
 
-        assertThrows(IllegalStateException.class, () -> builder.build(STRING_SERDE));
+        assertThrows(IllegalStateException.class, () -> builder.build(STRING_SERDE, Optional.empty()));
     }
 
     @Test
@@ -133,7 +133,7 @@ void testKBuilderRaftVersion1WithVoterSet() {
             .setRawSnapshotWriter(
                 new MockRawSnapshotWriter(snapshotId, buffer::set)
             );
-        try (RecordsSnapshotWriter<String> snapshot = builder.build(STRING_SERDE)) {
+        try (RecordsSnapshotWriter<String> snapshot = builder.build(STRING_SERDE, Optional.empty())) {
             snapshot.freeze();
         }
 
@@ -191,7 +191,7 @@ void testBuilderKRaftVersion1WithoutVoterSet() {
             .setRawSnapshotWriter(
                 new MockRawSnapshotWriter(snapshotId, buffer::set)
             );
-        try (RecordsSnapshotWriter<String> snapshot = builder.build(STRING_SERDE)) {
+        try (RecordsSnapshotWriter<String> snapshot = builder.build(STRING_SERDE, Optional.empty())) {
             snapshot.freeze();
         }