jsancio commented on code in PR #12050: URL: https://github.com/apache/kafka/pull/12050#discussion_r871808351
########## metadata/src/main/java/org/apache/kafka/controller/BootstrapMetadata.java: ########## @@ -0,0 +1,198 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.controller; + +import org.apache.kafka.common.metadata.FeatureLevelRecord; +import org.apache.kafka.common.metadata.MetadataRecordType; +import org.apache.kafka.controller.util.SnapshotFileReader; +import org.apache.kafka.controller.util.SnapshotFileWriter; +import org.apache.kafka.raft.Batch; +import org.apache.kafka.raft.BatchReader; +import org.apache.kafka.raft.RaftClient; +import org.apache.kafka.server.common.ApiMessageAndVersion; +import org.apache.kafka.server.common.MetadataVersion; +import org.apache.kafka.snapshot.SnapshotReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.ExecutionException; + + +/** + * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the + * format is the same as a KRaft snapshot. + */ +public class BootstrapMetadata { + private static final Logger log = LoggerFactory.getLogger(BootstrapMetadata.class); + + public static final String BOOTSTRAP_FILE = "bootstrap.snapshot"; + + private final MetadataVersion metadataVersion; + + private final List<ApiMessageAndVersion> records; + + BootstrapMetadata(MetadataVersion metadataVersion, List<ApiMessageAndVersion> records) { + this.metadataVersion = metadataVersion; + this.records = Collections.unmodifiableList(records); + } + + public MetadataVersion metadataVersion() { + return this.metadataVersion; + } + + public List<ApiMessageAndVersion> records() { + return records; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + BootstrapMetadata metadata = (BootstrapMetadata) o; + return metadataVersion == metadata.metadataVersion; + } + + @Override + public int hashCode() { + return Objects.hash(metadataVersion); + } + + @Override + public String toString() { + return "BootstrapMetadata{" + + "metadataVersion=" + metadataVersion + + '}'; + } + + /** + * A raft client listener that simply collects all of the commits and snapshots into a mapping of + * metadata record type to list of records. + */ + private static class BootstrapListener implements RaftClient.Listener<ApiMessageAndVersion> { + private final List<ApiMessageAndVersion> records = new ArrayList<>(); + + @Override + public void handleCommit(BatchReader<ApiMessageAndVersion> reader) { + try { + while (reader.hasNext()) { + Batch<ApiMessageAndVersion> batch = reader.next(); + records.addAll(batch.records()); + } + } finally { + reader.close(); + } + } + + @Override + public void handleSnapshot(SnapshotReader<ApiMessageAndVersion> reader) { + try { + while (reader.hasNext()) { + Batch<ApiMessageAndVersion> batch = reader.next(); + for (ApiMessageAndVersion messageAndVersion : batch) { + records.add(messageAndVersion); + } + } + } finally { + reader.close(); + } + } + } + + public static BootstrapMetadata create(MetadataVersion metadataVersion) { + return create(metadataVersion, new ArrayList<>()); + } + + public static BootstrapMetadata create(MetadataVersion metadataVersion, List<ApiMessageAndVersion> records) { + if (!metadataVersion.isKRaftSupported()) { + throw new IllegalArgumentException("Cannot create BootstrapMetadata with a non-KRaft metadata version."); + } + records.add(new ApiMessageAndVersion( + new FeatureLevelRecord() + .setName(MetadataVersion.FEATURE_NAME) + .setFeatureLevel(metadataVersion.featureLevel()), + FeatureLevelRecord.LOWEST_SUPPORTED_VERSION)); + + return new BootstrapMetadata(metadataVersion, records); + } + + /** + * 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. + * @return The read-only bootstrap metadata + * @throws Exception + */ + public static BootstrapMetadata load(Path bootstrapDir) throws Exception { + final Path bootstrapPath = bootstrapDir.resolve(BOOTSTRAP_FILE); + + if (!Files.exists(bootstrapPath)) { + log.debug("Missing bootstrap file, this appears to be a KRaft preview cluster."); + return BootstrapMetadata.create(MetadataVersion.IBP_3_0_IV0); + } + + BootstrapListener listener = new BootstrapListener(); + SnapshotFileReader reader = new SnapshotFileReader(bootstrapPath.toString(), listener); + reader.startup(); + try { + reader.caughtUpFuture().get(); Review Comment: It looks like this implementation starts a thread and blocks on that thread reading the bootstrap snapshot? Why is the thread needed? Why not read the snapshot on this thread? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
