...

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:

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

4. Compatibility, Deprecation, and Migration Plan

...