DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
Status
Current state: Under Discussion
Discussion thread: here
JIRA: here
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
This KIP builds upon KIP-932, which introduced the concept of Queues/Cooperative Consumption in Kafka. KIP-932 implementation introduced Share Group, Share Partitions and Delayed Share Fetch purgatory to enable this cooperative consumption behavior. The proposed metrics in this KIP will enhance the debugging and monitoring capabilities for cooperative consumption. By providing a more granular view of the underlying mechanisms, these metrics will aid in identifying and resolving potential issues and optimizing performance.
Public Interfaces
| Rationale | Metric Name | Type | Group | Tags | Description | JMX Bean |
|---|---|---|---|---|---|---|
Similar to existing fetch metric TotalFetchRequestsPerSec for Fetch, which record all topic cumulative and individual topic stats. | TotalShareFetchRequestsPerSec | Meter | ShareGroupMetrics | topic:{topic} | The fetch request rate per second. | kafka.server:type=ShareGroupMetrics,name=TotalShareFetchRequestsPerSec kafka.server:type=ShareGroupMetrics,name=TotalShareFetchRequestsPerSec,topic={topic} |
Similar to existing fetch metric FailedFetchRequestsPerSec for Fetch, which record all topic cumulative and individual topic stats. | FailedShareFetchRequestsPerSec | Meter | ShareGroupMetrics | topic:{topic} | The share fetch request rate for requests that failed. | kafka.server:type=ShareGroupMetrics,name=FailedShareFetchRequestsPerSec kafka.server:type=ShareGroupMetrics,name=FailedShareFetchRequestsPerSec,topic={topic} |
| Similar to above defined share fetch metrics, this metirc tracks the number of share release requests stats. | TotalShareReleaseRequestsPerSec | Meter | ShareGroupMetrics | topic:{topic} | The share release request rate per second. | kafka.server:type=ShareGroupMetrics,name=TotalShareReleaseRequestsPerSec kafka.server:type=ShareGroupMetrics,name=TotalShareReleaseRequestsPerSec,topic={topic} |
| Similar to above defined failed fetch metrics, this metirc tracks the number of failed share release requests stats. | FailedShareReleaseRequestsPerSec | Meter | ShareGroupMetrics | topic:{topic} | The share release request rate for requests that failed | kafka.server:type=ShareGroupMetrics,name=FailedShareReleaseRequestsPerSec kafka.server:type=ShareGroupMetrics,name=FailedShareReleaseRequestsPerSec,topic={topic} |
| Similar to above defined failed fetch metrics, this metirc tracks the number of failed share acknowledgement requests stats. | FailedShareAcknowledgementRequestsPerSec | Meter | ShareGroupMetrics | topic:{topic} | The share acknowledgement request rate for requests that failed | kafka.server:type=ShareGroupMetrics,name=FailedShareAcknowledgementRequestsPerSec kafka.server:type=ShareGroupMetrics,name=FailedShareAcknowledgementRequestsPerSec,topic={topic} |
In shared consumption, topic-partitions are locked before fetching data to ensure exclusive access for each share fetch request. Multiple consumers can request the same partitions, but only a subset can acquire locks. This metric represents the ratio of successfully fetched topic-partitions to the total requested in a share fetch. A low ratio indicates that many requests are unable to fetch from a significant portion of the requested partitions. Potential Causes for Low Ratios:
This metric, in conjunction with other metrics and configuration parameters, can help optimize shared consumption by adjusting consumer counts, partition limits, and partition assignments. | RequestTopicPartitionsFetchRatio | Meter | ShareGroupMetrics | The ratio of topic-partitions acquired to the total number of topic-partitions in share fetch request. | kafka.server:type=ShareGroupMetrics,name=RequestTopicPartitionsFetchRatio | |
| Specifies the number of share fetch requests where no topic-partition can be acquired for fetch. | RequestTopicPartitionsFetchEmptyCount | Gauge | ShareGroupMetrics | The number of share fetch requests completed without any acquired topic-partition. | kafka.server:type=ShareGroupMetrics,name=RequestTopicPartitionsFetchEmptyCount | |
Tracks the number of client timeouts acknowledging share fetches. Timeout is set by group.share.record.lock.duration.ms. | AcquisitionLockTimeoutCount | Gauge | SharePartitionMetrics | share-partition : {group-topic-partition} | The number of times acquisition lock timed out for share partition. | kafka.server:type=SharePartitionMetrics,name=AcquisitionLockTimeoutCount,share-partition={group-topic-partition} |
Defines the maximum number of in-flight messages per share partition, as set by group.share.partition.max.record.locks. A share partition cannot be acquired for fetching messages if this limit is reached. | InFlightMessageCount | Gauge | SharePartitionMetrics | share-partition : {group-topic-partition} | The number of in-flight messages for the share partition. | kafka.server:type=SharePartitionMetrics,name=InFlightMessageCount,share-partition={group-topic-partition} |
| Specifies the number of batches currently being processed by a share partition. This metric provides insights into memory consumption and performance bottlenecks, aiding in optimization efforts. | InFlightBatchCount | Gauge | SharePartitionMetrics | share-partition : {group-topic-partition} | The number of in-flight batches for the share partition. | kafka.server:type=SharePartitionMetrics,name=InFlightBatchCount,share-partition={group-topic-partition} |
| Specifies the number of messages in a single in-flight batch. | InFlightBatchMessageCount | Gauge | SharePartitionMetrics | share-partition : {group-topic-partition} | The number of messages in the in-flight batch. | kafka.server:type=SharePartitionMetrics,name=InFlightBatchMessageCount,share-partition={group-topic-partition} |
Similar to IncrementalFetchSessionEvictionsPerSec | ShareSessionEvictionsPerSec | Meter | ShareSessionCache | The share session eviction rate per second. | kafka.server:type=ShareSessionCache,name=ShareSessionEvictionsPerSec | |
Similar to NumIncrementalFetchPartitionsCached | SharePartitionsCount | Gauge | ShareSessionCache | The number of cached share partitions. | kafka.server:type=ShareSessionCache,name=SharePartitionsCount | |
Similar to NumIncrementalFetchSessions | ShareSessionsCount | Gauge | ShareSessionCache | The number of cached share sessions. | kafka.server:type=ShareSessionCache,name=ShareSessionsCount | |
Similar to ExpiresPerSec for other existing delayed operations i.e. DelayedFetch , etc. | ExpiresPerSec | Meter | DelayedShareFetchMetrics | The expired delayed share fetch operation rate per second. | kafka.server:type=DelayedShareFetchMetrics,name=ExpiresPerSec | |
Add ShareFetch to existing MessageConversionsPerSec metric. | MessageConversionsPerSec (ShareFetch added to existing metric for Produce and Fetch) | Meter | BrokerTopicMetrics | topic : {topic} | The message format conversion rate, for Produce, Fetch or ShareFetch requests, per topic. Omitting ‘topic={…}’ will yield the all-topic rate. | kafka.server:type=BrokerTopicMetrics,name={Produce|Fetch|ShareFetch}MessageConversionsPerSec,topic=([-.\w]+) |
Add ShareFetch to existing purgatory metrics. | PurgatorySize (ShareFetch) | Gauge | DelayedOperationPurgatory | delayedOperation : ShareFetch | The number of requests waiting in the share fetch purgatory. This is high if share consumers use a large value for fetch.wait.max.ms | kafka.server:type=DelayedOperationPurgatory,delayedOperation=ShareFetch,name=PurgatorySize |
Add ShareFetch to existing purgatory metrics. | NumDelayedOperations (ShareFetch) | Gauge | DelayedOperationPurgatory | delayedOperation : ShareFetch | The number of delayed operations for share fetch purgatory. | kafka.server:type=DelayedOperationPurgatory,delayedOperation=ShareFetch,name=NumDelayedOperations |
Move share-group-metrics defined in KIP-932 to Yammer based metrics.
Proposed Changes
We are proposing to add the above metrics. This will give a more insightful set of metrics for those using the Cooperative Consumption.
Compatibility, Deprecation, and Migration Plan
These are new metrics and as such shouldn't have compatibility concerns.
Test Plan
Unit and integration tests.
Rejected Alternatives
- The Meter metrics can be either Yammer or Kafka Metrics. To ensure uniformity in metric naming conventions and to align with existing Fetch and ShareFetch metrics (such as PurgatorySize), we suggest using Yammer metrics for all metrics introduced in this KIP, rather than Kafka Metrics.