KAFKA-4151; Update public docs for Cluster Id (KIP-78)

- Updated implementation docs with details on Cluster Id generation.
- Mention cluster id in "noteworthy changes for 0.10.1.0" in upgrade docs.

Author: Sumit Arrawatia <sumit.arrawatia@gmail.com>
Author: arrawatia <sumit.arrawatia@gmail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #1895 from arrawatia/kip-78-docs

docs/implementation.html

@@ -337,7 +337,7 @@ Each of the consumers in the group registers under its group and creates a znode
 Consumers track the maximum offset they have consumed in each partition. This value is stored in a ZooKeeper directory if <code>offsets.storage=zookeeper</code>.
 </p>
 <pre>
-/consumers/[group_id]/offsets/[topic]/[partition_id] --> offset_counter_value ((persistent node)
+/consumers/[group_id]/offsets/[topic]/[partition_id] --> offset_counter_value (persistent node)
 </pre>
 
 <h4><a id="impl_zkowner" href="#impl_zkowner">Partition Owner registry</a></h4>
@@ -350,6 +350,15 @@ Each broker partition is consumed by a single consumer within a given consumer g
 /consumers/[group_id]/owners/[topic]/[partition_id] --> consumer_node_id (ephemeral node)
 </pre>
 
+<h4><a id="impl_clusterid" href="#impl_clusterid">Cluster Id</a></h4>
+
+<p>
+    The cluster id is a unique and immutable identifier assigned to a Kafka cluster. The cluster id can have a maximum of 22 characters and the allowed characters are defined by the regular expression [a-zA-Z0-9_\-]+, which corresponds to the characters used by the URL-safe Base64 variant with no padding. Conceptually, it is auto-generated when a cluster is started for the first time.
+</p>
+<p>
+    Implementation-wise, it is generated when a broker with version 0.10.1 or later is successfully started for the first time. The broker tries to get the cluster id from the <code>/cluster/id</code> znode during startup. If the znode does not exist, the broker generates a new cluster id and creates the znode with this cluster id.
+</p>
+
 <h4><a id="impl_brokerregistration" href="#impl_brokerregistration">Broker node registration</a></h4>
 
 <p>

docs/upgrade.html

@@ -52,17 +52,24 @@ Note: Because new protocols are introduced, it is important to upgrade your Kafk
     <li> The open file handlers of 0.10.0 will increase by ~33% because of the addition of time index files for each segment.</li>
     <li> The time index and offset index share the same index size configuration. Since each time index entry is 1.5x the size of offset index entry. User may need to increase log.index.size.max.bytes to avoid potential frequent log rolling. </li>
     <li> Due to the increased number of index files, on some brokers with large amount the log segments (e.g. >15K), the log loading process during the broker startup could be longer. Based on our experiment, setting the num.recovery.threads.per.data.dir to one may reduce the log loading time. </li>
-    <li> ListOffsetRequest v1 is introduced and used by default to support accurate offset search based on timestamp.
+
 </ul>
 
 <h5><a id="upgrade_1010_notable" href="#upgrade_1010_notable">Notable changes in 0.10.1.0</a></h5>
 <ul>
     <li> The new Java consumer is no longer in beta and we recommend it for all new development. The old Scala consumers are still supported, but they will be deprecated in the next release
          and will be removed in a future major release. </li>
+    <li> Kafka clusters can now be uniquely identified by a cluster id. It will be automatically generated when a broker is upgraded to 0.10.1.0. The cluster id is available via the kafka.server:type=KafkaServer,name=ClusterId metric and it is part of the Metadata response. Serializers, client interceptors and metric reporters can receive the cluster id by implementing the ClusterResourceListener interface. </li>
     <li> The BrokerState "RunningAsController" (value 4) has been removed. Due to a bug, a broker would only be in this state briefly before transitioning out of it and hence the impact of the removal should be minimal. The recommended way to detect if a given broker is the controller is via the kafka.controller:type=KafkaController,name=ActiveControllerCount metric. </li>
     <li> The new Java Consumer now allows users to search offsets by timestamp on partitions.
 </ul>
 
+<h5><a id="upgrade_1010_new_protocols" href="#upgrade_1010_new_protocols">New Protocol Versions</a></h5>
+<ul>
+    <li> ListOffsetRequest v1 is introduced and used by default to support accurate offset search based on timestamp.
+    <li> MetadataRequest/Response v2 has been introduced. v2 adds a new field "cluster_id" to MetadataResponse.
+</ul>
+
 <h4><a id="upgrade_10" href="#upgrade_10">Upgrading from 0.8.x or 0.9.x to 0.10.0.0</a></h4>
 0.10.0.0 has <a href="#upgrade_10_breaking">potential breaking changes</a> (please review before upgrading) and possible <a href="#upgrade_10_performance_impact">  performance impact following the upgrade</a>. By following the recommended rolling upgrade plan below, you guarantee no downtime and no performance impact during and following the upgrade.
 <br>