Page History

...

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 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.

...

During startup of the KafkaRaftServer, we attempt to read the meta.properties file

• If meta.properties  does not exist and the node is a broker/observer controller, write meta.properties V2 with node.id  and directory.id
  • This means either this node skipped formatting, or the file/disk was lost
• If meta.properties does not exist and the node is a bootstrap controller, throw a runtime exception
  • This is the same behavior as kafka today
• If meta.properties exists without a cluster.id and is V2, it will be discovered later (described below)
  • One correctness invariant of this feature is that updating the cluster.id in-memory must occur after persisting it to meta.properties.
• If meta.properties exists with a cluster.id, kafka behaves as it does today:
  • The node assumes it to be correct and passes it to KRaft
  • If this ID doesn't match the KRaft leader's ID, the leader will reject requests from the node
• If a broker/observer controller has already written a cluster id to its meta.properties, either from formatting or discovering it from the cluster metadata, it is impossible for it to learn of another cluster id via Fetch/FetchSnapshot.
  • For the broker (with clusterid = X) to receive a non-error FetchResponse with metadata records (which would be the only way to learn of a different ClusterIdRecord), the KRaft leader (clusterid = Y) must either receive a request without clusterid, or a request whose clusterid is Y. The broker fulfills neither of these conditions.
  • This case could happen when bootstrap endpoints point to the wrong cluster during restart of a node. KRaft's own clusterid checks would mean startup of this node would time out and shut down the node because it is not able to contact the quorum of another cluster.

...