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

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 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 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 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

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. 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 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

...

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 offsetpartition start 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." }
        ]}
      ]}
    ]}
  ]
}

...

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

...

{
  "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:

...

{
  "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": "InFlightTerminalRecordsDeliveryCompleteCount", "type": "int32", "versions": "0+", "taggedVersions": "0+", "tag": 0, "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": "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." }
    ]}
  ]
}  

...

{
  "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": "InFlightTerminalRecordsDeliveryCompleteCount", "type": "int32", "versions": "0+", "taggedVersions": "0+", "tag": 0, "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." }
    ]}
  ]
}

...