...

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;
    // ...
}
```

...

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)

...

`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)

...

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)

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

> A record is he system guarantees that no data is lost by ensuring a record is only acknowledged (ACCEPT) only after `tasktask.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)

If the worker crashes after `acknowledge(ACCEPT)` but before `commitSync()`:
- The implicit acknowledgment mode sends acks on the next `poll()`, so uncommitted acks may be lost
- The explicit mode (default) uses `commitSync()` which is durable. If the commit fails, the record stays in ACQUIRED and will time out and re-deliver.

Duplicate delivery can occur when a task successfully calls `task.put()` and `acknowledge(ACCEPT)` but crashes before the downstream system confirms persistence.

...

• .

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

For Kafka-to-Kafka pipelines (e.g., MirrorMaker2), exactly-once can be delivery is 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

...

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

...

1. (

...

1. Per-connector

...

1. override)

...

1. .

...

Important Note on Group IDs

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

...

: 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

...

-  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

...

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

...

• 1. to exceed your highest expected task.put()

...

• 1. latency.

...

• Rollback: Removing the share config reverts the connector to

4.3 Rollback

...

• standard consumer groups.

  • Note

...

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

4.4 Deprecation

...

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

5.1 Unit Tests

...

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

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

...

6. Rejected Alternatives

...

...

...

...

Alternative 2: Use Share Groups Only for MirrorMaker2

...

...

Alternative 3: Exactly-Once from Day One

...

...

...

...