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.

Status

Current state: "Under Discussion"

Discussion thread: will be updated

JIRA: here

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Setting up a Kafka Connect cluster requires provisioning three internal topics:

  1. Offset Storage Topic – Tracks offsets of source connectors.
  2. Status Storage Topic – Maintains the state of connectors and tasks.
  3. Config Storage Topic – Stores connector configurations.

This design choice simplifies migration—enabling seamless replication of management topics across regions. However, it introduces operational overhead:

While each cluster only requires three topics, their cumulative impact grows significantly as more connect clusters are deployed.

But as these topics have very light traffic and are compacted, instead of provisioning dedicated topics for every cluster, Kafka Connect clusters can share internal topics across multiple deployments. This brings immediate benefits:


Why Share Kafka Connect Internal Topics Across Clusters?

1. Operational Efficiency & Reduced Overhead

2. Simplified Cluster Management & Deployment

3. Resource Optimization & Cost Reduction

4. Improved Topic Governance & Consistency

5. Reduced Metadata Load on Kafka Brokers

6. Streamlined Monitoring & Security Controls

Why this will not effect Kafka Connect Startup

BenchMarking with topic shared across 20 clusters each having 5 Sink jobs

Public Interfaces

New properties

public KafkaConfigBackingStore(Converter converter, DistributedConfig config, WorkerConfigTransformer configTransformer, Supplier<TopicAdmin> adminSupplier, String clientIdBase, String clusterGroupId) {
        this(converter, config, configTransformer, adminSupplier, clientIdBase, clusterGroupId, Time.SYSTEM);
    }

    KafkaConfigBackingStore(Converter converter, DistributedConfig config, WorkerConfigTransformer configTransformer, Supplier<TopicAdmin> adminSupplier, String clientIdBase, String clusterGroupId, Time time) {
        ...
        this.clusterGroupId = Optional.ofNullable(clusterGroupId);
    }


public KafkaStatusBackingStore(Time time, Converter converter, String clusterGroupId) {
        this(time, converter, null, "connect-distributed-", clusterGroupId);
    }

 public KafkaStatusBackingStore(Time time, Converter converter, Supplier<TopicAdmin> topicAdminSupplier, String clientIdBase, String clusterGroupId) {
        this.time = time;
        this.converter = converter;
        this.tasks = new Table<>();
        this.connectors = new HashMap<>();
        this.topics = new ConcurrentHashMap<>();
        this.topicAdminSupplier = topicAdminSupplier;
        this.clientId = Objects.requireNonNull(clientIdBase) + "statuses";
        this.clusterGroupId = Optional.ofNullable(clusterGroupId);
    }


public OffsetStorageReaderImpl(OffsetBackingStore backingStore, String namespace,
                                   Converter keyConverter, Converter valueConverter, String clusterGroupId) {
        this.backingStore = backingStore;
        this.namespace = namespace;
        this.keyConverter = keyConverter;
        this.valueConverter = valueConverter;
        this.closed = new AtomicBoolean(false);
        this.offsetReadFutures = new HashSet<>();
        this.clusterGroupId = Optional.ofNullable(clusterGroupId);
    }


public OffsetStorageWriter(OffsetBackingStore backingStore, String namespace, String clusterGroupId, Converter keyConverter, Converter valueConverter) {
        this.backingStore = backingStore;
        this.namespace = namespace;
        this.keyConverter = keyConverter;
        this.valueConverter = valueConverter;
        this.clusterGroupId = Optional.ofNullable(clusterGroupId);
    }


    private Optional<List<String>> parseKey(ConsumerRecord<String, byte[]> record) {
        String key = record.key();
        if (key == null || key.isEmpty()) {
            log.error("Empty or null key provided in record: {}", record);
            return Optional.empty();
        }

        List<String> parts = Arrays.asList(key.split("\\" + CLUSTER_GROUP_SEPARATOR));
        if (parts.size() != 2) {
            log.error("Invalid key format: '{}'. Key should be in the format 'clusterGroupId.actualKey'", key);
            return Optional.empty();
        }

        return Optional.of(parts);
    }

    Optional<String> clusterGroupIdFromRecord(ConsumerRecord<String, byte[]> record) {
        return parseKey(record).map(parts -> parts.get(0));
    }

    Optional<String> keyFromRecord(ConsumerRecord<String, byte[]> record) {
        return parseKey(record).map(parts -> parts.get(1));
    }



KafkaConnectTopicMigrator


We have created a new Transformer to handle all type of the migrations and the consumer need not worry. The migration will be similar to how we migrate in the current world with only the addition of a very light weight transformer handling the cluster transformation logic. The configs are as below:

Property NamePurposeDefault
source.cluster.nameslist of cluster names which are to be migrated, empty in case migrating old client or migrating a particular job.Empty List
destination.cluster.namedestination cluster name to which the migration is to be performed.No default value as this will be required in all the cases.
topic.typetype of the topic which is being migratedNo default value as this is always needed
migrate.jobconfig to control whether only a particluar job is to be migratedFALSE
job.namename of the job in case a particular job is being migratedempty string
old.clientconfig to control the migration of an old client to a new clientTrue, as we assume that migration will be primarily needed for an old client to the new client

Proposed Changes

As 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 :

status-connector-A

then with this change it will be modified to the following:

B.status-connector-A

While processing the “parseKey” will split this on the “.” which will result in the following:

Cluster Group Id derived from the record ("B") will be validated against the actual ClusterGroupId of the cluster and if matched the original logic will continue else the record will be skipped for this cluster.

Why “.” is chosen:

The intention was just not to choose “-” as it is already being used so the we can preserve all the logic with minimal changes.

Compatibility, Deprecation, and Migration Plan

Compatibility

    1. 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.
    2. A cluster using the older client can still share the management topics with the cluster upgraded to the newer client.
Kafka Client AKafka Client BCan Share Topic?
OldOldNo
OldNewYes
NewOldYes
NewNewYes

Deprication

    1. No interfaces or api’s are being deprecated as part of this change.

Migration

    1. Old clients migrating to cluster operating with new client:
      1. 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.
      2. If the cluster where migration is desired is already hosting other cluster with older client then “
        KafkaConnectTopicMigrator” can be used to do the migration.
    2. New cluster migrating to another new cluster:
      1. “KafkaConnectTopicMigrator” can be used along as transformer along with kafka to kafka mirror.

Test Plan

  1. Unit and integration tests will be added for both the forward and backward compatibility

Rejected Alternatives

Another way to pass cluster information was to put the cluster information in the header. But there were two challenges with this approach:

  1. This breaks the backward incompatibility as the older client has no way to automatically work with the newer clients with topic sharing feature turned on.
  2. This required a lot more code changes and hence a larger testing and review cycle.