...

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)

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.

Ensuring No Data Loss (At-Least-Once)

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

> A record is acknowledged (ACCEPT) only after `task.put()` returns successfully.

If the task or worker crashes between `poll()` and `acknowledge()`:
- The record remains in ACQUIRED state on the broker
- The acquisition lock timer expires after `share.acquisition.lock.timeout.ms`
- The broker transitions the record back to AVAILABLE
- Another task acquires and processes it

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.

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.

Note: Share groups use a different state topic (__share_group_state), but looks like __consumer_offsets will be used for memebership, so if we do not delete the group before switching it can cause problem.

...

4. Compatibility, Deprecation, and Migration Plan

...