This version is still in development and is not considered stable yet. For the latest stable version, please use Korvet 0.19!

Consumer Group Initialization State Management

This document describes how Korvet tracks which consumer groups have been initialized in Redis Streams.

This page is internal implementation documentation for contributors and operators debugging consumer-group startup behavior.

Problem Statement

Redis Streams consumer groups must be created with an initial offset using XGROUP CREATE. However, the broker doesn’t know what offset to use until the client sends its first FetchRequest.

The client determines the initial offset based on:

  1. Committed offsets (from OffsetFetchRequest)

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

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

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

Design Decision

Redis consumer groups are created lazily on the first Fetch request, not during SyncGroup or OffsetFetch.

This approach:

  • ✅ Allows the client to control the initial offset

  • ✅ Avoids hardcoding offset assumptions in the broker

  • ✅ Works correctly with auto.offset.reset=earliest and auto.offset.reset=latest

  • ✅ Handles committed offsets correctly

Implementation

Broker State Tracking

The broker maintains in-memory state to track which consumer groups have been initialized:

// In DeliveryTracker
// InitKey is a record: (GroupId groupId, String topic, int partition)
private final Set<InitKey> initializedGroups = ConcurrentHashMap.newKeySet();

public boolean isInitialized(GroupId groupId, String topic, int partition) {
    return initializedGroups.contains(initKey(groupId, topic, partition));
}

public void markInitialized(GroupId groupId, String topic, int partition) {
    initializedGroups.add(initKey(groupId, topic, partition));
}

private static InitKey initKey(GroupId groupId, String topic, int partition) {
    return new InitKey(groupId, topic, partition);
}

First Fetch Request

When FetchHandler receives a Fetch request for a consumer group:

// Check if group is initialized
if (!deliveryTracker.isInitialized(groupId, topic, partition)) {
    // Decode fetch_offset to Redis Stream entry ID
    String entryId = offsetMapper.toEntryId(fetchOffset);

    // Create Redis consumer group
    try {
        commands.xgroupCreate(
            XReadArgs.StreamOffset.from(streamKey, entryId),
            groupId,
            XGroupCreateArgs.Builder.mkstream()
        );
    } catch (RedisBusyException e) {
        // Group already exists (e.g., after broker restart)
        // The existing group position is preserved
        log.debug("Consumer group already exists, using existing position");
    }

    // Mark as initialized
    deliveryTracker.markInitialized(groupId, topic, partition);
}

// Now read using XREADGROUP with >
StreamMessage<String, byte[]> messages = commands.xreadgroup(
    Consumer.from(groupId, consumerId),
    XReadArgs.StreamOffset.lastConsumed(streamKey)
);

Subsequent Fetch Requests

After initialization, all Fetch requests use XREADGROUP with >:

// Group is already initialized - just read
StreamMessage<String, byte[]> messages = commands.xreadgroup(
    Consumer.from(groupId, consumerId),
    XReadArgs.StreamOffset.lastConsumed(streamKey)  // Uses >
);

State Persistence

The initializedGroups set is not persisted to Redis. It’s purely in-memory state.

Broker Restart Behavior

When the broker restarts:

  1. The initializedGroups set is empty

  2. On the first Fetch after restart, the broker attempts XGROUP CREATE

  3. If the group already exists in Redis, XGROUP CREATE fails with RedisBusyException

  4. The broker catches this exception and preserves the existing group position

  5. The broker marks the group as initialized in memory

This ensures correctness even after broker restarts - the consumer group continues from where it left off.

Committed-Offset PEL Reconciliation

An offset commit persists the committed offset first and then issues XACK as a separate round-trip. A broker crash between the two leaves entries in the group’s PEL whose offsets are below the committed offset. Regular group reads follow the last-delivered (>) cursor, so those entries would either linger forever or - once idle long enough for the stale-PEL reclaimer - be redelivered as duplicates the client already committed.

Initialization is therefore also the reconciliation point: after XGROUP CREATE (or the RedisBusyException no-op), the broker claims the stream’s pending entries (XAUTOCLAIM) and acknowledges the ones below the group’s committed offset. Entries at or above the committed offset are left pending for the stale-PEL reclaimer to redeliver. Reconciliation is best-effort - a failure is logged, never blocks the fetch, and is retried on the next group initialization.

Alternative Approaches Considered

Option 1: Check Redis on Every Fetch

Query Redis to check if the consumer group exists before each XREADGROUP:

XINFO GROUPS stream-key

Rejected: Adds an extra Redis round-trip on every Fetch request, which is unacceptable for performance.

Option 2: Create Groups During SyncGroup

Create Redis consumer groups during SyncGroup with a hardcoded offset (e.g., "0" or "$"):

// In SyncGroupHandler
commands.xgroupCreate(streamOffset, groupId, XGroupCreateArgs.Builder.mkstream());

Rejected: The broker doesn’t know what offset to use. The client’s auto.offset.reset configuration is not sent to the broker.

Option 3: Lazy Initialization with Error Handling

Try XREADGROUP first, and if it fails with "NOGROUP", create the group:

try {
    messages = commands.xreadgroup(...);
} catch (NoGroupException e) {
    commands.xgroupCreate(...);
    messages = commands.xreadgroup(...);
}

Rejected: Same problem as Option 2 - we don’t know what offset to use when creating the group.

Related Documentation