Consumer Group Management

This page describes how Korvet handles Kafka consumer group operations.

This page is implementation-oriented and focuses on Kafka-to-Redis mapping details.

Join Consumer Group (Implicit Group Creation)

Kafka Wire Protocol

JoinGroupRequest (API Key: 11)
JoinGroupRequest {
  group_id: "my-group"
  session_timeout_ms: 45000
  rebalance_timeout_ms: 300000
  member_id: ""                    // empty for first join
  group_instance_id: null
  protocol_type: "consumer"
  protocols: [
    {
      name: "range"                // partition assignment strategy
      metadata: {
        version: 1
        topics: ["orders"]
        user_data: null
      }
    }
  ]
}
JoinGroupResponse (API Key: 11)
JoinGroupResponse {
  throttle_time_ms: 0
  error_code: 0
  generation_id: 1
  protocol_name: "range"
  leader: "consumer-1-uuid"        // first member becomes leader
  member_id: "consumer-1-uuid"     // assigned by broker
  members: [
    {
      member_id: "consumer-1-uuid"
      group_instance_id: null
      metadata: {
        version: 1
        topics: ["orders"]
        user_data: null
      }
    }
  ]
}

Sequence Diagram

JoinGroupRequest Sequence

Redis Commands Detail

JoinGroup is handled by the broker-side group coordinator and consumer-group registry. Redis stream consumer groups are not created during JoinGroup.

Implementation Notes

  • Implicit Creation: Consumer groups are created automatically on first JoinGroup request

  • Initial Position: Redis consumer groups are created lazily on the first Fetch request (not JoinGroup or SyncGroup)

    • The client determines the initial offset based on:

      • OffsetFetchRequest result (if group has committed offsets)

      • auto.offset.reset configuration (if no committed offset exists)

      • ListOffsets API (to resolve "earliest" or "latest" to actual offset)

    • The client sends this offset in the fetch_offset field of the first FetchRequest

    • The broker creates the Redis consumer group using this offset:

      • XGROUP CREATE

    • The broker tracks which groups have been initialized in memory

    • See Consumer Group Workflow and Consumer Group Initialization for complete details

  • Member ID Assignment:

    • Generate UUID for new members (empty member_id in request)

    • Reuse existing member_id for rejoining members

  • Generation ID:

    • Increments on each rebalance

    • Stored in group metadata

    • Used to detect stale member operations

  • Leader Election:

    • First member to join becomes leader

    • Leader performs partition assignment

    • Stored in group metadata

  • Protocol Selection:

    • Broker selects protocol supported by all members

    • Common protocols: range, roundrobin, sticky

  • Rebalance:

    • Triggered when members join/leave

    • All members must rejoin and get new partition assignments

  • Heartbeat: Members send periodic heartbeats to maintain membership (separate HeartbeatRequest)

  • Session Timeout: Member removed if no heartbeat within session_timeout_ms

Delete Consumer Group

Kafka Wire Protocol

DeleteGroupsRequest (API Key: 42)
DeleteGroupsRequest {
  groups_names: ["my-group", "old-group"]
}
DeleteGroupsResponse (API Key: 42)
DeleteGroupsResponse {
  throttle_time_ms: 0
  results: [
    {
      group_id: "my-group"
      error_code: 0            // NONE
    },
    {
      group_id: "old-group"
      error_code: 69           // GROUP_ID_NOT_FOUND
    }
  ]
}

Sequence Diagram

DeleteGroupsRequest Sequence

Redis Commands Detail

DeleteGroups removes broker-side group state, asks the storage layer to delete group artifacts for known stream keys, and deletes explicit committed offsets for those stream keys.

Implementation Notes

  • Group Doesn’t Exist: Return GROUP_ID_NOT_FOUND (error code 69)

  • Active Members:

    • Cannot delete group with active members

    • Return NON_EMPTY_GROUP (error code 68)

    • Check broker in-memory state for active members

  • Storage Cleanup:

    • Group deletion delegates per-stream cleanup to the storage layer

    • Explicit committed offsets are also deleted from the committed-offset store

  • Cascading Effects:

    • Active consumers will get errors on next operation

    • Consumers should handle UNKNOWN_MEMBER_ID and rejoin

  • Performance: The handler builds a stream-key list from known topics and partitions, then performs storage cleanup asynchronously

Reset Consumer Group Offsets (Admin API)

POST /api/v1/consumer-groups/{groupId}/reset-offsets is an HTTP Admin API operation for the control-plane UI. It is not a Kafka wire-protocol request.

Cleanup Detail

Offset reset is group-wide and uses the current topic registry to resolve the stream keys for all known topic partitions.

The broker performs cleanup in this order:

  1. Delete storage-backed Redis consumer group state for the stream-key snapshot.

  2. Delete explicit committed offsets from the committed-offset store for the same streams.

  3. Clear in-memory delivery tracking after storage cleanup succeeds, even when committed-offset deletion later reports an error.

Deleting storage-backed group state first keeps explicit committed offsets authoritative if the reset fails partway through. In-memory delivery tracking follows storage-backed group state so active fetches can recreate Redis groups after storage cleanup succeeds.

Implementation Notes

  • Unknown Group: The Admin API returns 404 when the group is absent from active coordinator state and durable committed-offset group IDs.

  • Empty Durable Groups: Groups known only from durable committed offsets are returned by GET /api/v1/consumer-groups and GET /api/v1/consumer-groups/{groupId} with EMPTY state and no members.

  • Fallback Offsets: Clearing storage-backed group state matters because OffsetFetch can fall back to Redis group last-delivered-id when no explicit committed offset exists.

  • Scope: Resetting offsets does not delete the broker’s coordinator metadata for an active group; it only clears offset-related state.

  • Active-group race: Reset is clear-only and does not pause delivery. An actively fetching consumer can recreate the Redis group immediately after the clear, so callers should expect a reset to be eventually consistent for groups with live members.

  • Authorization and audit: This operation performs no authorization check and emits no audit event. Both are intentionally out of scope for this slice and must be added before the endpoint is exposed to untrusted callers.

Related Documentation