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.
StatusCurrent state: "Under Discussion"
...
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
MotivationSetting up a Kafka Connect cluster requires provisioning three internal topics:
...
- Drastically Reduces Topic Proliferation – Eliminates unnecessary topic creation.
- Faster Kafka Connect Cluster Deployment – No waiting for new topic provisioning.
- Large Enterprises with Multiple Teams Using Kafka Connect
- Scenario: In large organisations, multiple teams manage different Kafka Connect clusters for various data pipelines.
- Benefit: Instead of waiting for new internal topics to be provisioned each time a new cluster is deployed, teams can immediately start using pre-existing shared topics, reducing lead time and improving efficiency.
- Cloud-Native & Kubernetes-Based Deployments
- Scenario: Many organisations deploy Kafka Connect in containerised environments (e.g., Kubernetes), where clusters are frequently scaled up/down or recreated dynamically.
- Benefit: Since internal topics are already available, new clusters can spin up instantly, without waiting for topic provisioning or Kafka ACL approvals.
- How this will help different organisations:
- Lower Operational Load – Reduces disk-intensive cleanup operations.
- Broker resource utilization is expected to decrease by approximately 20%, primarily due to reduced partition count and metadata overhead. This optimization can enable further cluster downscaling, contributing directly to lower infrastructure costs (e.g., fewer brokers, reduced EBS storage footprint, and lower I/O throughput).
- Administrative overhead and monitoring complexity are projected to reduce by 30%, due to:
- Fewer topics to configure, monitor, and apply retention/compaction policies to.
- Reduced rebalancing operations during cluster scale-in or scale-out events.
- Minimized data movement and replication workload in failure scenarios (e.g., EBS volume replacement), as fewer partitions are involved.
- Considering the current load
- Simplified Management – Less overhead in monitoring and maintaining internal topics.
| Anchor |
|---|
| Why Share Kafka Connect Internal Topics Across Clusters? |
|---|
| Why Share Kafka Connect Internal Topics Across Clusters? |
|---|
|
Why Share Kafka Connect Internal Topics Across Clusters?
1. Operational Efficiency & Reduced Overhead
...
- Instead of monitoring and securing numerous individual topics, SRE teams can:
- Implement unified monitoring for fewer topics.
- Apply standardized ACLs & RBAC policies.
- Simplify compliance auditing (fewer entities to track).
| Anchor |
|---|
| Why this will not effect Kafka Connect Startup |
|---|
| Why this will not effect Kafka Connect Startup |
|---|
|
Why this will not effect Kafka Connect Startup- The growth of the shared topic will be limited by O(n), where “n” is the number of clusters.
- Kafka already shares the “__Consumer_Offset” topic across all the consumer groups on a single kafka cluster and very efficiently manages even if the number of consumer groups keeps on scaling linearly and this motivates to share the connect internal topics without adding any appreciable time delay in starting connect at least to an allowed limit.an allowed limit.
| Anchor |
|---|
| BenchMarking with topic shared across 20 clusters each having 5 Sink jobs |
|---|
| BenchMarking with topic shared across 20 clusters each having 5 Sink jobs |
|---|
|
BenchMarking with topic shared across 20 clusters each having 5 Sink jobs- Each of the Config Topic and Status Topic has roughly 7-8 k records.
- Config Topic → 32MB (segment size)
- Status Topic → 3MB (segment size)
- In NON-TOPIC-SHARING mode size remains in KBs. Still:
- Connect framework roughly took only:
- 400-500 ms for reading the entire Status Topic.
- around 500 ms for reading entire Config topic (Slightly more time than status as config has high payload size)
- Also one key thing to consider is this reading only happens while starting and once that is done there is NOeffect on connect resource api’s.
| Anchor |
|---|
| Public Interfaces |
|---|
| Public Interfaces |
|---|
|
Public InterfacesNew properties
- A single worker-level configuration property will be added for distributed mode:
Name |
Type |
Default |
Importance | Description |
connect.management.topic.sharing.support |
String
|
disabled |
HIGH | Whether to enable sharing of Kafka Connect management topic sharing across clusters by using cluster group ID to filter the different states along the connect journey.
To enable the connect management topic sharing on a new cluster, set this property to "enabled" |
...
| Property Name | Purpose | Default |
| source.cluster.names | list of cluster names which are to be migrated, empty in case migrating old client or migrating a particular job. | Empty List |
| destination.cluster.name | destination cluster name to which the migration is to be performed. | No default value as this will be required in all the cases. |
| topic.type | type of the topic which is being migrated | No default value as this is always needed |
| migrate.job | config to control whether only a particluar job is to be migrated | FALSE |
| job.name | name of the job in case a particular job is being migrated | empty string |
| old.client | config to control the migration of an old client to a new client | True, as we assume that migration will be primarily needed for an old client to the new client |
| Anchor |
|---|
| Proposed Changes |
|---|
| Proposed Changes |
|---|
|
Proposed ChangesAs part of this KIP, we are proposing the add the “ClusterGroupId” as prefix to the existing keys. So for ex:
If the current key for “connector status” for connector “A” for Cluster "B” is :
...
The intention was just not to choose “-” as it is already being used so the we can preserve all the logic with minimal changes.
| Anchor |
|---|
| Compatibility, Deprecation, and Migration Plan |
|---|
| Compatibility, Deprecation, and Migration Plan |
|---|
|
Compatibility, Deprecation, and Migration PlanCompatibility
- The changes are backward compatible, meaning anyone can freely and smoothly upgrade to the new client to which these changes will be added, without any issues. As the changes are dependent on whether it can extract the “CLUSTER_ID” from the key, in case if it doesn’t it goes into the normal flow.
- A cluster using the older client can still share the management topics with the cluster upgraded to the newer client.
...
- Old clients migrating to cluster operating with new client:
- Can choose to just mirror all the connect topics to the topics of the cluster where migration is intended if that cluster is not sharing the topic with any older clients.
- If the cluster where migration is desired is already hosting other cluster with older client then “
KafkaConnectTopicMigrator” can be used to do the migration.
- New cluster migrating to another new cluster:
- “KafkaConnectTopicMigrator” can be used along as transformer along with kafka to kafka mirror.
Test Plan- Unit and integration tests will be added for both the forward and backward compatibility
| Anchor |
|---|
| Rejected Alternatives |
|---|
| Rejected Alternatives |
|---|
|
Rejected AlternativesAnother way to pass cluster information was to put the cluster information in the header. But there were two challenges with this approach:
...