DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
The existing method that builds consumer properties is modified to detect `group.protocol=share` and construct `KafkaShareConsumer` configs instead of `KafkaConsumer` configs:
``````java
// In Worker.java
static Map<String, Object> baseConsumerConfigs(...) {
Map<String, Object> consumerProps = new HashMap<>();
String groupProtocol = // resolve from worker + connector config
if ("share".equals(groupProtocol)) {
consumerProps.put(ShareConsumerConfig.GROUP_ID_CONFIG,
connConfig.getString("share.group.id", SinkUtils.consumerGroupId(connName)));
// Share consumer specific configs
consumerProps.put(ShareConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, config.bootstrapServers());
// No ENABLE_AUTO_COMMIT -- share groups don't have this concept
// No AUTO_OFFSET_RESET -- share groups start from the share partition start offset
} else {
// existing consumer group config path (unchanged)
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, SinkUtils.consumerGroupId(connName));
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
// ...
}
return consumerProps;
}
```
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.
| 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(). |
4. Compatibility, Deprecation, and Migration Plan
...