DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
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
| Property | Type | Default | Description |
consumer.group.protocol | string | consumer | Existing property. When set to share, the Connect worker creates a KafkaShareConsumer instead of a KafkaConsumer for sink tasks. |
3.1.1 Worker-level configuration (`connect-distributed.properties`)
| Property | Type | Default | Description |
consumer.override.group.protocol | string | (inherited from worker) | Per-connector override. Set to share to opt a single connector into queue semantics. |
share.group.id | string | connect-<connector-name> | The Share Group ID. Defaults to the same naming convention as consumer groups. |
share.acknowledgement.mode | string | explicit | explicit: worker calls acknowledge(ACCEPT) after task.put() succeeds. implicit: acknowledgments are sent on the next poll() (simpler, lower latency, weaker guarantee). |
share.acquisition.lock.timeout.ms | int | 30000 | Maximum time a record remains in ACQUIRED state before the broker releases it for re-delivery. Must be greater than the expected task.put() latency. |
share.delivery.semantics | string | at-least-once | at-least-once or exactly-once. Exactly-once requires KIP-1289 and a transactional producer. |
share.max.delivery.count | int | 5 | Maximum number of times a record can be re-delivered before being sent to the Dead Letter Queue (if configured). Maps to Share Group's group.share.record.lock.partition.limit. |
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.
```
// 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:
4. Compatibility, Deprecation, and Migration Plan
...