DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
3.1 New Configuration Properties
3.1.1 Worker-level configuration (`connect-distributed.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.
...
2 Connector-level configuration (
...
per-connector JSON)
| 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.2 New / Modified Java Interfaces
...
The difference is entirely in the worker runtime:
| Aspect | WorkerSinkTask (today) | WorkerShareSinkTask (proposed) |
| Consumer | KafkaConsumer | KafkaShareConsumer |
| Subscription | consumer.subscribe(topics, rebalanceListener) | shareConsumer.subscribe(topics) |
| Poll | consumer.poll() | shareConsumer.poll() |
| Offset tracking | currentOffsets map + consumer.commitSync() | Per-record shareConsumer.acknowledge(record, ACCEPT) + shareConsumer.commitSync() |
| Rebalance | ConsumerRebalanceListener calling task.open()/close() | No rebalances. task.open() called once at startup for all subscribed topics. |
| Failure handling | RetriableException -> pause consumer, retry batch | RetriableException -> acknowledge(RELEASE) for batch, records re-delivered by broker |
3.2.2`Worker.baseConsumerConfigs()` (modified)
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;
// ...
}
```
4. Compatibility, Deprecation, and Migration Plan
...