Status

Current state: DraftUnder discussion

Discussion thread: here

JIRA:

Jira

server	ASF JIRA
serverId	5aa69414-a9e9-3523-82ec-879b028fb15b
key	KAFKA-20367

1 Motivation

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

Currently, Kafka Connect sink connectors consume from Kafka topics using rely on traditional consumer groups that enforce a strict 1:1 mapping between partitions and tasks. T

In this his model , each partition is exclusively assigned to one task. This creates two problems for Kafka-to-Kafka (and Kafka-to-external) pipelinesis often incompatible with unordered message processing and creates three primary bottlenecks for task queue workloads:

1. Partition-Coupled Scaling is coupled to : Parallelism is hard-limited by the partition count.

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

...

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

...

2. Head-of-Line Blocking: Because partition ownership is exclusive, a single slow task—often caused by downstream latency—stalls all subsequent records in its assigned partitions

3. Rebalance-Driven Gaps: Adding or removing tasks triggers "rebalance storms."

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 Elastic Scaling: Decouples parallelism from partition count,
- No Head-of-Line Blocking: Supports unordered message processing; if a task slows down, records time out and are re-delivered to another taskredelivered to available workers.
- 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

...

Seamless Scaling: Eliminates "rebalance storms" by removing the partition assignment protocol, ensuring zero downtime during task membership changes.

Note: The share groups are only suitable for connectors with idempotent, order-independent processing.

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

...

(What we are building)

New Task Type: Introducing WorkerShareSinkTask to handle Share Group logic without changing existing connector code.
Flexible Activation: Toggle queue semantics globally or per-connector via consumer.override.group.protocol=

...

share.

...

Delivery Guarantees: * At-least-once

...

: Standard support for all

...

sink types.

...

- Exactly-once

...

- : Supported for same-cluster Kafka-to-Kafka

...

- paths (requires KIP-1289

...

- ).

...

Observability: New Share

...

Group metrics (acquisition, release, and rejection rates) integrated into the existing

...

sink-task-

...

metrics group.

2.2 Out of Scope

...

(Future/Separate efforts)

Source Connectors: Share Group support

...

is currently for Sinks only (Source support and MirrorMaker 2

...

are excluded).
Cross-Cluster EOS: Exactly-once delivery

...

between different Kafka clusters is not supported in this phase.
API Changes: No modifications will be made to the public SinkTask Java API or individual connector codebases.
Complex Transactions: External 2PC coordinators and cross-cluster transactional protocols are not addressed.

...

3. Public Interfaces

3.1 New Configuration Properties

...

Worker-

...

Level (

...

`connect-distributed.`

...

`properties`)

consumer.group.protocol

...

: Set to share to enable KafkaShareConsumer globally for all sink tasks (Default: consumer).

Connector-Level (Per-connector JSON)

Property

TypePer-connector override.

	Default	Description
`consumer.override.group.protocol`

string(inherited from worker)

Inherited

Set to share to opt a

single

specific connector into queue semantics.

share.group.id

string

connect-

<connector-name>

<name>

The

Custom Share Group ID

. Defaults to the same naming convention as consumer groups

; follows standard naming conventions.

share.acknowledgement.mode

string

explicit explicit:

worker calls acknowledge(ACCEPT)

Acknowledge after task.put()

succeeds

. implicit:

acknowledgments are sent

Acknowledge on the next poll

() (simpler, lower latency, weaker guarantee)

.

share.acquisition.lock.timeout.ms

int

30000

Maximum

Max time a record

remains in ACQUIRED state before the broker releases it for

stays acquired before re-delivery. Must

be greater than the expected

exceed task.put() latency.

share.delivery.semantics

string

at-least-once Toggle between at-least-once

or

and exactly-once

. Exactly-once

(requires KIP-1289

and a transactional producer

).

share.max.delivery.count

int

5

Maximum number of times a record can be

Max re-

delivered

delivery attempts before

being sent

sending to

the

a Dead Letter Queue

(if configured)

.

Maps to Share Group's group.share.record.lock.partition.limit.

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 changeclass—parallel to WorkerSinkTask—that drives SinkTask using a KafkaShareConsumer.

No API Changes: The public SinkTask interface and put() contract remain identical; existing connectors require no code modifications.

