...

Prior to this KIP, it is expected that all nodes running kafka have invoked the kafka-storage format command and have persisted a V1 meta.properties with a cluster.id file prior to starting kafka. We do this by trying to read in the meta.properties during startup, and crashing if that file does not contain all the data we expect.

Observer controllers are controllers who are either not part of the static voter set in kraft.version=0, or controllers who format with --no-initial-controllers  in kraft.version=1. Non-observer/"bootstrap controllers" are therefore either part of the static voter set in kraft.version=0, or controllers who format with --initial-controllers or --standalone  in a dynamic quorum setup. This distinction is important because observer controllers are not responsible for the initial KRaft leader election, but bootstrap controllers are. The most precise definition of bootstrap controller for this purposes of this KIP is a controller who participates in the election process of the initial KRaft leader.

We can still enforce that bootstrap controllers must have formatted (and therefore persisted a cluster id to a V1 meta.properties) prior to starting kafka. The validation of meta.properties during startup described above can still be done whenever the node is a bootstrap controller (i.e. part of the static voters config, or if a 0-0.checkpoint exists). Just like with kafka today, a consequence of this requirement is that when KRaft elects the initial leader, it is guaranteed to have a non-null cluster id.

After this KIP, it is no longer required for nodes to run kafka-storage format in order to start kafka. Additionally, the --cluster-id argument for kafka-storage format will now be optional, rather than required. However, operators still have the option to format nodes to set the MV, feature versions, scram credentials, or to properly provision a kraft.version=1 cluster. If an operator decides to format a node, they will still need to specify --cluster-id if the MV which is resolved by the formatter does not support the ClusterIdRecord  feature.

A consequence of the above relaxation is the the initial KRaft leader can be elected without a cluster.id in its local meta.properties. In this case, the initial KRaft leader (active controller) will randomly generate a cluster id as part of the bootstrap metadata write (this either writes the bootstrap records as a transaction if the MV supports it, or as an atomic batch). This is similar to how cluster.id is generated in ZK-based kafka.We still need to do this validation mainly for kraft.version=0 clusters with newer software versions but an older MV (kraft.version=1 clusters require proper formatting of a majority of bootstrap controllers to elect a leader). Otherwise, these static quorum clusters will have no way to persist a cluster id to meta.properties if the operator skips formatting on all nodes (which would be possible without this requirement). 

Proposed Changes

Remove the requirement of brokers and observer controllers nodes to format before starting kafka

• After KIP-1262, kafka operators no longer need to invoke kafka-storage format to start kafka. Clusters who skip formatting essentially have a bootstrapping "state" of the latest MetadataVersion and a ClusterIdRecord who is generated by the initial active controller.
• Operators can still format clusters the same way as prior to this KIP. However, the --cluster-id  flag now becomes optional.
  • When --cluster-id  is specified, the formatter writes meta.properties  V1.
  • When --cluster-id  is not specified, and the formatter's resolved MV supports ClusterIdRecord, write meta.properties V2 with an empty cluster.id.
    • If the formatter's resolved MV does not support ClusterIdRecord, formatting will fail.
  format all nodes. They only need to format the "bootstrap" controller nodes before starting kafka. 
  • "Bootstrap controllers" are the nodes listed the controller.quorum.voters  static config when using a static quorum, or the controllers who format with --initial-controllers/--standalone  in a dynamic quorum setup.
  • Formatting brokers and observer controllers is now optional. Failing to run kafka-storage format on these nodes before starting kafka will no longer crash startup.
  • Faling to run kafka-storage format on bootstrap controllers will cause kafka to crash during startup. This behavior is identical to how kafka behaves today with respect to bootstrap controllers not being formatted properly. 

meta.properties 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 the following points: formatting a node (specifically controllers who can become leader), the bootstrap metadata write of the initial active controller, and upgrading the MV using using kafka-features upgrade 
    • During formatting, nodes must resolve a MV with which to format. This comes from the --release-version/--feature flag 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 (0-0.checkpoint post KIP-1170)
    • 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 

...

• 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
  • Besides the raft client, readers of cluster id initialized during startup can block for both the above conditions to be met before being initialized.
  • The raft client's clusterId will be updated after discovering this value from the metadata pipeline. One detail here is that observer controllers with auto-join must wait until they have a cluster id before trying to add or remove themselves.  
• This can be implemented as a MetadataPublisher that registers to the raft client alongside the MetadataLoader.
• This can be implemented as a MetadataPublisher that registers to the MetadataLoader.
• It is the responsibility of this MetadataPublisher to persist a discovered value of cluster.id to all meta.properties files before making the cluster.id available to in-memory data structures.
  • Because of JBOD, nodes can have many log.dirs that each have a meta.properties file (or we need to write each instance of meta.properties V2 during startup without cluster.id if we skipped formatting). 
• The MetadataPublisher is the single writer of cluster.id (unless it is already set by formatting, in which case there are no writers to cluster.id) both in persisting the value to meta.properties, and writing the in-memory representation of cluster.id
  • There are many readers of this value, but most of them can block their initialization until this value is discovered for the sake of simplicity.
  • The KafkaRaftClient and QuorumController are the readers who need to handle the concept of an empty/not-set cluster id. 

Pros:

• Fetch replication automatically handles persistence of the cluster id for each local node
• Raft module remains independent from metadata module in that KRaft is only responsible for consensus. ClusterID is simply another piece of metadata on which Kraft achieves consensus

...

• Rough design:
  • Node can complete a future to allow this value to be discovered by readers outside of kraft layer who need it during startup
  • Raft layer is brought up early during startup, so it is fine to wait until this future completes to proceed with initializing the server
  • Brokers/observers can start Kafka with no cluster id, and rely on the fetch/another RPC response to discover it in-memory
    • If discovered, node persists cluster id to meta.properties during the startup process before in-memory readers of cluster id
• Pros:
  • Backwards compatibility is straightforward, since new nodes on old clusters keep using meta.properties for persisting cluster id
  • This functionality is not tied to a MetadataVersion, meaning that any Kafka broker/observer with a software version that supports this KIP can use it, rather than the whole cluster needing to be on some MV >= X.
  • Kraft can easily do its own cluster ID validation for its RPCs, since nodes receive cluster ID via the fetch response if they do not know it and can update that state in-memory + persist it
• Cons:
  • Since each local node's meta.properties  is its source of truth, if this file is deleted or is changed, it means the broker/observer controller can join another cluster.
    • This is no worse than what exists currently.

Remove the requirement for formatting for brokers and observer controllers, but still require it for bootstrap controllers

Bootstrap controllers are those who participate in the initial KRaft leader election. These are the nodes specified by controller.quorum.voters in a static quorum, and the nodes who format with --standalone/--initial-controllers  in a dynamic quorum. 

The main reason we can relax this to say no nodes require formatting is that KRaft can elect a leader when all nodes do not have cluster.id defined.