DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
| Table of Contents |
|---|
Status
Current state: Under DiscussionAccepted
Discussion thread: here
Vote thread: here
JIRA: here
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
KIP-932 introduced share groups to allow multiple share consumers to consume messages from the same partition concurrently. While this adds flexibility, it also introduces challenges in observability. Currently, there is no visibility into consumption progress at the granularity of individual share partitions, making it difficult to detect imbalances, identify slow share consumers, or troubleshoot performance issues.
Introducing share partition lag provides fine-grained visibility into consumption progress, enabling operators to monitor share group behavior more effectively and make informed operational decisions. It also enables future automation opportunities—external autoscalers such as KEDA could use this lag to dynamically scale share consumers based on real-time workload demand, improving efficiency and reducing manual intervention.
Looking ahead, the plan is to build on this foundation by introducing an assignor that can allocate share group members to partitions based on their partition-level backlogs, ensuring more balanced load distribution and improved overall consumption efficiency.
Proposed Changes
Share Partition Lag Computation
...
AVAILABLE
ACQUIRED
ACKNOWLEDGED
ARCHIVED
To measure the lag, we first need to determine the highest offset in the underlying partition log, which defines the upper boundary of records currently available for consumption.
...
For read-committed consumers, the Log Stable Offset (LSO) is used, as it marks the end of all committed data visible to consumers.
...
Similar to regular consumer groups, this offset will be retrieved using the read-uncommitted isolation level. Consequently, the Log End Offset (LEO) will be used as the reference point for measuring lag, as it represents the latest offset in the partition, including both committed and uncommitted records.
Based on this, the lag for a share partition is defined as:
...
For example, consider the following topic - partition:
| Code Block |
|---|
+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+
| 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | <- offset
| Archv | Archv | Acqrd | Avail | Acqrd | Acked | Archv | Avail | Avail | Avail | Avail | <- state
| | | 1 | 2 | 1 | | | | | | | <- delivery count
+-------+-------+---^---+-------+-------+-------+-------+-------+-------+---^---+-------+
| |
+-- Share-partition start offset (SPSO) +-- Share-partition end offset (SPEO) |
...
To make this information available to users, the Group Coordinator retrieves it through the ReadShareGroupStateSummary API and computes the share-partition lag, which is then included in the response to the DescribeShareGroupOffsets request invoked by Admin.listShareGroupOffsets(). To calculate the lag, the Group Coordinator issues an Admin.listOffsets() call to fetch the end offset of the underlying partition and then applies the share-partition lag formula defined above to derive the lag value.Looking ahead, the plan is to implement an assignor that allocates members to partitions based on partition-level backlogs
| Info |
|---|
The SPEO is intentionally excluded from both the external interfaces and the share partition lag calculations, since future changes may allow sparse in-flight records, and the distance between the SPSO and the SPEO can vary significantly. The concept of lag introduced in this KIP is therefore designed to remain flexible and extensible to accommodate such future evolutions. |
Public Interfaces
Client API changes
...
| Code Block |
|---|
{
"apiKey": 90,
"type": "response",
"name": "DescribeShareGroupOffsetsResponse",
// Version 0 is the initial version (KIP-932).
// Version 1 adds Lag (KIP-share-lag).
"validVersions": "0-1",
"flexibleVersions": "0+",
// Supported errors:
// - GROUP_AUTHORIZATION_FAILED (version 0+)
// - TOPIC_AUTHORIZATION_FAILED (version 0+)
// - NOT_COORDINATOR (version 0+)
// - COORDINATOR_NOT_AVAILABLE (version 0+)
// - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
// - GROUP_ID_NOT_FOUND (version 0+)
// - INVALID_REQUEST (version 0+)
// - UNKNOWN_SERVER_ERROR (version 0+)
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "Groups", "type": "[]DescribeShareGroupOffsetsResponseGroup", "versions": "0+",
"about": "The results for each group.", "fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The group identifier." },
{ "name": "Topics", "type": "[]DescribeShareGroupOffsetsResponseTopic", "versions": "0+",
"about": "The results for each topic.", "fields": [
{ "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
{ "name": "TopicId", "type": "uuid", "versions": "0+",
"about": "The unique topic ID." },
{ "name": "Partitions", "type": "[]DescribeShareGroupOffsetsResponsePartition", "versions": "0+", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "StartOffset", "type": "int64", "versions": "0+",
"about": "The share-partition start offset." },
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The leader epoch of the partition." },
{ "name", "Lag", "type": "int64", "versions": "1+", "ignorable": "true", "default": -1,
"about": "The share-partition lag." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The partition-level error code, or 0 if there was no error." },
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The partition-level error message, or null if there was no error." }
]}
]},
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The group-level error code, or 0 if there was no error." },
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The group-level error message, or null if there was no error." }
]}
]
} |
...