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.
Current state: [One of "Under Discussion", "Accepted", "Rejected"]
Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]
JIRA: here [Change the link from KAFKA-1 to your own ticket]
Authors: Daniel Urban, Gergely Harmadas, David Simon, Viktor Somogyi-Vass
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Apache Kafka is widely adopted for building distributed event streaming platforms. As Kafka clusters grow in size and usage, a common challenge arises: different teams or workloads often share the same physical Kafka cluster, which can cause operational and organizational friction.
Currently, Kafka lacks native support for logical separation of tenants (e.g., applications, teams) within a shared cluster. This leads to several challenges:
This KIP proposes introducing Virtual Clusters (VCs) to Kafka, a mechanism to logically separate workloads while sharing the same physical Kafka cluster. VCs will allow teams to operate independently, improving resource isolation, security, and management capabilities.
After considering a few approaches, listed in the Rejected Alternatives section, below is what we think is the least obtrusive approach to support virtual clusters in Kafka.
At the very highest level, we would like to create a new type of resource called “virtual clusters” and connect other resources like topics and groups and users to them. This would allow the brokers to handle requests transparently for the clients. In addition to this, generalizing the Cluster ACL resource, we could properly form the authorization layer necessary in a backward compatible way.
We would form a directional one level tree-like structure where we’d link topics to virtual clusters which would be stored in the __cluster_metadata topic. This solution is more advantageous over other approaches as links can be created and modified much easier than for instance a directory structure that has been proposed originally in KIP-37, allowing more elasticity to manage namespaces and there would be no changes required in the storage layer. Moreover this can align with the proposal for topic renames in KIP-516: Topic Identifiers.
Furthermore, links can be created with an alternative name to topics, allowing topic renaming in virtual clusters, but also linking topics between virtual clusters in order to achieve more complex use cases, like sharing topics between virtual clusters to provide readable views, or implement aggregation use-cases.
Currently topic names are heavily embedded into the protocol, therefore it’s challenging to rely only on topic IDs. We wanted to make our implementation as independent from physical topic names as possible while also respecting the current state of the protocol.
There are two major ways to create a topic in a virtual cluster compatible Kafka:
Deleting topics in virtual clusters would simply mean deleting the links. If all links are deleted, then the topic doesn’t belong to any virtual cluster, so administrators would simply need to delete the topic to get rid of it forever. Deleting the physical topic wouldn’t be allowed as long as there are links pointing to it. This is necessary to prevent dangling links.
One of our assumptions is that clients and users usually belong to the same virtual cluster and would rarely use topics in others. But if they’d do, then virtual cluster admins could link topics across virtual clusters. Therefore it would make sense to bind users and client-ids to virtual clusters and create one-to-many connections, where one virtual cluster could have many users/client-ids associated with, but user/client-ids can only be associated with one virtual cluster. Storing this information in the brokers also opens up the possibility for users/client-ids to act transparently in the virtual cluster. For instance they wouldn’t need to prefix topics they address with the name of the VC, instead they would continue using the same topic name. This is advantageous for backward compatibility reasons as clients can simply be added to virtual clusters without them having to make any protocol or configuration changes.
In a similar fashion to users, consumer groups can be linked to virtual clusters as well, so groups in different virtual clusters may have the same local name. Storing the offset of virtual groups would be the same as in the classical case. They would continue using __consumer-offsets. Furthermore, __consumer-offsets should stay as a global internal topic to preserve backward compatibility. When consumers are moved into virtual clusters, their group ID will be prefixed with the name of the virtual cluster and that will be the reference for the __consumer-offsets topic. Since clients don’t use this topic directly but through various APIs (admin APIs, offset commit, list offsets, etc.), they aren’t exposed directly to them and therefore virtual clusters don’t need to expose it towards clients.
Transactions should behave similarly to consumer groups. Since transactions are more cross-cutting since there are markers in user topics and in the __consumer-offsets topic as well. Therefore logically it makes sense to handle transactions similarly to consumer groups. This means that transactions will be linked to virtual clusters (to the same as the user creating them) and the transaction ID will be translated to a unique ID on the broker side.
We would like to extend the current authorization model of Kafka to accommodate the needs of virtual clusters. Kafka already provides a CLUSTER ACL resource but currently it is hardcoded with the “kafka-cluster” value and it’s not possible to specify a different one. By generalizing and elevating it to a proper resource, we could actually be able to enable administrators to create clusters and properly manage ACLs within resources. Since many inter-broker protocols already depend on this resource, we would leave the pre-defined “kafka-cluster” as it is for internal protocols. Furthermore any user or client who has privileges on this resource, would have global access on the whole Kafka cluster for compatibility reasons, so on upgrades, everything stays the same.
By creating CRUD type permissions on cluster resources, we can form the basics of virtual cluster management. VC Admins can be created with ACLs and also a global admin by setting the CRUD ACLs on the “kafka-cluster” resource.
The figure above represents the ACL structure of the cluster (and not any physical one). The topmost ACL is the “kafka-cluster” which defines cluster-wide ACLs that apply everywhere. In this, there are all ACLs that come from pre-upgrade but in addition to that, there can be virtual cluster ACLs defined that apply only on resources, users and clients linked to that virtual cluster.
A cluster-wide admin is defined by being able to give permissions to other admins in the whole cluster. This can be achieved by granting CREATE_ACLS, DELETE_ACLS and DESCRIBE_ACLS on the “kafka-cluster” resource. Lower level administrators in virtual clusters wouldn’t need to be allowed to perform operations on the whole cluster, just in their assigned virtual clusters, therefore they would be granted access to CREATE_ACLS, DELETE_ACLS and DESCRIBE_ACLS on their respective virtual clusters.
We have to reinterpret some of the actions that can be performed on a CLUSTER resource since we would be able to specify the cluster as well. Generally, with protocols that are meant to be for inter-broker communication, we wouldn’t allow changing them from the “kafka-cluster” cluster, since that would compromise the behavior of the whole cluster.
On the cluster resource, these are the following protocols that can be used:
New APIs:
No protocols can be associated with the “Read” action on virtual clusters.
No protocols can be associated with the “Write” action on virtual clusters.
By properly setting the virtual cluster’s *_VIRTUAL_CLUSTER ACLs, one should be able to configure users to have an administrator role for a given virtual cluster. A super admin that is responsible for managing the physical cluster will be a user who has the appropriate *_CLUSTER ACLs on the “kafka-cluster” cluster resource. This admin will be able to grant further permissions to lower level administrators.
We would like to introduce a new family of quotas, called “virtual-clusters”. These are similar in behavior to other types of quotas and applied from top to bottom in decreasing priority. Quotas outside “virtual-clusters” would be global and applied in every cluster. We wouldn’t allow the scenario for setting quotas for the “kafka-cluster” virtual cluster as that is a reserved name and would introduce ambiguity as it would have the same effect as using the non-virtual-clusters quota.
There are two exceptions to this above. When we define a virtual cluster, we would like to have an upper limit on virtual clusters, so not one virtual cluster and their resources could monopolize the whole cluster’s resources. The first one would put a limit on a specific virtual cluster and all quotas defined in that virtual cluster, while the second would give a default value on virtual clusters quota if the first one isn’t specified.
/config/virtual-clusters/<virtual-cluster>
/config/virtual-clusters/<default>
Besides the above, other configurations are decreasing in priority from top to bottom:
/config/virtual-clusters/<virtual-cluster>/users/<user>/clients/<client-id>
/config/virtual-clusters/<virtual-cluster>/users/<user>/clients/<default>
/config/virtual-clusters/<virtual-cluster>/users/<user>
/config/virtual-clusters/<virtual-cluster>/users/<default>/clients/<client-id>
/config/virtual-clusters/<virtual-cluster>/users/<default>/clients/<default>
/config/virtual-clusters/<virtual-cluster>/users/<default>
/config/virtual-clusters/<virtual-cluster>/clients/<client-id>
/config/virtual-clusters/<virtual-cluster>/clients/<default>
/config/users/<user>/clients/<client-id>
/config/users/<user>/clients/<default>
/config/users/<user>
/config/users/<default>/clients/<client-id>
/config/users/<default>/clients/<default>
/config/users/<default>
/config/clients/<client-id>
/config/clients/<default>
Kafka Connect is a fairly independent framework from core Kafka and lacks some capabilities. On an operational level from Kafka’s point of view, we have connect-offsets, connect-configs and connect-status internal topics. We see no reason against using them in virtual clusters. This makes it possible to host multiple Connect clusters with the same Kafka cluster.
Mirror Maker is very similar to Connect as it builds on that. Similarly to connect, one should be able to configure their MM2 both for an existing Connect cluster using VCs, but also in dedicated mode, where MM2 runs its own Connect cluster. We wouldn’t like to add exceptions to MM2-internal topics, so users should be able to run multiple dedicated MM2 clusters on Kafka too.
Streams applications could be placed inside virtual clusters. This would mean that topics created by streams would be created in a virtual cluster already. From this perspective, they would fit virtual clusters well.
When users upgrade to a virtual cluster compatible Kafka version, everything stays the same as it was before. There will be no virtual cluster resources created and the only cluster ACL is the “kafka-cluster” (and this is also true for a fresh installation).
Users should first create an administrator account that is a super admin and manage virtual clusters. Then with the appropriate command line tools it can create virtual clusters and set the ACLs as it fits to them. After this, they link their existing topics to these newly created virtual clusters and assign users to virtual clusters. Virtual clusters can co-exist with pre-upgrade topics and other resources, so the migration wouldn’t cause operational downtime.
One common use-case is when users use prefixes to simulate tenants. A typical example would be a topic called “emea.eu.sales.SalesStream” where “emea.eu.sales” prefix could be a user space for a company’s sales department. Then a prefix ACL could be enforced with a CreateTopicPolicy to control what kind of topics users of the sales department can create.
So in this setup clients will always address the prefixed name of the topic whereas in this proposal we suggest that clients should use just the topic’s name without any organizational prefixes, so it becomes transparent, but it also contradicts the previous usage pattern. This however can be mitigated easily by the following steps:
Users who are allowed for CREATE_VIRTUAL_CLUSTER on CLUSTER resources, will be able to create virtual clusters.
{
"apiKey": 91,
"type": "request",
"listeners": ["broker", "controller"],
"name": "CreateVirtualClustersRequest",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "VirtualClusters", "type": "[]CreatableVirtualCluster", "versions": "0+",
"about": "The virtual clusters to create.", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "mapKey": true,
"about": "The name of the virtual cluster." }
]},
{ "name": "timeoutMs", "type": "int32", "versions": "0+", "default": "60000",
"about": "How long to wait in milliseconds before timing out the request." },
{ "name": "validateOnly", "type": "bool", "versions": "1+", "default": "false", "ignorable": false,
"about": "If true, check that the virtual clusters can be created as specified, but don't create anything." }
]
}
|
{
"apiKey": 91,
"type": "response",
"name": "CreateVirtualClustersResponse",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "VirtualClusters", "type": "[]CreatableVirtualClustersResult", "versions": "0+",
"about": "Results for each topic we tried to create.", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "mapKey": true,
"about": "The name of the virtual cluster." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "ErrorMessage", "type": "string", "versions": "1+", "nullableVersions": "0+", "ignorable": true,
"about": "The error message, or null if there was no error." }
]}
]
} |
Users who are allowed for ALTER_VIRTUAL_CLUSTER on CLUSTER resources, will be able to modify resource connections in a virtual cluster. This means that they can add and remove topic links, associate users and clients with the CLUSTER resource and also disassociate them.
{
"apiKey": 92,
"type": "request",
"listeners": ["broker", "controller"],
"name": "AlterVirtualClustersRequest",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "VirtualCluster", "type": "[]VirtualClusterData", "versions": "0+",
"about": "The virtual clusters to Alter.", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "mapKey": true,
"about": "The name of the virtual cluster." },
{ "name": "Resources", "type": "[]VirtualClusterResource", "versions": "0+",
"about": "The updates for each resource.", "fields": [
{ "name": "ResourceType", "type": "int8", "versions": "0+", "mapKey": true,
"about": "The resource type." },
{ "name": "ResourceName", "type": "string", "versions": "0+", "mapKey": true,
"about": "The resource name." },
{ "name": "ResourceOperation", "type": "int8", "versions": "0+", "mapKey": true,
"about": "The operation on the resource." }
]}
]},
{ "name": "timeoutMs", "type": "int32", "versions": "0+", "default": "60000",
"about": "How long to wait in milliseconds before timing out the request." },
{ "name": "validateOnly", "type": "bool", "versions": "1+", "default": "false", "ignorable": false,
"about": "If true, check that the virtual clusters can be created as specified, but don't create anything." }
]
}
|
Users who are allowed for DELETE_VIRTUAL_CLUSTER on CLUSTER resources, will be able to delete a virtual cluster. A virtual cluster can only be deleted when it’s empty and no users, clients and topics are associated with it.
Users who are allowed for LIST_VIRTUAL_CLUSTERS on CLUSTER resources, will be able to retrieve a list of virtual clusters in the Kafka cluster.
Users who are allowed for DESCRIBE_VIRTUAL_CLUSTER on CLUSTER resources, will be able to describe a virtual cluster. The listing would include the topics, users, clients, groups and transactions assigned to that virtual cluster.
Briefly list any new interfaces that will be introduced as part of this proposal or any existing interfaces that will be removed or changed. The purpose of this section is to concisely call out the public contract that will come along with this feature.
A public interface is any change to the following:
Binary log format
The network protocol and api behavior
Any class in the public packages under clientsConfiguration, especially client configuration
org/apache/kafka/common/serialization
org/apache/kafka/common
org/apache/kafka/common/errors
org/apache/kafka/clients/producer
org/apache/kafka/clients/consumer (eventually, once stable)
Monitoring
Command line tools and arguments
Describe the new thing you want to do in appropriate detail. This may be fairly extensive and have large subsections of its own. Or it may be a few sentences. Use judgement based on the scope of the change.
Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?
If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.