Status

Current state: Under discussion

Discussion thread:  here

JIRA:

1 Motivation

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

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

his model is often incompatible with unordered message processing and creates three primary bottlenecks for task queue workloads:

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

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."

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

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)

3.2 New / Modified Java Interfaces

3.2.1 `WorkerShareSinkTask` (new class)

A new internal class—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 not modified. Connectors do not need code changes. The `put(Collection<SinkRecord>)` contract remains the same.

The difference is entirely in the worker runtime:

3.2.2`Worker.baseConsumerConfigs()` (modified)

Updated to detect 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)

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

`WorkerShareSinkTask` Lifecycle

Initialization

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

1. Consumer Creation: Instantiates KafkaShareConsumer.

2. Subscription: Subscribes to topics directly (no RebalanceListener needed).

3. 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:

1. Poll: Pulls records from the share group.

2. Convert: Transforms messages into SinkRecords.

3. Deliver: Passes records to the connector via task.put().

4. 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).

5. 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.

 Configuration Resolution Order

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

Resolution Order:

1. Worker Config: consumer.group.protocol=share (Global default).

2. 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

4.1 Impact on Existing Users

-  No impact by default. The default `group.protocol` remains `consumer` (standard consumer group). Existing connectors continue to work identically.
- Opt-in only. Share Groups are enabled per-connector or per-worker via configuration.
- No connector code changes required. The `SinkTask` interface is unchanged. Any existing sink connector works with Share Groups without modification.

4.2 Migration Path

1. Pre-requisite: Kafka broker version must support Share Groups (4.0+).
2. Enable at worker level: Set `consumer.group.protocol=share` in `connect-distributed.properties` to make all sink connectors use Share Groups.
3. Or enable per-connector: Set `consumer.override.group.protocol=share` in the connector config JSON.
4. Tune acquisition lock timeout: Set `share.acquisition.lock.timeout.ms` to a value greater than the expected `task.put()` latency. The default of 30 seconds is suitable for most workloads.
5. Monitor: Use the new `share-sink-task.*` metrics to observe acknowledgment patterns and re-delivery rates.

4.3 Rollback

To revert, remove the `group.protocol=share` configuration. The connector will resume using standard consumer groups.

Note that Share Groups and consumer groups maintain separate offset tracking, so the consumer group will resume from its last committed offset

(which may be behind the Share Group's position).

4.4 Deprecation

No existing features are deprecated. This is purely additive.

5. Test Plan

5.1 Unit Tests

1. `WorkerShareSinkTaskTest`: Tests the core poll-put-acknowledge loop using a `MockShareConsumer`.
   - Verify ACCEPT after successful `task.put()`
   - Verify RELEASE after `RetriableException`
   - Verify REJECT after unrecoverable exception
   - Verify `commitSync()` is called at configured intervals

2. `WorkerTest` (modified): Verify that `baseConsumerConfigs()` returns correct configs for `group.protocol=share`.

3. `SinkConnectorConfigTest` (modified): Validate the new configuration properties and their defaults.

5.2 Integration Tests

1. Basic Share Group Sink: Deploy a sink connector with `group.protocol=share` and verify all records are delivered.
2. Elastic Scaling: Start with 2 tasks, scale to 6, verify no records are lost and throughput increases.
3. Task Failure and Re-delivery: Kill a task mid-processing, verify records are re-delivered to surviving tasks within `acquisition.lock.timeout.ms`.
4. No Duplicate Loss: Produce N records, consume with at-least-once Share Group sink, verify received count >= N.
5. Interoperability: Verify that standard consumer group connectors and Share Group connectors can coexist in the same Connect cluster.

5.3 System Tests

1. Long-running throughput test: Measure throughput and latency of Share Group vs. consumer group sink connectors under sustained load.
2. Chaos test: Randomly kill tasks and brokers, verify zero data loss with at-least-once semantics.

6. Rejected Alternatives

Alternative 1: Modify the SinkTask Interface to Add acknowledge()

We considered adding `acknowledge(SinkRecord)` and `release(SinkRecord)` methods to the `SinkTask` interface, giving connectors explicit control over acknowledgments. This was rejected because:
- It would break backward compatibility with all existing sink connectors
- Most connectors don't need per-record acknowledgment control
- The worker runtime can make correct acknowledgment decisions based on `put()` success/failure

Alternative 2: Use Share Groups Only for MirrorMaker2

We considered limiting Share Group support to the `MirrorSourceConnector` only, as Kafka-to-Kafka is the most obvious use case. This was rejected because:
- It would require changes to the MM2 `consumer.assign()` model, which is complex
- Generic sink connectors (e.g., JDBC, Elasticsearch, S3) benefit equally from elastic scaling
- Building it into the Connect runtime benefits all connectors automatically

Alternative 3: Exactly-Once from Day One

We considered requiring exactly-once semantics for the initial implementation. This was rejected because:
- KIP-1289 (transactional share acknowledgments) is not yet implemented
- At-least-once is sufficient for the majority of sink connector use cases
- Idempotent sinks (upsert to database, overwrite to S3) achieve effective exactly-once with at-least-once delivery
- Exactly-once can be added as a follow-up without breaking changes