Code Block

language	java
title	CreateVirtualClustersOptions

@InterfaceStability.Evolving
public class CreateVirtualClustersOptions extends AbstractOptions<CreateVirtualClustersOptions> {

    private boolean validateOnly = false;
    private boolean retryOnQuotaViolation = true;

    public CreateVirtualClustersOptions timeoutMs(Integer timeoutMs) {
        this.timeoutMs = timeoutMs;
        return this;
    }

    public CreateVirtualClustersOptions validateOnly(boolean validateOnly) {
        this.validateOnly = validateOnly;
        return this;
    }

    public boolean shouldValidateOnly() {
        return validateOnly;
    }

    public CreateVirtualClustersOptions retryOnQuotaViolation(boolean retryOnQuotaViolation) {
        this.retryOnQuotaViolation = retryOnQuotaViolation;
        return this;
    }

    public boolean shouldRetryOnQuotaViolation() {
        return retryOnQuotaViolation;
    }
}

Public Interfaces

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
Anything else that will likely break existing users in some way when they upgrade

Proposed Changes

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.

Compatibility, Deprecation, and Migration Plan

What impact (if any) will there be on existing users?
If we are changing behavior how will we phase out the older behavior?
If we need special migration tools, describe them here.
When will we remove the existing behavior?

Test Plan

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?

Rejected Alternatives

Code Block

language	java
title	Admin class new method

CreateVirtualClustersResult createVirtualClusters(Collection<NewVirtualCluster> virtualClusters, CreateVirtualClustersOptions options);

Alter Virtual Clusters

Code Block

language	java
title	AlterVirtualClustersResult

@InterfaceStability.Evolving
public class AlterVirtualClustersResult {

    private final Map<String, KafkaFuture<CreateVirtualClustersResponseData.CreatableVirtualClustersResult>> futures;

    public AlterVirtualClustersResult(Map<String, KafkaFuture<CreateVirtualClustersResponseData.CreatableVirtualClustersResult>> futures) {
        this.futures = futures;
    }

    public Map<String, KafkaFuture<Void>> values() {
        return futures.entrySet().stream()
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().thenApply(v -> null)));
    }

    public KafkaFuture<Void> all() {
        return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0]));
    }
}

Code Block

language	java
title	AlterVirtualClustersOptions

@InterfaceStability.Evolving
public class AlterVirtualClustersOptions extends AbstractOptions<AlterVirtualClustersOptions> {

    private boolean validateOnly = false;
    private boolean retryOnQuotaViolation = true;

    public AlterVirtualClustersOptions timeoutMs(Integer timeoutMs) {
        this.timeoutMs = timeoutMs;
        return this;
    }

    public AlterVirtualClustersOptions validateOnly(boolean validateOnly) {
        this.validateOnly = validateOnly;
        return this;
    }

    public boolean shouldValidateOnly() {
        return validateOnly;
    }


    public AlterVirtualClustersOptions retryOnQuotaViolation(boolean retryOnQuotaViolation) {
        this.retryOnQuotaViolation = retryOnQuotaViolation;
        return this;
    }

    public boolean shouldRetryOnQuotaViolation() {
        return retryOnQuotaViolation;
    }
}

Code Block

language	java
title	VirtualClusterAlteration

@InterfaceStability.Evolving
public class VirtualClusterAlteration {

    public enum ResourceChangeType {
        ADD, REMOVE
    }

    public enum ResourceType {
        USER, TOPIC, GROUP, TRANSACTIONAL_ID
    }

    private String virtualClusterName;
    private List<VirtualClusterEntityChange> changes;

    public VirtualClusterAlteration(String virtualClusterName, List<VirtualClusterEntityChange> changes) {
        this.virtualClusterName = virtualClusterName;
        this.changes = changes;
    }

    public String virtualClusterName() {
        return virtualClusterName;
    }

    public List<VirtualClusterEntityChange> changes() {
        return changes;
    }

    public static class VirtualClusterEntityChange {

        private ResourceType resourceType;
        private String entityName;
        private ResourceChangeType changeType;

        public VirtualClusterEntityChange(ResourceType resourceType, String entityName, ResourceChangeType changeType) {
            this.resourceType = resourceType;
            this.entityName = entityName;
            this.changeType = changeType;
        }

        public ResourceType entityType() {
            return resourceType;
        }

        public String entityName() {
            return entityName;
        }

        public ResourceChangeType changeType() {
            return changeType;
        }
    }
}

Code Block

language	java
title	Admin class new method

   AlterVirtualClustersResult alterVirtualClusters(Collection<VirtualClusterAlteration> alterations, AlterVirtualClustersOptions options);

Delete Virtual Clusters

Code Block

language	java
title	DeleteVirtualClustersOptions

@InterfaceStability.Evolving
public class DeleteVirtualClustersOptions extends AbstractOptions<DeleteVirtualClustersOptions> {

