DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
| Table of Contents |
|---|
Status
Current state: Under Discussion
...
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.
...
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.
...
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.
...
| Code Block |
|---|
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
| Code Block |
|---|
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).
| Code Block |
|---|
{
"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.
| Code Block |
|---|
{
"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
| 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+", "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:
...
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
| Code Block |
|---|
{
"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
| Code Block |
|---|
{
"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