This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: Under Discussion

Discussion thread: https://lists.apache.org/thread/vnzmqvcbfxo7hhyj9gzpgmdq59w3n7dy

JIRA: here 

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

The reason for this KIP is to remove the requirement of brokers needed to run the storage tool before starting Kafka. Currently, when brokers format in KRaft, they persist a UUID value representing a cluster ID, which is passed from the kafka-storage format  command to the meta.properties file. The main purpose of cluster id is to prevent nodes from contacting other Kafka clusters (ref KIP-78).

For brokers, meta.properties ’ other data: node.id  and directory id , are obtained from the node’s static config and randomly generated, respectively.

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

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

Background on cluster.id from ZooKeeper Kafka

Cluster id was a znode, /cluster/id , that was initially empty. During the startup of a cluster, brokers would race to write a random UUID in ZK to this znode, which would never change after being set, via getOrGenerateClusterId() .

Public Interfaces

meta.properties

Introduce meta.properties v2 with optional cluster id (same as v0).

ClusterIdRecord

Storage Tool

--cluster-id is now optional for 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

• 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 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
  • 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:
  
  • Tight coupling between KRaft layer and persisting a cluster id
    • Main question to answer is: why not use KRaft itself to bootstrap the cluster id like with other “cluster metadata” (e.g. metadata version)?

Option 2: Introduce a metadata record for cluster id

• Rough design:
  • Introduce a new Metadata Version that supports a ClusterIDRecord.
  • Brokers/observers can start Kraft with no cluster id, and rely on metadata publishing pipeline to discover it in-memory
  • Bootstrap controllers can add a mandatory “cluster id” record during formatting, or the initial leader can randomly generate a UUID as part of bootstrap metadata records write
    • Latter matches the ZK approach more closely, but may present some backwards compatibility challenges
• 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.
  • Users of cluster id must wait until after the node catches up to the metadata LEO and fetches the cluster ID
    • This might be okay, since we block controller server startup on things like authorizer futures completing, which also rely on fetching the metadata log.
    • Controller objects that are initialized with cluster id currently:
      • Authorizer futures: we block on these anyways, can move the construction of `endpointReadyFutures` to after local node fetches cluster ID
      • QuorumController/ClusterControlManager: Currently set to random UUID if DNE. Used to reject broker registration if IDs do not match.
      • ControllerApis: Reported in describe cluster response
      • ControllerRegistrationManager: reads but doesn’t use cluster ID
      • DynamicTopicCLusterQuotaPublisher: this is a publisher, so the handling is pretty trivial, just set cluster ID via onMetadataUpdate
  • Backwards compatibility/migration is non-trivial unless existing clusters are allowed to default to reading `meta.properties` if the cluster ID metadata record does not exist yet.
    • ClusterID record can only be written to the log after a leader with a new version containing this KIP is elected.

Compatibility, Deprecation, and Migration Plan

This section depends on which approach is chosen for the proposed changes.

Test Plan

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

Rejected Alternatives

WIP: Currently considering two solutions