    private boolean validateOnly = false;
    private boolean retryOnQuotaViolation = true;

    public DeleteVirtualClustersOptions timeoutMs(Integer timeoutMs) {
        this.timeoutMs = timeoutMs;
        return this;
    }

    public DeleteVirtualClustersOptions validateOnly(boolean validateOnly) {
        this.validateOnly = validateOnly;
        return this;
    }

    public boolean shouldValidateOnly() {
        return validateOnly;
    }

    public DeleteVirtualClustersOptions retryOnQuotaViolation(boolean retryOnQuotaViolation) {
        this.retryOnQuotaViolation = retryOnQuotaViolation;
        return this;
    }

    public boolean shouldRetryOnQuotaViolation() {
        return retryOnQuotaViolation;
    }
}

Code Block

language	java
title	DeleteVirtualClustersResult

@InterfaceStability.Evolving
public class DeleteVirtualClustersResult {

    private final Map<String, KafkaFuture<DeleteVirtualClustersResponseData.DeletableVirtualClusterResult>> futures;

    public DeleteVirtualClustersResult(Map<String, KafkaFuture<DeleteVirtualClustersResponseData.DeletableVirtualClusterResult>> futures) {
        this.futures = futures;
    }

    public Map<String, KafkaFuture<Void>> values() {
        return futures.entrySet().stream()
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().thenApply(v -> null)));
    }

    public KafkaFuture<Void> all() {
        return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0]));
    }
}

Code Block

language	java
title	Admin class new method

   DeleteVirtualClustersResult deleteVirtualClusters(Collection<String> virtualClusterNames, DeleteVirtualClustersOptions options);

List Virtual Clusters

Code Block

language	java
title	ListVirtualClustersResult

@InterfaceStability.Evolving
public class ListVirtualClustersResult {

    private final KafkaFuture<List<String>> futures;

    public ListVirtualClustersResult(KafkaFuture<List<String>> futures) {
        this.futures = futures;
    }

    public KafkaFuture<List<String>> listing() {
        return futures;
    }
}

Code Block

language	java
title	Admin class new method

ListVirtualClustersResult listVirtualClusters();

Describe Virtual Clusters

Code Block

language	java
title	DescribeVirtualClustersOptions

@InterfaceStability.Evolving
public class DescribeVirtualClustersOptions extends AbstractOptions<DescribeVirtualClustersOptions> {
}

Code Block

language	java
title	DescribeVirtualClustersResult

@InterfaceStability.Evolving
public class DescribeVirtualClustersResult {

    private Map<String, KafkaFuture<VirtualClusterDescription>> futures;

    public DescribeVirtualClustersResult(Map<String, KafkaFuture<VirtualClusterDescription>> futures) {
        this.futures = futures;
    }

    public Map<String, KafkaFuture<VirtualClusterDescription>> values() {
        return futures;
    }

    public KafkaFuture<Void> all() {
        return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0]));
    }
}

Code Block

language	java
title	VirtualClusterDescription

@InterfaceStability.Evolving
public class VirtualClusterDescription {

    public static class TopicLink {
        private String linkName;
        private String topicName;

        public TopicLink(String linkName, String topicName) {
            this.linkName = linkName;
            this.topicName = topicName;
        }

        public String linkName() {
            return linkName;
        }

        public String topicName() {
            return topicName;
        }
    }

    private String name;
    private List<TopicLink> topicLinks;
    private List<String> groupLinks;
    private List<String> userLinks;
    private List<String> clientLinks;
    private List<String> transactionalIdLinks;

    public VirtualClusterDescription(String name, List<TopicLink> topicLinks, List<String> groupLinks, List<String> userLinks, List<String> clientLinks, List<String> transactionalIdLinks) {
        this.name = name;
        this.topicLinks = topicLinks;
        this.groupLinks = groupLinks;
        this.userLinks = userLinks;
        this.clientLinks = clientLinks;
        this.transactionalIdLinks = transactionalIdLinks;
    }

    public String name() {
        return name;
    }

    public List<TopicLink> topicLinks() {
        return topicLinks;
    }

    public List<String> groupLinks() {
        return groupLinks;
    }

    public List<String> userLinks() {
        return userLinks;
    }

    public List<String> clientLinks() {
        return clientLinks;
    }

    public List<String> transactionalIdLinks() {
        return transactionalIdLinks;
    }
}

Code Block

language	java
title	Admin class new method

DescribeVirtualClustersResult describeVirtualClusters(Collection<String> virtualClusterIds, DescribeVirtualClustersOptions options);

Config Changes

At this point there may be no need for new configs. It could be possible to add configuration for setting up virtual clusters through static configuration, however due to the possible number of topics and users it can get quite extensive. Therefore users should rely on command line tools or the Admin API when migrating to virtual clusters.

