...

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.

...

Currently, Kafka relies on the operator/caller of kafkaof kafka-storage format on all intended nodes in a cluster to generate consensus on the actual value of clusterID during format time. Additionally, KRaft guarantees that upon forming a given quorum X, X's members will all have the same cluster id (or no cluster id) contained in metain meta.properties, and that RPCs will only be handled by the leader . Additionally, when a KRaft node sends a request with a non-null cluster id, this request is only handled if the request's cluster ID (if present) matches the leader's.We still maintain that controllers who are part of the bootstrapped voter set must format, but observer controllers do not need to format, just like brokers. This can be enforced by requiring —-cluster-id  when any of the KIP-853 format flags are provided or when the local node is part of its static voter setID matches the handler's cluster ID.

Background on cluster.id from ZooKeeper Kafka

...

Introduce a new metadata record to store cluster id and a new MetadataVersion that supports encoding/decoding this record. This means that during formatting, the bootstrap ClusterIdRecord is only written if the cluster node is formatted with a MV that supports this feature.

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.

{
  "apiKey": 29,
  "type": "metadata",
  "name": "ClusterIdRecord",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ClusterId", "type": "uuidstring", "versions": "0+",
      "about": "The unique ID of this cluster" }
  ]
}

Storage Tool + Formatting requirements

Prior to this KIP, it is expected that all nodes running kafka have invoked the kafka-storage format command and have persisted a 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 are not part of the bootstrap voter set defined by --initial-controllers or --standalone in kraft.version=1 --cluster-id is now optional when formatting brokers + observer controllers. This flag is still required for "bootstrapping" controllers (i.e. controllers who format today with --no-initial-controllers). Non-observer/bootstrap controllers are therefore either part of an initial dynamic the static voter set , determined by the --standalone  or --initial-controllers  flags, or who are part of a static voter set)in kraft.version=0, or controllers who format with --initial-controllers or --standalone. This distinction is important because observer controllers are not responsible for the initial KRaft leader election, but bootstrap controllers are.

After this KIP, brokers and observer controllers are no longer guaranteed to have persisted a meta.properties with a cluster.id  value prior to starting kafka if the intended metadata version of the cluster supports this feature. Even after this KIP, we can still enforce that bootstrap controllers must have formatted (and therefore persisted a cluster id to meta.properties) prior to starting kafka. The validation of meta.properties described above can be done whenever the node is a bootstrap controller (i.e. part of the static voters config, or if a 0-0.checkpoint exists with a VotersRecord). 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 formatting of at least one node to elect a leader). Otherwise, these clusters will have no way to persist a cluster id to meta.properties if the operator skips formatting on all nodes.

Proposed Changes

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

• During startup of the KafkaRaftServer, we attempt to read the meta.properties file
• If , write meta.properties if it   does not exist, write it with node.id  and directory.id
  • This means either this node skipped formatting, or the file/disk was lost
• If meta.properties exists without a cluster.id, it will be discovered later (described below)
  • One correctness invariant of this feature is that updating
  • Updating the cluster.id in-memory and persisting it to meta.properties  must happen togetherbe atomic.

...

• If meta.properties

...

• Upon discovering the cluster ID for the first time, these nodes need to persist this to meta.properties, and update the raft client in-memory.

• 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 the leader will reject these requests

...

Introduce a metadata record for cluster id

• 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 discover persist cluster id 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. 

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

Cons:

• Currently, KRaft client also needs to be aware of the cluster ID for its own RPC handling, but the raft module does not decode metadata records
  • Having a mechanism for “pushing-down” cluster ID from metadata to raft may be complicated.
  • We can could duplicate data and have a raft level control record for cluster ID.

Compatibility, Deprecation, and Migration Plan

Since this feature is associated with a new metadata record and MetadataVersion. Broker , broker bootstrapping with cluster ID must be supported is required on all MVs < X where X is the first MV that supports this feature. Because some MetadataVersion is resolved during each node's formatting, we can determine at format time if a cluster id ClusterIdRecord is needed as part of a controller's 0-0/bootstrap.checkpoint.

Test Plan

• Unit tests
• Integration tests
• System tests to verify cross-software-version compatibility

...