Status

Current state: Under DiscussionAccepted

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread] 

Vote thread:  here

JIRA: here [Change the link from KAFKA-1 to your own ticket]

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

In KIP-932 , introduced share groups were introduced to allow multiple share consumers to simultaneously consume messages from a single topic the same partition concurrently. While this improves throughput and adds flexibility, it also introduces complexities in monitoring, particularly regarding the consumption progress of individual share partitionschallenges in observability. Currently, there is no visibility into the lag consumption progress at the granularity of each individual share partitionpartitions, which makes making it challenging difficult to detect imbalances in consumption, identify slow share consumers, or troubleshoot performance bottlenecksissues.

Introducing the concept of lag in the context of share partition will provide share partition lag provides fine-grained observabilityvisibility into consumption progress, 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 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 the number of consumers in a share group based share consumers based on real-time workload 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.

...

, 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

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 Share-partition start offset (SPSO)  and Share-partition end offset (SPEO) ) 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 = (To measure the lag, we first need to determine the 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:

, which defines the upper boundary of records currently available for consumption. 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:

share-partition lag = (highest offset in the underlying partition log) – (share-partition start offset) + 1 – (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)

...

highest offset in the underlying partition log = 1110

share-partition start offset = 2

...

Thus, share-partition lag = 11 10 - 2 + 1 - 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.

Handling Control and Compacted Records

The concept of share partition lag is closely analogous to the traditional partition lag for a regular consumer, which is typically defined as:

partition lag = offset of the latest record produced - committed offset

However, not all offsets in the partition log correspond to records to be processed by the consumer application, such as:

1. Control Records — Internal records used by the broker for managing transactions, not carrying user data.
2. Compacted Records — Offsets that no longer correspond to records due to log compaction (applicable to compacted topics).

In a regular consumer scenario, lag computation includes these offsets. The consumer remains unaware of their nature until it fetches data up to those offsets. As a result, these offsets continue to be counted as part of the lag until the client reads beyond them and advances its committed offset.

Similarly, in the context of share partitions, lag computation follows comparable semantics. All non-record offsets that lie after the SPEO  are included in the lag. However, offsets within the in-flight boundary (between  SPSO  and SPEO ) require additional handling so that the lag more accurately reflects the number of records to be processed.

When a share consumer identifies missing records while processing record batches, it reports them to the share partition through special "gap" acknowledgements which indicate records that have been removed by compaction. Upon receiving such acknowledgements, the share partition:

• Marks the corresponding offsets as ARCHIVED, and
• Excludes them from subsequent lag computations.

Until these acknowledgements are received, such offsets remain included in the lag, even though they do not correspond to user-visible data. Consequently, the share partition lag may temporarily include these offsets.

Persistence

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.

...

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 backlogsfetch the end offset of the underlying partition and then applies the share-partition lag formula defined above to derive the lag value.

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.

...

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 InFlightTerminalRecordsDeliveryCompleteCount. 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": "InFlightTerminalRecordsDeliveryCompleteCount", "type": "int32", "versions": "1+", "ignorable": "true", "default": "-1",
          "about": "The number of ACKNOWLEDGED / ARCHIVED records offsets greater than or equal to share-partition start offset offset for which delivery has been completed."},
        { "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 InFlightTerminalRecordsDeliveryCompleteCount.

{
  "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": "InFlightTerminalRecordsDeliveryCompleteCount", "type": "int32", "versions": "1+", "ignorable": "true", "default": "-1",
          "about": "The number of ACKNOWLEDGED / ARCHIVED records offsets greater than or equal to share-partition start offset for which delivery has been completed."}
  ]
}

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+", "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." }
    ]}
  ]
}

Records

The InFlightTerminalRecords DeliveryCompleteCount 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

ShareSnapshotValue schema

{
  "apiKey": 0,
  "type": "coordinator-value",
  "name": "ShareSnapshotValue",
  "validVersions": "0",

{
  "apiKeyflexibleVersions": 1"0+",
  "fields": [
    { "name": "SnapshotEpoch", "type": "coordinator-valueint32", "versions": "0+",
      "nameabout": "ShareUpdateValue"The snapshot epoch." },
    { "name": "validVersionsStateEpoch", "type": "0int32",
  "flexibleVersionsversions": "0+",
      "fieldsabout": ["The state epoch for this share-partition." },
    { "name": "SnapshotEpochLeaderEpoch", "type": "int32", "versions": "0+",
      "about": "The snapshotleader epoch of the share-partition." },
    { "name": "LeaderEpochStartOffset", "type": "int32int64", "versions": "0+",
      "about": "The leader epoch of the share-partition start offset." },
	{ "name": "DeliveryCompleteCount", "type": "int32",   {"versions": "0+", "nametaggedVersions": "StartOffset0+", "typetag": "int64"0, "versionsdefault": "0+-1",
      "about": "The share-partition start offset,number of offsets greater than or -1equal ifto theshare-partition start offset is not being updated." for which delivery has been completed."},
    { "name": "EndOffsetCreateTimestamp", "type": "int64", "versions": "0+",
  "taggedVersions": "0+", "tag": 0, "defaultabout": "-1",
The time     "about": "The share-partition end offset."at which the state was created." },
    { "name": "InFlightTerminalRecordsWriteTimestamp", "type": "int32int64", "versions": "0+", "taggedVersions": "0+", "tag": 1, "default": "-1",
      "about": "The numbertime ofat ACKNOWLEDGEDwhich /the ARCHIVEDstate recordswas greaterwritten than or equal to share-partition start offset"rewritten." },
    { "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." }
    ]}
  ]
}

  

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 of the share-partition." },
    { "name": "EndOffsetStartOffset", "type": "int64", "versions": "0+",
 "taggedVersions": "0+", "tag": 0, "default     "about": "-1",
      "about": "The share-partition end offset."The share-partition start offset, or -1 if the start offset is not being updated." },
    { "name": "InFlightTerminalRecordsDeliveryCompleteCount", "type": "int32", "versions": "0+", "taggedVersions": "0+", "tag": 10, "default": "-1",
      "about": "The number of ACKNOWLEDGED / ARCHIVED records offsets greater than or equal to share-partition start offset for which delivery has been completed."},
    { "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_REQUEST at 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.java to verify the adminClient.listShareGroupOffsets returns and displays the lag for share partition when kafka-share-groups.sh --describe --offsets is used.

Rejected Alternatives

None