Tooling Changes

Overall we would like to introduce a single new command line tool to handle all these interface changes. This would be the virtual-clusters.sh tool.

The following subcommands would be available for the command: create, alter, list, describe, delete. Each of these would implement the previously described functionality. Each command is designed to carry out a single action for simplicity. For instance deleting multiple virtual clusters with a single execution of the command wouldn’t be available as the Admin API could provide better access for more complicated actions.

Subcommands

create

With the below command, an administrator would create a virtual cluster.

Code Block

language	bash
title	create

virtual-clusters.sh create
  --bootstrap-server localhost:9092
  --virtual-cluster my-virtual-cluster

alter

With the alter command below, one could create a resource link to the virtual cluster or can remove one. The command would be able to execute only one addition or removal of a selected resource.

Code Block

language	bash
title	alter

virtual-clusters.sh alter
  --bootstrap-server localhost:9092
  --virtual-cluster my-virtual-cluster
  [--add | --remove]
  [--user PRINCIPAL | --client CLIENT_ID | [--topic TOPIC --link LINK] | --group GROUP | --transactional-id TRANSACTIONAL_ID]

delete

Code Block

language	bash
title	delete

virtual-clusters.sh delete
  --bootstrap-server localhost:9092
  --virtual-cluster my-virtual-cluster

list

Code Block

language	bash
title	list

virtual-clusters.sh list
  --bootstrap-server localhost:9092

describe

Code Block

language	bash
title	describe

virtual-clusters.sh describe
  --bootstrap-server localhost:9092
  --virtual-cluster my-virtual-cluster

Examples

Code Block

language	bash
title	Examples

# create a new virtual cluster
virtual-clusters.sh create
  --bootstrap-server localhost:9092
  --virtual-cluster my-virtual-cluster

# create a topic link in it
virtual-clusters.sh alter
  --bootstrap-server localhost:9092
  --virtual-cluster my-virtual-cluster
  --add
  --topic my-topic --link test-topic

# assign a user to the virtual cluster
virtual-clusters.sh alter
  --bootstrap-server localhost:9092
  --virtual-cluster my-virtual-cluster
  --add
  --user jane-doe

# list the virtual clusters
virtual-clusters.sh list
  --bootstrap-server localhost:9092

# describe the virtual cluster
virtual-clusters.sh describe
  --bootstrap-server localhost:9092
  --virtual-cluster my-virtual-cluster

# delete the virtual-cluster
virtual-clusters.sh delete
  --bootstrap-server localhost:9092
  --virtual-cluster my-virtual-cluster

Metadata Layer

Although it isn’t an API towards users, we feel important to discuss this part as it affects the core of Kafka. We plan to store the data of virtual clusters in the metadata layer similarly to topics and configurations for instance.

Therefore the first step in this, is that there will be a VirtualClusterImage that holds all properties of the VirtualCluster itself along with all links/associations. With this we can define the delta classes that will symbolize the changes in virtual clusters.

Finally VirtualClustersImage will be put into the MetadataImage to make it part of the metadata.

Code Block

language	java
title	Metadata Layer

/**
 * Represents the image of all virtual clusters.
 */
public class VirtualClustersImage {
    private Map<String, VirtualClusterImage> virtualClusterImages;
}

/**
 * Represents the image of a single virtual cluster.
 */
public class VirtualClusterImage {

    public static class TopicLink {
        private String topicName;
        private String linkName;
    }

    private String name;
    private List<TopicLink> topics;
    private List<String> users;
    private List<String> clients;
    private List<String> groups;
    private List<String> transactionalIds;
}

/**
 * Represents the delta of virtual clusters on a change in virtual clusters.
 */
public class VirtualClustersDelta {
    private final VirtualClustersImage image;
    private final List<VirtualClusterDelta> clusterChanges;
}

/**
 * Represents the changes of a virtual cluster.
 */
public class VirtualClusterDelta {
    private final VirtualClusterImage image;
    private final List<VirtualClusterImage.TopicLink> topicLinkChanges;
    private final List<String> userChanges;
    private final List<String> clientChanges;
    private final List<String> groupChanges;
    private final List<String> transactionalIdChanges;
}

These images will be used by the VirtualClustersMetadataPublisher to update the metadata similarly to other existing features when required.

Deprecation, Backward Compatibility

Configuration

There are no new configurations and no planned deprecations.

Behavior

A Kafka cluster without any virtual clusters should behave the same as it did before the upgrade to a virtual cluster compatible version. After the upgrade, users will have the chance to set up their ACLs properly and add virtual clusters. This however won’t alter the behavior of the cluster as it will continue to operate as it did before. Users can migrate their clients to virtual clusters as described in a previous section. No behavior will be deprecated.

Interfaces

We add new admin methods without changing the behavior of existing ones. No existing interfaces will be deprecated.

