...

The implementation of getOrGenerateClusterId() is straightforward. It first tries to get the cluster id from /cluster/id. If the znode does not exist, it generates an id via UUID.randomUUID(), converts it to a String via URL-safe Base64 encoding and tries to create the znode. If the creation succeeds, the generated id is returned. If it fails with a ZkNodeExistsException, then another broker won the race, so we can just retrieve the cluster id from the expected znode path and return it. The returned cluster id will be logged at info level.

We will introduce a clusterId method to KafkaServer, which will be used by KafkaApis.handleTopicMetadataRequest to populate the cluster_id field in MetadataResponse version 2.

...

Client interceptors, serializers and metric reporters which implement the ClusterListener interface (checked via instanceof) will receive the current Cluster instance after every metadata response received by a consumer or producer. We will also log the cluster id received via the metadata response if it's different from the previously received cluster id (or if it is the first one). The clients will use an implementation of Metadata.Listener to hook into metadata updates.

Compatibility, Deprecation, and Migration Plan

...