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:

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. When running kafka-storage format , nodes are required to supply a --cluster-id  argument, which represents the cluster ID to which the node belongs, and this is persisted to the node's meta.properties  file. Below are the other data that each node persists to disk upon invoking kafka-storage format :

For brokers, meta.properties ’ other data: node.id  and directory id , are obtained from the node’s static config and randomly generated, respectively. Persisting this data does not need to be done before starting kafka.

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.

The main purpose of cluster id is to prevent nodes from contacting other Kafka clusters (ref KIP-78). This KIP seeks to preserve this behavior around cluster id, while removing the necessity of formatting brokers and observer controllers.

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 (or no cluster id) contained in meta.properties, and that RPCs will only be handled by the leader 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 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. However, node.id  and directory.id  are guaranteed to exist.

ClusterIdRecord + MetadataVersion

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 is formatted with a MV that supports this feature.

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

Storage Tool

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

meta.properties can 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

• 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.
• 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.
• 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 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 bootstrapping with cluster ID must be supported 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 is needed.

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