Protocols

We add new protocols without changing the behavior of old ones. No existing protocols will be deprecated.

Testing

We mainly plan to test this in the conventional ways: unit tests for asserting the correctness in a more primitive level, integration tests to assert the correctness of the Admin API, the protocols and the rest of the server side request handling and ductape tests to assert the correctness of the feature as a whole, starting from the command line tools which would spin the Admin APIs, then the protocols and the server side request handling as well. In the ductape tests a whole life cycle would be played through:

Create a virtual cluster through command line
Create links in the virtual cluster
Assign users, clients and other resources to the cluster
Run a consumer and producer to assert the happy path
Remove links and resources from the virtual cluster
Delete the virtual cluster

Documentation Plan

The Kafka multi-tenancy section will be amended with the contents of this KIP upon completion.

Rejected Alternatives

Hierarchical Clusters

While it would allow more complex use cases and we could implement a cluster of clusters, the operational complexity would be much higher. With hierarchical clusters we would need to introduce a separator to address resources in the non-leaf nodes of the cluster tree. There are multiple reasons against this:

The introduction of this notation itself could break backward compatibility as users would need to change their clients to use the fully qualified resource names when addressing resources in virtual clusters.
It would also complicate the ACL structure as we would need to introduce this hierarchy in ACLs and would need to create a much more complicated model to handle the relationships in the hierarchy which in our opinion isn’t worth the price as we may risk keeping backward compatibility or having weird edge cases.
Besides this, in an average organization there aren’t hundreds of teams who use Kafka, so a flat structure would likely satisfy most needs. Users can continue using “.” as a separator, so they could have “us.amer.ecommerce” as their virtual cluster.

KIP-37 Style Namespaces

KIP-37 proposed a similar idea, however since it never got through the discussion phase, it never expanded more on the details. We find that this solution has the drawbacks of the hierarchical clusters and additionally it represents namespaces hardcoded in the log directory structure. This isn’t beneficial as moving or renaming topics would become hard as one would need to replicate the data, which is obviously more costly, the more data there is.

Avoid Binding Users to Virtual Clusters

If we don’t bind users to virtual clusters, then the clients used by those users can address multiple VCs and thus resources like topics and groups could become ambiguous if the same resource is present in multiple places. This could be resolved by introducing a cluster separator and clients would need to reference topics as “cluster:topicname” for instance. This would most likely result in backward incompatibility as one would need to change clients in order to use virtual clusters. We could say that those resources which are unique in a VC could be referenced with relative addresses (where we omit the cluster part of the name), but it would break when a resource with the same name is created elsewhere. Besides this, with the current proposal if a client wants to consume or produce messages to topics in other virtual clusters, an administrator can link those topics to the client’s VC and restrict the access with ACLs only to that client. This provides safe access for the client and a well-defined way to share topics across virtual clusters.

Namespaces instead of Virtual Clusters

Apache Pulsar has a 3 level structure to address topics. First, the topmost layer is called “tenant” which defines configurations for a set of Pulsar clusters as it can span across multiple clusters. It is a higher level concept than the virtual clusters presented here. Then in every tenant there are namespaces which are the administrative units for quotas and configuration. This is mostly equivalent to our virtual clusters. We could have used the “namespace” name here but virtual clusters is a more descriptive name in our opinion.

Second, while we currently don’t want to add the “tenant” level unit of organization, it may be a future development possibilityIf 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.

Space shortcuts

Child pages

Versions Compared

Old Version 3

New Version 4

Key

Public Interfaces

Proposed Changes

Compatibility, Deprecation, and Migration Plan

Test Plan

Rejected Alternatives

Alter Virtual Clusters

Delete Virtual Clusters

List Virtual Clusters

Describe Virtual Clusters

Config Changes

Tooling Changes

Subcommands

create

alter

delete

list

describe

Examples

Metadata Layer

Deprecation, Backward Compatibility

Configuration

Behavior

Interfaces

Protocols

Testing

Documentation Plan

Rejected Alternatives

Hierarchical Clusters

KIP-37 Style Namespaces

Avoid Binding Users to Virtual Clusters

Namespaces instead of Virtual Clusters

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 3

New Version 4

Key

Public Interfaces

Proposed Changes

Compatibility, Deprecation, and Migration Plan

Test Plan

Rejected Alternatives

Alter Virtual Clusters

Delete Virtual Clusters

List Virtual Clusters

Describe Virtual Clusters

Config Changes

Tooling Changes

Subcommands

create

alter

delete

list

describe

Examples

Metadata Layer

Deprecation, Backward Compatibility

Configuration

Behavior

Interfaces

Protocols

Testing

Documentation Plan

Rejected Alternatives

Hierarchical Clusters

KIP-37 Style Namespaces

Avoid Binding Users to Virtual Clusters

Namespaces instead of Virtual Clusters