DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
3.1 New Configuration Properties
Worker-Level (connect-distributed.properties)
consumer.group.protocol: Set toshareto enableKafkaShareConsumerglobally for all sink tasks (Default:consumer).
Connector-Level (Per-connector JSON)
| Property | Default | Description |
consumer.override.group.protocol | Inherited | Set to share to opt a specific connector into queue semantics. |
share.group.id | connect-<name> | Custom Share Group ID; follows standard naming conventions. |
share.acknowledgement.mode | explicit | explicit: Acknowledge after task.put(). implicit: Acknowledge on the next poll. |
share.acquisition.lock.timeout.ms | 30000 | Max time a record stays acquired before re-delivery. Must exceed task.put() latency. |
share.delivery.semantics | at-least-once | Toggle between at-least-once and exactly-once (requires KIP-1289). |
share.max.delivery.count | 5 | Max re-delivery attempts before sending to a Dead Letter Queue. |
3.2 New / Modified Java Interfaces
...
3.2.1 `WorkerShareSinkTask` (new class)
A new internal
...
class—parallel to WorkerSinkTask—that drives SinkTask using a KafkaShareConsumer.
No API Changes: The public
SinkTaskinterface andput()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:
| Aspect | WorkerSinkTask ( |
| Traditional) | WorkerShareSinkTask ( |
| Proposed) | ||
| Consumer | KafkaConsumer | KafkaShareConsumer |
consumer.subscribe(topics, rebalanceListener)shareConsumer.subscribe(topics)consumer.poll()shareConsumer.poll()currentOffsets map + consumer.| Tracking | Consumer Offsets + |
commitSync() | Per-record |
acknowledge( |
ACCEPT) |
shareConsumer.commitSync()task.open()| Rebalance |
ConsumerRebalanceListener calling Rebalance listener triggers open/close |
None. task.open() called once at startup |
| . |
RetriableException -> pause consumer, | Failures | Pause consumer and retry batch |
RetriableException -> acknowledge(RELEASE) for |
| broker re- |
| delivery |
3.2.2`Worker.baseConsumerConfigs()` (modified)
The existing method that builds consumer properties is modified Updated to detect `groupgroup.protocol=share` and construct `KafkaShareConsumer` configs instead of `KafkaConsumer` configs:
...
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 successfulACCEPTacks.sink-record-release/reject: Rate/Total of records released for retry or rejected to DLQ.acknowledge-time: Time betweenpoll()andacknowledge().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
Metrics
These sensors are only registered by `WorkerShareSinkTask` -- they are not present when using a traditional `KafkaConsumer` via `WorkerSinkTask`.
This avoids publishing meaningless zeros and keeps dashboards clean. Operators can use the presence/absence of these metrics to confirm whether a connector is running in Share Group mode.
All metrics are registered under the existing `sink-task-metrics` group (same as `sinkTaskGroupName` in `ConnectMetricsRegistry`), tagged with `connector` and `task`.
This keeps them co-located with the existing `sink-record-read-total`, `sink-record-send-total`, etc. and avoids a separate metric namespace.
...
Conversely, the following existing `WorkerSinkTask` sensors have no Share Group equivalent and are not registered by `WorkerShareSinkTask`:
| Existing Sensor | Why not applicable to Share Groups |
partition-count | Share Groups don't assign partitions exclusively to tasks. All tasks consume from all subscribed partitions. |
offset-seq-number | Share Groups don't use consumer offsets. Acknowledgments replace offset commits. |
offset-commit-completion | No offset commits in Share Groups. Replaced by sink-record-acknowledge. |
offset-commit-completion-skip | No offset commits to skip. |
The existing sensors that are shared between both task types:
...
.
Proposed Changes
At‑Least‑Once (Share Group → SinkTask → External Sink)
...