# Share Groups for Confluent Platform You can enable and monitor share groups on Confluent Platform brokers, using server-side configuration settings, commands, and JMX metrics. For a Docker example to get a testing environment up and running, see [Configure Share Groups for Docker in Confluent Platform](../installation/docker/operations/kafka-queues-docker.md#share-groups-docker). For client-side details, see [Share Consumers for Confluent Platform](../clients/share-consumers.md#kafka-sharegroup). Share partition lag represents the number of offsets in a partition that are not delivery complete for a share group. This metric reports processing progress and helps you automate scaling of share consumer instances. For configuration details, see the `share-lag-offsets` metric in [share-lag-offsets](../kafka/broker-metrics.md#share-lag-offsets). ## Enable share groups Share groups are enabled by default on Confluent Platform 8.3 clusters. Broker administrators can tune parameters related to share group behavior and resource consumption. To enable share groups on existing Confluent Platform 8.2 clusters, use the `kafka-features.sh` tool to set `share.version=1`: 1. Generate a cluster UUID: ```bash KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)" ``` 2. Format the log directories. ```bash bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties ``` 3. Start the Kafka server. ```bash bin/kafka-server-start.sh config/server.properties ``` 4. Set the feature version. ```bash kafka-features.sh --bootstrap-server localhost:9092 upgrade --feature share.version=1 ``` ## Set a per-group configuration These settings can be overridden per share group using the `kafka-configs` tool: * `share.delivery.count.limit` * `share.partition.max.record.locks` * `share.renew.acknowledge.enable` To set these values for the share group, use `kafka-configs` with `--entity-type groups --entity-name ` and the `--add-config` flag: ```bash bin/kafka-configs.sh --bootstrap-server : \ --entity-type groups --entity-name my-share-group \ --alter --add-config share.delivery.count.limit=5,share.partition.max.record.locks=1000 ``` Adjust these broker-level settings to define the minimum and maximum bounds for the per-group overrides: * `group.share.min.delivery.count.limit` * `group.share.max.delivery.count.limit` * `group.share.min.partition.max.record.locks` * `group.share.max.partition.max.record.locks` Per-group values outside the bounds are adjusted to [clamp](../_glossary.md#term-clamp) to the nearest bound. ## Configurations for share groups The following broker configurations are pertinent to share groups. For more information, see [Kafka Broker and Controller Configuration Reference for Confluent Platform](../installation/configuration/broker-configs.md#cp-config-brokers). ### Share coordinator configurations These properties control the operation, resource usage, and persistence of the share coordinator component on the broker. * `group.share.max.share.sessions` * `group.share.assignors` * `confluent.share.coordinator.state.topic.placement.constraints` * `share.coordinator.state.topic.num.partitions` * `share.coordinator.state.topic.replication.factor` * `share.coordinator.state.topic.segment.bytes` * `share.coordinator.state.topic.min.isr` * `share.coordinator.state.topic.compression.codec` * `share.coordinator.append.linger.ms` * `share.coordinator.cached.buffer.max.bytes` * `share.coordinator.load.buffer.size` * `share.coordinator.snapshot.update.records.per.snapshot` * `share.coordinator.threads` * `share.coordinator.write.timeout.ms` * `share.fetch.purgatory.purge.interval.requests` ### Group configurations These properties establish the default or boundary settings for the share group behavior, which can often be overridden or constrained by dynamic group configuration. * `group.share.assignment.interval.ms` * `group.share.delivery.count.limit` * `group.share.heartbeat.interval.ms` * `group.share.max.assignment.interval.ms` * `group.share.max.delivery.count.limit` * `group.share.max.heartbeat.interval.ms` * `group.share.max.partition.max.record.locks` * `group.share.max.record.lock.duration.ms` * `group.share.max.session.timeout.ms` * `group.share.max.size` * `group.share.min.assignment.interval.ms` * `group.share.min.delivery.count.limit` * `group.share.min.heartbeat.interval.ms` * `group.share.min.partition.max.record.locks` * `group.share.min.record.lock.duration.ms` * `group.share.min.session.timeout.ms` * `group.share.partition.max.record.locks` * `group.share.record.lock.duration.ms` * `group.share.session.timeout.ms` ### Dynamic group configurations These configurations are dynamic, meaning they are applied per share group and managed by the broker. All of the configurations in the previous two lists are applied to the broker. These configurations are applied to the group itself: * `share.auto.offset.reset` * `share.delivery.count.limit` * `share.heartbeat.interval.ms` * `share.isolation.level` * `share.partition.max.record.locks` * `share.record.lock.duration.ms` * `share.renew.acknowledge.enable` * `share.session.timeout.ms` ## Metrics for share groups Confluent Platform exposes several metrics through Java Management Extensions (JMX) that are useful for monitoring share groups. For more information, see [Monitor Kafka with JMX in Confluent Platform](../kafka/monitoring.md#kafka-monitoring). ### Broker topic metrics (general share fetch) #### TotalShareFetchRequestsPerSec MBean: `kafka.server:type=BrokerTopicMetrics,name=TotalShareFetchRequestsPerSec,topic={topic}` : The fetch request rate per second. This metric is tracked cumulatively and per-topic. #### FailedShareFetchRequestsPerSec MBean: `kafka.server:type=BrokerTopicMetrics,name=FailedShareFetchRequestsPerSec,topic={topic}` : The share fetch request rate per second for requests that failed. This metric is tracked cumulatively and per-topic. #### TotalShareAcknowledgementRequestsPerSec MBean: `kafka.server:type=BrokerTopicMetrics,name=TotalShareAcknowledgementRequestsPerSec,topic={topic}` : The acknowledgement request rate per second. This metric is tracked cumulatively and per-topic. #### FailedShareAcknowledgementRequestsPerSec MBean: `kafka.server:type=BrokerTopicMetrics,name=FailedShareAcknowledgementRequestsPerSec,topic={topic}` : The share acknowledgement request rate per second for requests that failed. This metric is tracked cumulatively and per-topic. #### share-lag-offsets MBean: `kafka.server:type=tenant-metrics,group-protocol=share,partition={partitionid},group-id={sharegroupid},topic={topicname}` : Attribute: `share-lag-offsets`
This metric represents the number of offsets in a partition that are not delivery complete for a specific share group ID, partition ID, topic name, and group protocol (share). The group protocol specifies the rebalance protocol used by the group, currently `share`. This metric reports share lag in offsets only and does not report latency. In addition, it is not reported for any groups that are not alive or are empty.
To enable share lag, set `confluent.share.lag.calculator.enabled` to `true`. This property enables share lag monitoring, and is `false` by default. This high-cardinality metric could impact systems consuming it.
Additionally, `org.apache.kafka.common.metrics.JmxReporter` must be added as part of the `metric.reporters` broker configuration to see the metric in JMX. ### Share group metrics #### RecordAcknowledgementsPerSec MBean: `kafka.server:type=ShareGroupMetrics,name=RecordAcknowledgementsPerSec,ackType={Accept|Release|Reject}` : The rate per second of records acknowledged, segmented by the acknowledgement type, such as Accept, Release, or Reject. #### PartitionLoadTimeMs MBean: `kafka.server:type=ShareGroupMetrics,name=PartitionLoadTimeMs` : The time taken to load the share partitions. #### RequestTopicPartitionsFetchRatio MBean: `kafka.server:type=ShareGroupMetrics,name=RequestTopicPartitionsFetchRatio,group={group_id}` : The ratio of topic-partitions successfully acquired for fetching to the total number of topic-partitions requested in a share fetch for a specific share group. #### TopicPartitionsAcquireTimeMs MBean: `kafka.server:type=ShareGroupMetrics,name=TopicPartitionsAcquireTimeMs,group={group_id}` : The time elapsed (in milliseconds) to acquire any topic partition for a fetch for a specific share group. ### Share partition metrics #### AcquisitionLockTimeoutPerSec MBean: `kafka.server:type=SharePartitionMetrics,name=AcquisitionLockTimeoutPerSec,group={group_id},topic={topic_name},partition={partition}` : The rate of acquisition locks for records that were not acknowledged within the configured timeout. This is tracked per share partition. #### InFlightMessageCount MBean: `kafka.server:type=SharePartitionMetrics,name=InFlightMessageCount,group={group_id},topic={topic_name},partition={partition}` : The current number of in-flight messages (unacknowledged records) for the share partition. #### InFlightBatchCount MBean: `kafka.server:type=SharePartitionMetrics,name=InFlightBatchCount,group={group_id},topic={topic_name},partition={partition}` : The number of in-flight batches currently being processed for the share partition. #### InFlightBatchMessageCount MBean: `kafka.server:type=SharePartitionMetrics,name=InFlightBatchMessageCount,group={group_id},topic={topic_name},partition={partition}` : The number of messages in the in-flight batch for the share partition. #### FetchLockTimeMs MBean: `kafka.server:type=SharePartitionMetrics,name=FetchLockTimeMs,group={group_id},topic={topic_name},partition={partition}` : The time elapsed (in milliseconds) while a share partition is held under lock for fetching messages. #### FetchLockRatio MBean: `kafka.server:type=SharePartitionMetrics,name=FetchLockRatio,group={group_id},topic={topic_name},partition={partition}` : The fraction of time the share partition is held under lock for fetching messages. ### Share session cache metrics #### ShareSessionEvictionsPerSec MBean: `kafka.server:type=ShareSessionCache,name=ShareSessionEvictionsPerSec` : The share session eviction rate per second. #### SharePartitionsCount MBean: `kafka.server:type=ShareSessionCache,name=SharePartitionsCount` : The number of cached share partitions in the session cache. #### ShareSessionsCount MBean: `kafka.server:type=ShareSessionCache,name=ShareSessionsCount` : The number of cached share sessions. ### Delayed operation metrics #### ExpiresPerSec MBean: `kafka.server:type=DelayedShareFetchMetrics,name=ExpiresPerSec` : The expired delayed share fetch operation rate per second. #### PurgatorySize MBean: `kafka.server:type=DelayedOperationPurgatory,delayedOperation=ShareFetch,name=PurgatorySize` : The number of requests currently waiting in the share fetch purgatory. #### NumDelayedOperations MBean: `kafka.server:type=DelayedOperationPurgatory,delayedOperation=ShareFetch,name=NumDelayedOperations` : The number of delayed operations for the share fetch purgatory. ## Related content - [Share Consumers for Confluent Platform](../clients/share-consumers.md#kafka-sharegroup) - [Monitor Kafka with JMX in Confluent Platform](../kafka/monitoring.md#kafka-monitoring) - [Kafka Broker and Controller Configuration Reference for Confluent Platform](../installation/configuration/broker-configs.md#cp-config-brokers) - [Configure Share Groups for Docker in Confluent Platform](../installation/docker/operations/kafka-queues-docker.md#share-groups-docker)