Status

Current state: Draft

Discussion thread:  

JIRA: KAFKA-20367 - Getting issue details... STATUS

1 Motivation

1.1 The Problem: Scaling Kafka-to-Kafka Pipelines Today

Kafka Connect sink connectors consume from Kafka topics using traditional consumer groups.

In this model, each partition is exclusively assigned to one task. This creates two problems for Kafka-to-Kafka (and Kafka-to-external) pipelines:

1. Scaling is coupled to partition count.

If a topic has 12 partitions, you can run at most 12 sink tasks. I

ncreasing parallelism beyond the partition count requires repartitioning the source topic -- an operationally expensive and disruptive change.

2. Slow tasks block partitions.

If one sink task is slow (e.g., network latency to a downstream system), the records on its assigned partitions back up.

Other idle tasks cannot help because partition ownership is exclusive. This creates head-of-line blocking at the partition level.

3. Rebalance storms cause processing gaps.

When tasks are added, removed, or crash, consumer group rebalances revoke and reassign partitions.

During a rebalance, no task processes records from revoked partitions. With cooperative sticky rebalancing this is mitigated but not eliminated.

1.2 How Share Groups Solve This

Share Groups (KIP-932) introduce queue semantics for Kafka consumers. Unlike consumer groups, Share Groups do not assign partitions exclusively.

Instead, records from a partition are acquired by any available consumer in the group. After processing, the consumer acknowledges the record (ACCEPT, RELEASE, ARCHIEVED, or REJECT).

This provides:

- Elastic scaling independent of partition count.

50 tasks can process a 12-partition topic because records are distributed at the record level, not the partition level.
- No head-of-line blocking.

If one task is slow, acquired records time out and are re-delivered to another task.
- No rebalance disruption.

Share Groups have no partition assignment protocol. Adding or removing tasks does not trigger reassignment of partitions.

1.3 Kafka Connect as the Natural Integration Point

Kafka Connect is the standard framework for building data pipelines into and out of Kafka. Integrating Share Groups into Connect's sink connector

runtime gives every existing sink connector access to queue semantics with a configuration change only -- no connector code modifications required.

1.4 Delivery Semantics

At-least-once is the initial target. Exactly-once requires KIP-1289 (Transactional Acknowledgments for Share Groups) to be implemented and is described as a future phase.

1.5 Error Handling / Delivery Semantics

This KIP integrates with KIP-1191. In Share Group mode, Connect uses `AcknowledgeType.REJECT` for fatal errors and relies on the broker-side DLQ configured by KIP-1191. Retriable failures use `AcknowledgeType.RELEASE`. If a DLQ is configured for the share group, the broker is the single source of DLQ records; Connect does not emit its own DLQ records in this mode.

EO Constraints

For exactly-once, records with pending transactional acknowledgments must not be re-delivered while the transaction is open; KIP-1289 must suppress or renew acquisition locks until commit/abort. Operationally, `share.acquisition.lock.timeout.ms` must exceed worst-case `task.put()` plus transaction commit latency, otherwise duplicates are possible even with EOS.

2. Scope

2.1 In Scope

- Add Share Group support to Kafka Connect sink connectors via a new `WorkerShareSinkTask`, with no changes required to the `SinkTask` API.
- Config-driven enablement (`consumer.override.group.protocol=share`) with per-connector overrides.
- At-least-once delivery for all sinks using Share Groups.
- Exactly-once delivery only for Kafka-to-Kafka pipelines within the same cluster, gated by KIP-1289 and a transactional producer.
- Share Group–specific metrics integrated into existing `sink-task-metrics` group.

2.2 Out of Scope

- Share Group support for source connectors and MirrorMaker 2 (separate effort).
- Exactly-once delivery for cross-cluster Kafka-to-Kafka pipelines (source cluster A → sink cluster B).
- External two-phase commit coordinators or cross-cluster transactional protocols.
- Changes to the public `SinkTask` API or connector implementations.

3. Public Interfaces

3.1 New Configuration Properties

3.1.1 Worker-level configuration (`connect-distributed.properties`)

3.1.2 Connector-level configuration (per-connector JSON)

3.2 New / Modified Java Interfaces

3.2.1 `WorkerShareSinkTask` (new class)

A new internal class in `org.apache.kafka.connect.runtime` that extends `WorkerTask` and drives the `SinkTask` using a `KafkaShareConsumer` instead of a `KafkaConsumer`. This is the core runtime change.

```
// New class: parallel to WorkerSinkTask but backed by ShareConsumer
class WorkerShareSinkTask extends WorkerTask<ConsumerRecord<byte[], byte[]>, SinkRecord> {
    private final ShareConsumer<byte[], byte[]> shareConsumer;
    private final SinkTask task;
    // ...
}
```

Note: The existing `SinkTask` interface is **not modified**. Connectors do not need code changes. The `put(Collection<SinkRecord>)` contract remains the same.

The difference is entirely in the worker runtime:

4. Compatibility, Deprecation, and Migration Plan

5. Test Plan

6. Future Work

7. Rejected Alternatives