...

The reason for this KIP is to remove the requirement of brokers needed needing to run the kafka-storage tool format  before starting Kafka. When running kafka-storage format , nodes are required to supply a --cluster-id  argument, which represents the cluster ID to which the node belongs, and this is persisted to the node's meta.properties  file. Below are the other data that each node persists to disk upon invoking kafka-storage format :

For brokers, meta.properties ’ other data: node.id  and directory id , are obtained from the node’s static config and randomly generated, respectively. Persisting this data does not need to be done before starting kafkahave to be done during storage formatting and can be done later during startup.

For controllers, meta.properties ’ directory id may come from —-initial-controllers or --standalone , but otherwise controllers are the same as brokers with respect to the above data.

...

Introduce a new metadata record to store cluster id and a new MetadataVersion (MV) that supports encoding/decoding this record. This means that during formatting, the bootstrap ClusterIdRecord is only written if the node is formatted with a MV that supports this feature. When a node runs kafka-storage format, some value for MV is always resolved and written to disk alongside meta.properties . The value either comes from --release-version  or --feature , or it defaults to the latest production MV.

If this feature is supported by the cluster's MV, the first elected KRaft leader will write the ClusterIdRecord upon becoming the active controller alongside the other bootstrap metadata records.

...

Proposed Changes

meta.properties can will be written during kafka broker/controller startup if it doesn't exist already (from formatting)

...

• Introduce a new MetadataVersion for this feature alongside a new metadata record called ClusterIdRecord
• One invariant of this feature is that if the persisted MV supports this feature, a ClusterIdRecord must also be persisted
  • This is enforceable along the write-path for MV, which occurs at two points: formatting a node (specifically controllers who can become leader) and upgrading the MV using kafka-features upgrade 
    • During formatting, nodes must resolve a MV with which to format. This comes from the --release-version/--feature flag and defaults to the latest production MV. If the MV at format time supports this feature, a ClusterIdRecord must be written as part of the bootstrap metadata checkpoint.
      • The first active controller will write the ClusterIdRecord + MV as part of the bootstrap metadata records write if the MV supports this feature
    • During MV upgrades, successfully upgrading the MV to one that supports this feature requires successfully committing a ClusterIdRecord alongside the new MetadataVersion feature record.
  • There is a precedent already for this kind of invariant which is enforced along the write path with kraft.version and the VotersRecord 
• However, kafka should still be able to handle the case where a leader is elected who does not have clusterId in meta.properties, which can occur if a majority of voters do not have a clusterId
  • This is because KRaft does not need cluster.id  in order to elect a leader (i.e. the cluster id for all voters can be empty and elect a leader)
  • In this case, the active controller will write a cluster id record during the bootstrap metadata write.

Nodes When nodes discover persist cluster.id from the metadata publishing pipeline, they persist it to meta.properties from metadata publishing pipeline

• Nodes whose meta.properties do not contain cluster.id  will discover this value via the metadata publishing pipeline
• The point at which nodes can discover this value in-memory is after both:
  • Learning of a HWM from the leader, which the leader allows for because it will send valid fetch responses back to nodes who do not have a cluster id
  • The MetadataLoader is registered as a listener to the raft layer
  • The readers of cluster id initialized during startup can wait for both the above before being initialized. 

...