THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
Status
Current state: "Under Discussion"
Discussion thread: http://kafka.markmail.org/thread/glcyw3bvngtvobbs
JIRA: KAFKA-4093
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
It is increasingly common for organisations to have multiple Kafka clusters and there are a number of situations where it would be useful if the different clusters could be uniquely identified: monitoring, auditing, log aggregation, preventing clients and brokers from connecting to the wrong cluster and more. This KIP focuses on the monitoring and auditing use cases while ensuring that the design can be evolved to satisfy other use cases (this is covered in more detail in the "Potential Future Improvements" section).
One point worth covering is whether we already have something that could be used to uniquely identify a cluster. Today, each Kafka Cluster contains a set of brokers and an associated set of Zookeeper instances. You can identify each broker by a hostname or IP address (or a set of hostnames and IP addresses) and port (or ports), and can identify each Zookeeper instance by a hostname or IP address (or a set of hostnames and IP addresses) and port (or ports). Unfortunately, sets of hostnames, IP addresses, and ports are not a great way to uniquely identify a cluster as they might change over time. Brokers (and zookeeper instances) might die and be replaced with new instances on new machines, or might be moved to different machines. Clusters might be consolidated together, or hosts might be reused in new clusters.
Given that, we propose the introduction of a new concept: a Kafka cluster id.
Goals
All Kafka clusters should have a cluster id.
The cluster id should be unique and immutable so that external tools can associate data to a particular cluster.
The cluster id should be available via broker metrics so that it's easily exported to monitoring tools with existing metric reporters (e.g. JmxReporter).
The cluster id should be exposed to client and broker metric reporters so that they can tag, categorise or namespace metrics based on the cluster id.
Client interceptors should have access to the cluster id so that they can associate it with the message metadata being tracked.
Client serializers should have access to the cluster id so that they can include it in the message, if desired (e.g. as part of a standard message header).
Public Interfaces
Overview
We propose adding the cluster_id to the Metadata response, storing it in ZooKeeper, exposing it via metrics, to serializers and client interceptors. The cluster_id should be unique, immutable, auto-generated and it may be used in situations where the allowed character set is restricted. Given this, we limit the allowed characters to the following regular expression [a-zA-Z0-9_\-]+, which corresponds to the characters used by the URL-safe Base64 variant.
Protocol
We will introduce a cluster_id field to MetadataResponse and we will make no changes to MetadataRequest. Both will be bumped to version 2:
Metadata Request (Version: 2) => [topics]
topics => STRING
Metadata Response (Version: 2) => cluster_id [brokers] controller_id [topic_metadata]
cluster_id => STRING
brokers => node_id host port rack
node_id => INT32
host => STRING
port => INT32
rack => NULLABLE_STRING
controller_id => INT32
topic_metadata => topic_error_code topic is_internal [partition_metadata]
topic_error_code => INT16
topic => STRING
is_internal => BOOLEAN
partition_metadata => partition_error_code partition_id leader [replicas] [isr]
partition_error_code => INT16
partition_id => INT32
leader => INT32
replicas => INT32
isr => INT32
Java APIs
We will add an id field to the Cluster class, which is populated from MetadataResponse and accessible from the internal Metadata class and the newly introduced ClusterListener interface:
package org.apache.kafka.common;
interface ClusterListener {void onClusterUpdate(Cluster cluster);
}
Note that Cluster is currently a public API class, but it is not exposed to users via public APIs. ClusterListener changes the latter.
Client interceptors, serializers and metric reporters who implement the interface above will receive the updated cluster once each client receives it. Note that this is optional and existing implementations will not be affected.
ZooKeeper and Tools
We propose the introduction of a new ZooKeeper znode /cluster/id where we store a unique and immutable id automatically generated by the first broker that is able to create the znode. For consistency with other znodes stored by Kafka, we will store the data as JSON. An example follows:
{"version": 1,
"id": "vPeOCWypqUOSepEvx0cbog"
}
Proposed Changes
Broker
We update KafkaServer.startup() to invoke getOrGenerateClusterId() after initZk().
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.
Finally, we will add a Yammer metrics Gauge that exposes the cluster id via kafka.cluster:type=Cluster,name=ClusterId in KafkaServer.
Clients
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
A new protocol version will be added for the Metadata protocol message, so it should not affect existing clients.
ClusterListener is a new interface so existing serializers, client interceptors and metric reports will not be affected unless they choose to implement it.
Tools that want to take advantage of the cluster id while supporting older brokers will have to handle the fact that the older brokers won't have a cluster id.
Test Plan
Extend
Metadataprotocol tests to check that the cluster id is always returned.Modify the existing upgrade system tests to verify that the cluster id is set after the upgrade.
Test that the cluster id is exposed correctly in client and broker metrics.
Test that verifies that serializers, client interceptors and metric reporters that implement
ClusterListenerreceive the cluster id once a metadata response is received.
Rejected Alternatives
1. Add a method in each interceptor interface that is called on metadata updates to pass the cluster id (and potentially other metadata)
This seems like a good option, but it's not backwards compatible while we still support Java 7. In Java 8, we would have the option to introduce a default method with an empty implementation. This is also specific to interceptors. If we wanted to apply the same approach to metric reporters and serializers, we would probably want a common interface (as in the current proposal).
2. Expose the cluster id in ProducerRecord and ConsumerRecord
This would make it available to some of the client interceptor methods. The cluster id looks out of place in either of these classes as they represent records while the cluster id is a mostly static piece of data.
3. Don't expose the cluster id to client pluggable interfaces
We leave it to the implementations to do the metadata request themselves if they need the data. Since we don't have public API for this yet, it would require using internal request classes or reimplementing them.
4. Pass the cluster id via the configure method of Configurable classes
In order to make the cluster id available via the configure method, we would need to delay its invocation until after we get the first metadata response for each client.
The configure method of interceptors and serializers would no longer be invoked during the constructor of KafkaConsumer and KafkaProducer. Instead, it would be invoked after the first metadata response is received. This could pose problems for implementations that were relying on the previous behaviour. Furthermore, the first ProducerInterceptor.onSend() call would happen after the first metadata response is received instead of immediately after KafkaProducer.send() is called.
Potential Future Improvements
Expose cluster information including cluster id via AdminClient: this is likely to happen as part of KIP-4.
Add a human-readable cluster name to complement the id. This is useful, but it's worth exploring separately as there are a few options in the design space. One option that looks promising is to allow user-defined tags on resources (cluster, brokers, topics, etc.)
Use the cluster id to ensure that brokers are connected to the right cluster: it's useful, but something that can be done later via a separate KIP. One of the discussion points is how the broker knows its cluster id (e.g. via a config or by storing it after the first connection to the cluster).
Use the cluster id to ensure that clients are connected to the right cluster: given that clients could potentially connect to multiple clusters for failover reasons (i.e.
bootstrap.serverscan point to a VIP), it may make sense for the config be a list of cluster ids or a cluster id regex.Expose cluster id as a client metric. Kafka Metrics doesn't support non-numeric metric values at the moment so it would have to be extended to support that.