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
In KIP-932, share groups were introduced to allow multiple consumers to simultaneously consume messages from a single topic partition. While this improves throughput and flexibility, it also introduces complexities in monitoring, particularly regarding the consumption progress of individual share partitions. Currently, there is no visibility into the lag at the granularity of each share partition, which makes it challenging to detect imbalances in consumption, identify slow consumers, or troubleshoot performance bottlenecks.
Introducing the concept of lag in the context of share partition will provide fine-grained observability, enabling operators to monitor share group progress more accurately, detect potential issues proactively, and optimize resource allocation. This enhancement will improve reliability and operational transparency, making share group consumption easier to manage in production environments.
In addition, introducing share-partition lag opens the door to autoscaling capabilities. External event-driven autoscalers, such as KEDA, could leverage this lag to dynamically scale the number of consumers in a share group based on real-time demand. This ensures that workloads are processed efficiently under varying traffic patterns, improves utilization of cluster resources, and reduces operational overhead by automating scaling decisions.
Proposed Changes
The lag for a share partition, unlike the lag for a regular partition in consumer groups, is more complex to compute. By definition, lag should capture the number of records that are either still being processed or have not yet been processed. In the case of share groups, since a single partition can be consumed by multiple share consumers, record processing does not always occur in strict order.
Furthermore, in-flight records (records that lie between startOffset and endOffset) in a share partition can exist in one of the following four states:
AVAILABLE
ACQUIRED
ACKNOWLEDGED
ARCHIVED
Based on this, the lag for a share partition is defined as:
share-partition lag = (highest offset in the underlying partition log) – (share-partition start offset) – (in-flight records that are already processed)
For example, consider the following topic-partition:
+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+
| 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)
The backlog for this partition is 7. The records to be consumed are offsets 2, 3, 4, 7, 8 and 9 (these are in-flight) and then offset 10 is after the SPEO.
As per the lag definition above,
highest offset in the underlying partition log = 11
share-partition start offset = 2
in-flight records that are already processed = 2 (offsets 5 and 6)
Thus, share-partition lag = 11 - 2 - 2 = 7
Note: The value for the highest offset in the underlying partition log can depend on the use case. If the isolation level for share consumers is read-committed, then the Log Stable Offset (LSO) is the appropriate choice. In case the isolation level is read-uncommitted, the Log End Offset (LEO) is more suitable.
The start offset is regularly updated and persisted through the writeShareGroupState RPC call in the __share_group_state topic, managed by the Share Coordinator. This KIP introduces a mechanism to calculate and persist the count of in-flight records that have already been processed, and makes it available to the users.
For each share partition, the leader will compute this new information and include it in the writeShareGroupState RPC, which will be persisted via the ShareSnapshot and ShareUpdate records in the __share_group_state topic. The same information will also be returned in the response of ReadShareGroupStateSummary.
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.
Public Interfaces
Client API changes
AdminClient
ListShareGroupOffsetsResult
A very small breaking change is made compared with KIP-932 to accommodate the lag. This is permitted because it is still marked as an evolving interface.
The return type of the all() method is changed from KafkaFuture<String, Map<TopicPartition, OffsetAndMetadata>> to KafkaFuture<String, Map<TopicPartition, SharePartitionOffsetInfo>>.
The partitionsToOffsetAndMetadata method is removed and replaced with partitionsToOffsetInfo.
package org.apache.kafka.clients.admin;
/**
* The result of the {@link Admin#listShareGroupOffsets(Map<String, ListShareGroupOffsetsSpec>, ListShareGroupOffsetsOptions)} call.
* <p>
* The API of this class is evolving, see {@link Admin} for details.
*/
@InterfaceStability.Evolving
public class ListShareGroupOffsetsResult {
/**
* Return a future which yields all Map<String, Map<TopicPartition, SharePartitionOffsetInfo> objects, if requests for all the groups succeed.
*/
public KafkaFuture<Map<String, Map<TopicPartition, SharePartitionOffsetInfo>>> all() {
}
/**
* Return a future which yields a map of topic partitions to offset information for the specified group.
*/
public KafkaFuture<Map<TopicPartition, SharePartitionOffsetInfo>> partitionsToOffsetInfo(String groupId) {
}
}
SharePartitionOffsetInfo
package org.apache.kafka.clients.admin;
/**
* This class is used to contain the offset and lag information for a share-partition.
@InterfaceStability.Evolving
public class SharePartitionOffsetInfo {
public SharePartitionOffsetInfo(long startOffset, Optional<Integer> leaderEpoch, Optional<Long> lag);
public long startOffset();
public Optional<Integer> leaderEpoch();
public Optional<Long> lag();
}
Command-line tools
kafka-share-groups.sh
A new column LAG is added to the output from kafka-share-groups.sh --describe --offsets. The value is displayed as - if the lag is not available.
Kafka protocol changes
This KIP introduces new versions of the following APIs:
WriteShareGroupState API
Request schema
Version 1 adds the new field InFlightTerminalRecords. This provides information about the number of records in the Share Partition, that lies after the startOffset, and are in a Terminal state (ACKNOWLEDGED / ARCHIVED).
{
"apiKey": 85,
"type": "request",
"listeners": ["broker"],
"name": "WriteShareGroupStateRequest",
"validVersions": "0",
"validVersions": "0-1",
"flexibleVersions": "0+",
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0+",
"about": "The group identifier." },
{ "name": "Topics", "type": "[]WriteStateData", "versions": "0+",
"about": "The data for the topics.", "fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+",
"about": "The topic identifier." },
{ "name": "Partitions", "type": "[]PartitionData", "versions": "0+",
"about": "The data for the partitions.", "fields": [
{ "name": "Partition", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "StateEpoch", "type": "int32", "versions": "0+",
"about": "The state epoch of the share-partition." },
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The leader epoch of the share-partition." },
{ "name": "StartOffset", "type": "int64", "versions": "0+",
"about": "The share-partition start offset, or -1 if the start offset is not being written." },
{ "name": "InFlightTerminalRecords", "type": "int32", "versions": "1+", "ignorable": "true", "default": "-1",
"about": "The number of ACKNOWLEDGED / ARCHIVED records greater than or equal to share-partition start offset"},
{ "name": "StateBatches", "type": "[]StateBatch", "versions": "0+",
"about": "The state batches for the share-partition.", "fields": [
{ "name": "FirstOffset", "type": "int64", "versions": "0+",
"about": "The first offset of this state batch." },
{ "name": "LastOffset", "type": "int64", "versions": "0+",
"about": "The last offset of this state batch." },
{ "name": "DeliveryState", "type": "int8", "versions": "0+",
"about": "The delivery state - 0:Available,2:Acked,4:Archived." },
{ "name": "DeliveryCount", "type": "int16", "versions": "0+",
"about": "The delivery count." }
]}
]}
]}
]
}
Response schema
Version 1 is the same as version 0.
ReadShareGroupStateSummary
Request schema
Version 1 is the same as version 0.
Response schema
Version 1 adds the new field InFlightTerminalRecords.
{
"apiKey": 87,
"type": "response",
"name": "ReadShareGroupStateSummaryResponse",
"validVersions": "0-1",
"flexibleVersions": "0+",
// - NOT_COORDINATOR (version 0+)
// - COORDINATOR_NOT_AVAILABLE (version 0+)
// - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
// - GROUP_ID_NOT_FOUND (version 0+)
// - UNKNOWN_TOPIC_OR_PARTITION (version 0+)
// - FENCED_LEADER_EPOCH (version 0+)
// - INVALID_REQUEST (version 0+)
"fields": [
{ "name": "Results", "type": "[]ReadStateSummaryResult", "versions": "0+",
"about": "The read results.", "fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+",
"about": "The topic identifier." },
{ "name": "Partitions", "type": "[]PartitionResult", "versions": "0+",
"about" : "The results for the partitions.", "fields": [
{ "name": "Partition", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The error message, or null if there was no error." },
{ "name": "StateEpoch", "type": "int32", "versions": "0+",
"about": "The state epoch of the share-partition." },
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The leader epoch of the share-partition." },
{ "name": "StartOffset", "type": "int64", "versions": "0+",
"about": "The share-partition start offset." },
{ "name": "InFlightTerminalRecords", "type": "int32", "versions": "1+", "ignorable": "true", "default": "-1",
"about": "The number of ACKNOWLEDGED / ARCHIVED records greater than or equal to share-partition start offset"}
]
}
DescribeShareGroupOffsets
Request schema
Version 1 is the same as version 0.
Response schema
The new field LAG is added
{
"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+", "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." }
]}
]
}
Records
The InFlightTerminalRecords received in the writeShareGroupState RPC is also persisted by the Share Coordinator. In order to persist this information, the schemas for the following records are also updated:
ShareSnapshotShareUpdate
The version remains the same, instead the new field is added as a tagged field, with a default value. This has been done to ensure that records already written from a certain versioned broker can be read by a different versioned broker, in case the broker is upgraded or rolled-back.
ShareSnapshotKey
Remains the same; no change introduced here
ShareSnapshotValue schema
{
"apiKey": 0,
"type": "coordinator-value",
"name": "ShareSnapshotValue",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "SnapshotEpoch", "type": "int32", "versions": "0+",
"about": "The snapshot epoch." },
{ "name": "StateEpoch", "type": "int32", "versions": "0+",
"about": "The state epoch for this share-partition." },
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The leader epoch of the share-partition." },
{ "name": "StartOffset", "type": "int64", "versions": "0+",
"about": "The share-partition start offset." },
{ "name": "InFlightTerminalRecords", "type": "int32", "versions": "0+", "taggedVersions": "0+", "tag": 0, "default": "-1",
"about": "The number of ACKNOWLEDGED / ARCHIVED records greater than or equal to share-partition start offset"},
{ "name": "CreateTimestamp", "type": "int64", "versions": "0+",
"about": "The time at which the state was created." },
{ "name": "WriteTimestamp", "type": "int64", "versions": "0+",
"about": "The time at which the state was written or rewritten." },
{ "name": "StateBatches", "type": "[]StateBatch", "versions": "0+",
"about": "The state batches.", "fields": [
{ "name": "FirstOffset", "type": "int64", "versions": "0+",
"about": "The first offset of this state batch." },
{ "name": "LastOffset", "type": "int64", "versions": "0+",
"about": "The last offset of this state batch." },
{ "name": "DeliveryState", "type": "int8", "versions": "0+",
"about": "The delivery state - 0:Available,2:Acked,4:Archived." },
{ "name": "DeliveryCount", "type": "int16", "versions": "0+",
"about": "The delivery count." }
]}
]
}
ShareUpdateKey schema
Remains the same; no change introduced here
ShareUpdateValue schema
{
"apiKey": 1,
"type": "coordinator-value",
"name": "ShareUpdateValue",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "SnapshotEpoch", "type": "int32", "versions": "0+",
"about": "The snapshot epoch." },
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The leader epoch of the share-partition." },
{ "name": "StartOffset", "type": "int64", "versions": "0+",
"about": "The share-partition start offset, or -1 if the start offset is not being updated." },
{ "name": "InFlightTerminalRecords", "type": "int32", "versions": "0+", "taggedVersions": "0+", "tag": 0, "default": "-1",
"about": "The number of ACKNOWLEDGED / ARCHIVED records greater than or equal to share-partition start offset"},
{ "name": "StateBatches", "type": "[]StateBatch", "versions": "0+",
"about": "The state batches that have been updated.", "fields": [
{ "name": "FirstOffset", "type": "int64", "versions": "0+",
"about": "The first offset of this state batch." },
{ "name": "LastOffset", "type": "int64", "versions": "0+",
"about": "The last offset of this state batch." },
{ "name": "DeliveryState", "type": "int8", "versions": "0+",
"about": "The delivery state - 0:Available,2:Acked,4:Archived." },
{ "name": "DeliveryCount", "type": "int16", "versions": "0+",
"about": "The delivery count." }
]}
]
}
Compatibility, Deprecation, and Migration Plan
The existing functionality is not modified. Clusters with upgraded brokers will be able to store and report lag for share partitions.
The RPCs WriteShareGroupState and ReadShareGroupStateSummary are only meant for inter-broker communications and thus have no consequences for clients. If brokers supporting different versions of the RPC are communicating with each other, they both will agree to use the minimum of the highest version supported by each broker, resolving any conflicts. But if by chance a broker running the old version of the code receives any of these requests with version 1, it should consider this an error and return the appropriate error code (
Errors.INVALID_REQUESTat the time of writing).In contrast, DescribeShareGroupOffsets is a client-facing RPC. However, since ListShareGroupOffsetsResult is already annotated with
@InterfaceStability.Evolving, it provides the necessary flexibility to introduce interface modifications without violating API stability guarantees.For the ShareSnapshot and ShareUpdate records, the schema version remains unchanged to maintain compatibility. The newly added field is tagged and assigned a default value, ensuring that older brokers safely ignore the field when reading newer records, while newer brokers correctly populate it with the default value when reading older records — all without requiring a schema version bump.
Test Plan
Updates will be made to the existing tests which verify the functioning of the updated RPCs to see if the new field is correctly persisted and reported.
New unit tests will be added in
tools/src/test/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandTest.javato verify theadminClient.listShareGroupOffsetsreturns and displays the lag for share partition whenkafka-share-groups.sh --describe --offsetsis used.
Rejected Alternatives
None