DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
1.1 The Problem: Scaling Kafka-to-Kafka Pipelines Today
Currently, Kafka Connect sink connectors consume from Kafka topics using rely on traditional consumer groups that enforce a strict 1:1 mapping between partitions and tasks. T
In this his model , each partition is exclusively assigned to one task. This creates two problems for Kafka-to-Kafka (and Kafka-to-external) pipelinesis often incompatible with unordered message processing and creates three primary bottlenecks for task queue workloads:
1. Partition-Coupled Scaling is coupled to : Parallelism is hard-limited by the partition count.
If a topic has 12 partitions, you can run at most 12 sink tasks. I
...
2. Head-of-Line Blocking: Because partition ownership is exclusive, a single slow task—often caused by downstream latency—stalls all subsequent records in its assigned partitions
3. Rebalance-Driven Gaps: Adding or removing tasks triggers "rebalance storms."
If one sink task is slow (e.g., network latency to a downstream system), the records on its assigned partitions back up.
...
When tasks are added, removed, or crash, consumer group rebalances revoke and reassign partitions.
During a rebalance, no task processes records from revoked partitions. With cooperative sticky rebalancing this is mitigated but not eliminated.
1.2 How Share Groups Solve This
...
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)
...
This keeps them co-located with the existing `sink-record-read-total`, `sink-record-send-total`, etc. and avoids a separate metric namespace.
| Sensor Name | Metric Name | Type | Traditional Consumer (group.protocol=consumer) | Share Consumer (group.protocol=share) |
sink-record-acquire | sink-record-acquire-rate | Rate | not registered | Records/sec acquired from the share group |
sink-record-acquire-total | CumulativeSum | not registered | Total records acquired from the share group | |
sink-record-acknowledge | sink-record-acknowledge-rate | Rate | not registered | Records/sec acknowledged (ACCEPT) |
sink-record-acknowledge-total | CumulativeSum | not registered | Total records acknowledged (ACCEPT) | |
sink-record-release | sink-record-release-rate | Rate | not registered | Records/sec released (RELEASE) for re-delivery |
sink-record-release-total | CumulativeSum | not registered | Total records released for re-delivery | |
sink-record-reject | sink-record-reject-rate | Rate | not registered | Records/sec rejected (REJECT) to DLQ |
sink-record-reject-total | CumulativeSum | not registered | Total records rejected to DLQ | |
acknowledge-time | acknowledge-time-max | Max | not registered | Max time (ms) between poll() and acknowledge() |
acknowledge-time-avg | Avg | not registered | Avg time (ms) between poll() and acknowledge() | |
sink-record-redelivery | sink-record-redelivery-total | CumulativeSum | not registered | Total records with delivery count > 1 |
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:
| Sensor | Behavior |
sink-record-read | Registered by both. Counts records polled (same semantics). |
sink-record-send | Registered by both. Counts records delivered to task.put(). |
sink-record-active-count | Registered by both. In Share Groups, this is the number of records currently ACQUIRED but not yet acknowledged. |
put-batch-time | Registered by both. Time spent in task.put(). |
Proposed Changes
At‑Least‑Once (Share Group → SinkTask → External Sink)
...