...

Currently, Kafka relies on the operator/caller of 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 contained in (or no cluster id) contained in meta.properties, and  and that RPCs will only be handled by the leader if the request's cluster ID (if present) matches the leader's.

...

Introduce meta.properties v2 with optional cluster id (same as v0)id. However, node.id  and directory.id  are guaranteed to exist.

ClusterIdRecord + MetadataVersion

...

--cluster-id is now optional for when formatting brokers + observer controllers. This flag is still required for "bootstrapping" controllers (i.e. controllers who are part of an initial dynamic voter set (, determined by the --standalone  or --initial-controllers)   flags, or who are part of a static voter set).

Proposed Changes

...

Option 1: Continue to persist cluster id in meta.properties but have KRaft discover it + persist it

...

can

...

• If discovered, node persists cluster id to meta.properties during the startup process before in-memory readers of cluster id

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

• During startup, write meta.properties if it does not exist with node.id  and directory.id
• If meta.properties exists without a cluster.id, it will be discovered later
  • Updating the cluster.id in-memory and persisting it to meta.properties  must happen together.

Introduce a metadata record for cluster id + observers persist cluster id to meta.properties from metadata publishing pipeline

...

• 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

...

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

...

• Introduce a new Metadata Version that supports a ClusterIDRecord.
• Brokers/observers, and non-bootstrap controllers can start KRaft with no cluster id, and rely on metadata publishing pipeline to discover it in-memory
  • 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.
  • Ideally, starting KRaft with no cluster id should only be allowed the "first time" (i.e. if the cluster metadata partition doesn't exist on the node yet).
  Upon restart, nodes need to compare the cluster id from the metadata partition with their local meta.properties . The value in the metadata partition takes precedence.
  • If these values are different, log an error or crash.
  If cluster ID exists in the metadata partition but not in meta.properties , write it to meta.properties .
• Nodes that startup with a cluster.id in meta.properties  assume it to be correct and pass it to KRaft
  • If this ID doesn't match the leader's the leader will reject these requests
• Bootstrap controllers can add a mandatory “cluster id” record during formatting
• However, kafka should still be able to handle the case where a leader is elected without a cluster id in meta.properties , since KRaft does not need cluster.id  in order to elect a leader
  • In this case, the active controller will write a cluster id record during the bootstrap metadata write
  , or the initial leader can randomly generate a UUID as part of bootstrap metadata records writeUpon restart, in terms of ensuring a node does not talk to another cluster during its lifetime in the presence of deletions/changes to meta.properties , this approach is no better than Option 1 unless the local node can use the cluster id value in its local metadata partition BEFORE contacting the leader and learning of the HWM. This means a node needs to read its local metadata log BEFORE discovering the HWM from the leader during startup to retrieve the cluster id.The local node does not know if the cluster id record in its metadata log is committed or not until it contacts the leader and learns the HWM. However, in order to ensure the local node does not talk to another cluster, it needs to provide a cluster ID in its fetch request to discover the HWM. This is a circularity.However, in practice this should be okay. If ClusterIdRecord with value X was written as part of the bootstrap metadata records write (this is the only way it is possible for a node to have this record in its metadata partition), this means that a KRaft quorum with cluster id X was formed. The offset might be uncommitted and change if the bootstrap write fails or leadership changes, but consensus for the value X as the cluster ID has already been achieved upon electing a leader
  • .
• 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
    Can duplicate the cluster ID as a control record
    • Having a mechanism for “pushing-down” cluster ID from metadata to raft may be complicated.
    • We can duplicate data and have a raft level control record for cluster ID.The fact that the raft client does currently do validation on cluster id does make it unique (i.e. it is used by both metadata and raft layers to prevent nodes from talking to different “clusters”, which is an argument for option 1).For example, if Kraft was used to replicate other data besides the metadata partition, there would still be a concept of cluster id, which needs to be the same across all partitions on the node being managed by Kraft.

Compatibility, Deprecation, and Migration Plan

...

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

Rejected Alternatives

...

Continue to persist cluster id in meta.properties but have KRaft discover it + persist it via FetchResponse

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