Table of Contents

Status

Current state: Under discussion

...

JIRA:

1 Motivation

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

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

...

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.

...

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.

...

3.2.2`Worker.baseConsumerConfigs()` (modified)

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

```
// In Worker.java
static Map<String, Object> baseConsumerConfigs(...) {
    Map<String, Object> consumerProps = new HashMap<>();
    
    String groupProtocol = // resolve from worker + connector config
    
    if ("share".equals(groupProtocol)) {
        consumerProps.put(ShareConsumerConfig.GROUP_ID_CONFIG, 
            connConfig.getString("share.group.id", SinkUtils.consumerGroupId(connName)));
        // Share consumer specific configs
        consumerProps.put(ShareConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, config.bootstrapServers());

    } else {
        // existing consumer group config path (unchanged)
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, SinkUtils.consumerGroupId(connName));
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // ...
    }
    return consumerProps;
}
```

Metrics

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

...

Proposed Changes

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

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

`WorkerShareSinkTask` Lifecycle

Initialization

```
void initialize() {
    // 1. Create KafkaShareConsumer with resolved configs
    this.shareConsumer = new KafkaShareConsumer<>(shareConsumerConfigs);
    
    // 2. Subscribe to configured topics
    List<String> topics = SinkConnectorConfig.parseTopicsList(taskConfig);
    shareConsumer.subscribe(topics);
    
    // 3. Open the task (no partition-level open/close with share groups)
    task.initialize(context);
    task.start(taskConfig);
}
```

Main Loop (iteration)

```
void iteration() {
    // 1. Poll records from share group
    ConsumerRecords<byte[], byte[]> records = shareConsumer.poll(Duration.ofMillis(pollTimeoutMs));
    
    if (records.isEmpty()) return;
    
    // 2. Convert to SinkRecords (same as today)
    List<SinkRecord> sinkRecords = convertMessages(records);
    
    // 3. Deliver to task
    try {
        task.put(sinkRecords);
        
        // 4a. Success: acknowledge all records as ACCEPT
        for (ConsumerRecord<byte[], byte[]> record : records) {
            shareConsumer.acknowledge(record, AcknowledgeType.ACCEPT);
        }
        
    } catch (RetriableException e) {
        // 4b. Retriable failure: RELEASE records for re-delivery
        for (ConsumerRecord<byte[], byte[]> record : records) {
            shareConsumer.acknowledge(record, AcknowledgeType.RELEASE);
        }
        log.warn("Retriable error, records released for re-delivery", e);
        
    } catch (Throwable t) {
        // 4c. Fatal failure: REJECT records (to DLQ if configured)
        for (ConsumerRecord<byte[], byte[]> record : records) {
            shareConsumer.acknowledge(record, AcknowledgeType.REJECT);
        }
        throw new ConnectException("Unrecoverable error", t);
    }
    
    // 5. Commit acknowledgments to broker
    if (shouldCommit()) {
        shareConsumer.commitSync();
    }
}
```

Ensuring No Data Loss (At-Least-Once)

The at-least-once guarantee is achieved through the following invariant:

...

This is inherent to at-least-once semantics. Sink connectors targeting idempotent systems (databases with upsert, object stores with overwrite) naturally handle this.

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

For Kafka-to-Kafka pipelines (e.g., MirrorMaker2), exactly-once can be achieved by binding the Share Group acknowledgment to the producer's transaction:

```
// Exactly-once CTP pattern in WorkerShareSinkTask
void iterationExactlyOnce() {
    ConsumerRecords<byte[], byte[]> records = shareConsumer.poll(Duration.ofMillis(pollTimeoutMs));
    if (records.isEmpty()) return;
    
    producer.beginTransaction();
    
    try {
        // Produce transformed records to output topics
        for (SinkRecord record : convertMessages(records)) {
            producer.send(new ProducerRecord<>(outputTopic, record.key(), record.value()));
        }
        
        // Bind share acks to this transaction (KIP-1289)
        producer.sendShareAcksToTransaction(
            ShareAcknowledgements.fromRecords(records, AcknowledgeType.ACCEPT),
            shareConsumer.groupMetadata()
        );
        
        producer.commitTransaction();
        // Output records AND source acknowledgments commit atomically
        
    } catch (Exception e) {
        producer.abortTransaction();
        // Both output records AND source acknowledgments are rolled back
        // Records will be re-delivered by the broker
    }
}
```

 Configuration Resolution Order

```
Worker config (connect-distributed.properties)
    -> consumer.group.protocol=share          (global default)
    
Connector config (per-connector JSON)
    -> consumer.override.group.protocol=share  (per-connector override)
    -> share.group.id=my-custom-group          (explicit share group name)
    -> share.acknowledgement.mode=explicit     (ack behavior)
```

The existing `consumer.override.*` mechanism in Kafka Connect (governed by `connector.client.config.override.policy`) is reused. No new override mechanism is introduced.

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.

...

(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

...

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