```
// 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 **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:

Aspect	WorkerSinkTask (todayTraditional)	WorkerShareSinkTask (proposedProposed)
Consumer	`KafkaConsumer`	`KafkaShareConsumer`
Subscription	`consumer.subscribe(topics, rebalanceListener)`	`shareConsumer.subscribe(topics)`
Poll	`consumer.poll()`	`shareConsumer.poll()`
Tracking	Consumer Offsets +	Offset tracking	`currentOffsets` map + `consumer.commitSync()`	Per-record `shareConsumer.acknowledge(record, ACCEPT)` + `shareConsumer.commitSync()`
Rebalance	`ConsumerRebalanceListener` calling task.open()Rebalance listener triggers `open/close()`	No rebalancesNone. `task.open()` called once at startup for all subscribed topics.
Failures	Pause consumer and	Failure handling	`RetriableException` -> pause consumer, retry batch	`RetriableException` -> `acknowledge(RELEASE)` for batch, records broker re-delivered by brokerdelivery

3.2.2`Worker.baseConsumerConfigs()` (modified)

The existing method that builds consumer properties is modified Updated to detect `group.protocol=share` and construct `KafkaShareConsumer` configs instead of `KafkaConsumer` configs:

...

group.protocol=share. It dynamically constructs ShareConsumerConfig properties (like share.group.id

...

) instead of traditional consumer configs.

Metrics

New sensors are registered only in Share Group mode to keep dashboards clean and verify the connector state. All metrics belong to the existing sink-task-metrics group.

New Share-Specific Metrics:

sink-record-acquire: Rate/Total of records pulled from the group.
sink-record-acknowledge: Rate/Total of successful ACCEPT acks.
sink-record-release/reject: Rate/Total of records released for retry or rejected to DLQ.
acknowledge-time: Time between poll() and acknowledge().
sink-record-redelivery: Total records with delivery count > 1.

Exclusions: The following traditional sensors are not registered in share mode as they are not applicable: partition-count, offset-seq-number, and offset-commit-completion.

Proposed Changes

At‑Least‑Once (Share Group → SinkTask → External Sink)

Image Added

Exactly‑Once (Same‑Cluster Kafka‑to‑Kafka, KIP‑1289)

Image Added

`WorkerShareSinkTask` Lifecycle

Initialization

The setup phase is simplified because Share Groups eliminate partition-level management.

Consumer Creation: Instantiates KafkaShareConsumer.
Subscription: Subscribes to topics directly (no RebalanceListener needed).
Task Startup: Calls task.initialize() and task.start(). Unlike traditional sinks, task.open() is called once at startup for all topics since there are no rebalances.

Main Loop (Iteration)

The worker drives the at-least-once delivery guarantee by following this execution flow:

Poll: Pulls records from the share group.
Convert: Transforms messages into SinkRecords.
Deliver: Passes records to the connector via task.put().
Acknowledge:
- Success: Marks all records as ACCEPT.
- Retriable Error: Marks records as RELEASE for immediate broker re-delivery.
- Fatal Error: Marks records as REJECT (routes to Dead Letter Queue if configured).
Commit: Calls shareConsumer.commitSync() to durably persist the acknowledgments on the broker.

Ensuring No Data Loss (At-Least-Once)

he system guarantees that no data is lost by ensuring a record is only acknowledged (ACCEPT) after task.put() returns successfully.

Failure Recovery: If a task crashes before acknowledging, the record remains in an ACQUIRED state until the share.acquisition.lock.timeout.ms expires.
The broker then makes the record AVAILABLE for another task.
Durable Commits: The default explicit mode uses commitSync() to ensure acknowledgments are persistent. If a commit fails, the record is re-delivered.
Duplicates: Occasional duplicates may occur if a crash happens after task.put() but before acknowledgment. This is standard for at-least-once delivery and is best handled by idempotent sinks (e.g., upserts).

Exactly-Once Semantics (Future Phase, requires KIP-1289)

For Kafka-to-Kafka pipelines, exactly-once delivery is achieved by atomizing the producer's records and the consumer's acknowledgments within a single transaction via KIP-1289.

Step	Action
1. Begin	Start producer transaction.
2. Produce	Send transformed records to output topics.
3. Bind	Call `producer.sendShareAcksToTransaction()` to link share acks to the transaction.
4. Commit	Atomically commit both output records and source acknowledgments.

Configuration Resolution Order

The proposal reuses the existing consumer.override.* mechanism in Kafka Connect for a seamless transition.

Resolution Order:

Worker Config: consumer.group.protocol=share (Global default).
Connector Config: consumer.override.group.protocol=share (Per-connector override).

Important Note on Group IDs: Share groups use a specific state topic (__share_group_state).
To avoid membership conflicts, users must ensure that a Share Group ID does not match an existing Consumer Group ID. A validation check will be implemented to prevent this collision.

4. Compatibility, Deprecation, and Migration Plan

Impact on Users: KIP-1302 is opt-in only and purely additive. Standard consumer groups remain the default, and zero code changes are required for existing connectors.
Migration Path:
1. Ensure brokers are version 4.0+.
2. Set group.protocol=share at the worker level (all connectors) or per-connector JSON.
3. Tune share.acquisition.lock.timeout.ms to exceed your highest expected task.put() latency.
Rollback: Removing the share config reverts the connector to standard consumer groups.
- Note: Consumer groups and Share Groups track progress separately; a rollback may result in processing records that were already handled by the Share Group.

5. Test Plan

Test Level	Key Objectives
Unit	Verify the poll-put-acknowledge loop (`ACCEPT`/`RELEASE`/`REJECT`) and config resolution.
Integration	Test elastic scaling (adding/removing tasks), task failure recovery, and interoperability between share and traditional connectors.
System	Performance benchmarking against traditional consumer groups and chaos testing to verify zero data loss.

6. Rejected Alternatives

Alternative	Reason for Rejection
Modify `SinkTask` API	Adding explicit `acknowledge()` methods would break backward compatibility for all existing connectors.
Limit to MirrorMaker 2	Generic sinks (S3, JDBC, etc.) benefit just as much from elastic scaling as Kafka-to-Kafka pipelines.
Require EOS Initially	At-least-once is sufficient for most use cases, and exactly-once is blocked by the pending KIP-1289/KIP-1310.

Metrics

These sensors are only registered by `WorkerShareSinkTask` -- they are not present when using a traditional `KafkaConsumer` via `WorkerSinkTask`.

This avoids publishing meaningless zeros and keeps dashboards clean. Operators can use the presence/absence of these metrics to confirm whether a connector is running in Share Group mode.

All metrics are registered under the existing `sink-task-metrics` group (same as `sinkTaskGroupName` in `ConnectMetricsRegistry`), tagged with `connector` and `task`.

This keeps them co-located with the existing `sink-record-read-total`, `sink-record-send-total`, etc. and avoids a separate metric namespace.

...

Conversely, the following existing `WorkerSinkTask` sensors have no Share Group equivalent and are not registered by `WorkerShareSinkTask`:

Existing Sensor	Why not applicable to Share Groups
`partition-count`	Share Groups don't assign partitions exclusively to tasks. All tasks consume from all subscribed partitions.
`offset-seq-number`	Share Groups don't use consumer offsets. Acknowledgments replace offset commits.
`offset-commit-completion`	No offset commits in Share Groups. Replaced by `sink-record-acknowledge`.
`offset-commit-completion-skip`	No offset commits to skip.

The existing sensors that are shared between both task types:

Sensor	Behavior
`sink-record-read`	Registered by both. Counts records polled (same semantics).
`sink-record-send`	Registered by both. Counts records delivered to `task.put()`.
`sink-record-active-count`	Registered by both. In Share Groups, this is the number of records currently ACQUIRED but not yet acknowledged.
`put-batch-time`	Registered by both. Time spent in `task.put()`.

4. Compatibility, Deprecation, and Migration Plan

5. Test Plan

6. Future Work

...

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 9

New Version Current

Key

Table of Contents

Status

1 Motivation

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

1.2 How Share Groups Solve This

1.3 Kafka Connect as the Natural Integration Point

1.4 Delivery Semantics

1.5 Error Handling / Delivery Semantics

EO Constraints

2. Scope

2.1 In Scope

(What we are building)

2.2 Out of Scope

(Future/Separate efforts)

3. Public Interfaces

3.1 New Configuration Properties

Worker-

Level (

connect-distributed.

properties)

Connector-Level (Per-connector JSON)

3.2 New / Modified Java Interfaces

3.2.1 `WorkerShareSinkTask` (new class)

3.2.2`Worker.baseConsumerConfigs()` (modified)

Metrics

Proposed Changes

At‑Least‑Once (Share Group → SinkTask → External Sink)

Exactly‑Once (Same‑Cluster Kafka‑to‑Kafka, KIP‑1289)

`WorkerShareSinkTask` Lifecycle

Initialization

Main Loop (Iteration)

Ensuring No Data Loss (At-Least-Once)

Exactly-Once Semantics (Future Phase, requires KIP-1289)

Configuration Resolution Order

4. Compatibility, Deprecation, and Migration Plan

5. Test Plan

6. Rejected Alternatives

6. Rejected Alternatives

Metrics

4. Compatibility, Deprecation, and Migration Plan

5. Test Plan

6. Future Work

`connect-distributed.`

`properties`)