You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "kevin-wu24 (via GitHub)" <gi...@apache.org> on 2025/10/15 16:21:51 UTC
Re: [PR] KAFKA-19648: Formatter refactoring [kafka]
kevin-wu24 commented on code in PR #20707:
URL: https://github.com/apache/kafka/pull/20707#discussion_r2433217087
##########
metadata/src/main/java/org/apache/kafka/metadata/storage/Formatter.java:
##########
Review Comment:
There is a clearer way to make these changes. This method is what writes the `0-0.checkpoint` currently. We should pass the bootstrap metadata object here and append the metadata records using `RecordsSnapshotWriter.append(bootstrapMetadata.records())`.
##########
metadata/src/main/java/org/apache/kafka/metadata/storage/Formatter.java:
##########
@@ -437,14 +438,40 @@ 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);
+ File parentDir = new File(writeLogDir);
+ File clusterMetadataDirectory = new File(parentDir, String.format("%s-%d",
+ CLUSTER_METADATA_TOPIC_PARTITION.topic(),
+ CLUSTER_METADATA_TOPIC_PARTITION.partition()));
+ Files.createDirectories(clusterMetadataDirectory.toPath());
+ BootstrapDirectory bootstrapDirectory = new BootstrapDirectory(clusterMetadataDirectory.getPath());
bootstrapDirectory.writeBinaryFile(bootstrapMetadata);
if (directoryTypes.get(writeLogDir).isDynamicMetadataDirectory()) {
- writeDynamicQuorumSnapshot(writeLogDir,
+ writeDynamicQuorumSnapshot(clusterMetadataDirectory.getPath(),
initialControllers.get(),
featureLevels.get(KRaftVersion.FEATURE_NAME),
controllerListenerName);
+ File createdBoostrapCheckpoint = new File(clusterMetadataDirectory.getPath() + "/" + BootstrapDirectory.BINARY_BOOTSTRAP_FILENAME);
+ File created000Checkpoint = new File(clusterMetadataDirectory.getPath() + "/" + BootstrapDirectory.BINARY_CHECKPOINT_FILENAME);
+ Files.write(
+ createdBoostrapCheckpoint.toPath(),
+ Files.readAllBytes(created000Checkpoint.toPath()),
+ StandardOpenOption.APPEND);
+ try {
+ created000Checkpoint.delete();
+ createdBoostrapCheckpoint.renameTo(created000Checkpoint);
+ } catch (Exception ex) {
+ throw new RuntimeException("Failed operation to combine metadata and kraft records: ", ex);
+ }
+ } else {
+ File createdBoostrapCheckpoint = new File(clusterMetadataDirectory.getPath() + "/" + BootstrapDirectory.BINARY_BOOTSTRAP_FILENAME);
+ File created000Checkpoint = new File(clusterMetadataDirectory.getPath() + "/" + BootstrapDirectory.BINARY_CHECKPOINT_FILENAME);
+ try {
+ createdBoostrapCheckpoint.renameTo(created000Checkpoint);
+ } catch (Exception ex) {
+ throw new RuntimeException("Failed to rename file: ", ex);
+ }
Review Comment:
This code is confusing. Instead of doing this renaming and deleting. We should instead remove the call to write the bootstrap metadata to disk on Line 447, since we're no longer writing `bootstrap.checkpoint` anymore, and follow the other comments for writing metadata records to `0-0.checkpoint`.
We can check if an old `bootstrap.checkpoint` exists and delete it, since IIRC that was part of the KIP.
##########
metadata/src/main/java/org/apache/kafka/metadata/storage/Formatter.java:
##########
@@ -437,14 +438,40 @@ 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);
+ File parentDir = new File(writeLogDir);
+ File clusterMetadataDirectory = new File(parentDir, String.format("%s-%d",
+ CLUSTER_METADATA_TOPIC_PARTITION.topic(),
+ CLUSTER_METADATA_TOPIC_PARTITION.partition()));
+ Files.createDirectories(clusterMetadataDirectory.toPath());
+ BootstrapDirectory bootstrapDirectory = new BootstrapDirectory(clusterMetadataDirectory.getPath());
bootstrapDirectory.writeBinaryFile(bootstrapMetadata);
if (directoryTypes.get(writeLogDir).isDynamicMetadataDirectory()) {
- writeDynamicQuorumSnapshot(writeLogDir,
+ writeDynamicQuorumSnapshot(clusterMetadataDirectory.getPath(),
Review Comment:
We should rename `writeDynamicQuorumSnapshot` to `writeZeroSnapshot`, and always write it going forward. The semantics that change here are that we should not write the KRaft control records (KRaft version and voter set) when `!isDynamicMetadataDirectory()`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org