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

Consumer: Fetch Messages

This page describes how Korvet handles Kafka FetchRequest operations.

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

Consumer Fetches Messages

Kafka Wire Protocol

FetchRequest (API Key: 1)

FetchRequest {
  replica_id: -1             // -1 for consumer (not replica)
  max_wait_ms: 500           // max time to wait if min_bytes not met
  min_bytes: 1               // min bytes to accumulate before responding
  max_bytes: 52428800        // max bytes for entire response (50MB)
  isolation_level: 0         // 0=read_uncommitted, 1=read_committed
  session_id: 0
  session_epoch: -1
  topics: [
    {
      name: "orders"
      partitions: [
        {
          partition: 0
          fetch_offset: 42        // start reading from offset 42
          log_start_offset: 0
          partition_max_bytes: 1048576  // max 1MB per partition
        }
      ]
    }
  ]
  forgotten_topics: []
}

FetchResponse (API Key: 1)

FetchResponse {
  throttle_time_ms: 0
  error_code: 0
  session_id: 0
  topics: [
    {
      name: "orders"
      partitions: [
        {
          partition: 0
          error_code: 0          // NONE
          high_watermark: 44     // next offset to be written
          last_stable_offset: 44
          log_start_offset: 0
          aborted_transactions: []
          preferred_read_replica: -1
          records: [
            {
              offset: 42
              timestamp: 1234567890000
              key: "order-123"
              value: {"product": "widget", "quantity": 5}
              headers: [
                {key: "source", value: "web"},
                {key: "version", value: "1.0"}
              ]
            },
            {
              offset: 43
              timestamp: 1234567891000
              key: "order-124"
              value: {"product": "gadget", "quantity": 3}
              headers: []
            }
          ]
        }
      ]
    }
  ]
}

Sequence Diagram

Redis Commands Detail

Standalone fetches convert fetch_offset to an exclusive Redis start ID and read with XREAD.

Consumer-group fetches resolve the member and group from broker-side registry state, lazily create Redis consumer groups for first fetches, then read with XREADGROUP … >.

Read sizing

Storage reads are capped by record count, so each partition’s COUNT <n> is a byte budget divided by the topic’s measured average message size (1 KiB until produce samples exist), with a minimum of one record. The byte budget is the smaller of:

the partition’s partition_max_bytes (or korvet.broker.fetch-partition-max-bytes when the client sends 0), and
an equal share of the request’s max_bytes across the request’s partitions that this broker will read (known topics it leads; partitions later rejected by ACL checks are still counted).

The equal share keeps a multi-partition fetch from asking Redis for more data than the whole response may carry. With the Java client defaults (fetch.max.bytes 50 MiB, max.partition.fetch.bytes 25 MiB) and 8 partitions, each partition reads up to 6.25 MiB instead of 25 MiB. Partitions are not filled in request order, because reads are sized before any is issued: an earlier, caught-up partition would otherwise leave later partitions only the one-record minimum. A request max_bytes of 0 or less applies no request-level cap.

The trade-off: when one partition lags while many others are caught up, the lagging partition reads only its share per fetch (for example 512 KiB with 100 partitions on the defaults) where Kafka would return up to its full partition_max_bytes, so it catches up in more, cheaper round trips.

Partition Metadata Fields

Each partition in the FetchResponse includes critical metadata that consumers use for offset management and monitoring:

high_watermark

The high watermark (HWM) is the offset of the next message that will be written to the partition. In other words, it’s one greater than the offset of the last message in the partition.

Calculation

# Get stream info including first and last entry IDs
XINFO STREAM korvet:storage:local:orders:0
# Returns stream metadata including:
#   "last-entry" -> ["1234567890123-5", [...]]

# Convert last entry ID to Kafka offset and add 1
# If last entry ID is "1234567890123-5":
#   last_offset = (1234567890123 << 10) | 5 = 1264197008485893
#   high_watermark = 1264197008485893 + 1 = 1264197008485894

Consumer Usage

Consumers use HWM to determine if they’ve reached the end of the partition
Lag calculation: lag = high_watermark - consumer_position
The endOffsets() API returns the high watermark for each partition

log_start_offset

The log start offset is the offset of the first available message in the partition. This may be greater than 0 if messages have been deleted due to retention policies or manual trimming.

Calculation

# Get stream info including first and last entry IDs
XINFO STREAM korvet:storage:local:orders:0
# Returns stream metadata including:
#   "first-entry" -> ["1234567890100-0", [...]]

# Convert first entry ID to Kafka offset
# If first entry ID is "1234567890100-0":
#   log_start_offset = (1234567890100 << 10) | 0 = 1264197008486400

Consumer Usage

Consumers seeking to "earliest" will start from this offset
If a consumer requests an offset below log_start_offset, Korvet returns OFFSET_OUT_OF_RANGE error
Monitoring tools use this to detect message deletion/trimming

last_stable_offset

The last stable offset (LSO) is the offset up to which all transactions have been completed. Since Korvet does not support transactions, LSO always equals the high watermark.

last_stable_offset = high_watermark  // No transaction support

Implementation Notes

Empty Response: If no messages available and max_wait_ms > 0, wait up to max_wait_ms for new messages
Offset Decoding: Stateless conversion from Kafka offset to Stream entry ID using bit operations
Unknown Topics: Unknown topics are filtered before read execution and returned as UNKNOWN_TOPIC_OR_PARTITION.
High Watermark / Log Start: The broker derives high_watermark and log_start_offset from stream metadata after the read.
Group Initialization: First group fetches create Redis groups using a start ID derived from fetch_offset, not a fixed 0 or $ position.
Delivered Message Tracking: Group fetches track delivered Redis message IDs in broker memory so later offset commits can acknowledge them.
Performance: Reads and metadata lookups are batched, and blocking group fetches may be sliced into shorter polling windows.

Kafka FetchRequest

FetchRequest {
  topics: [
    {
      name: "orders"
      partitions: [
        { index: 0, fetchOffset: 5, maxBytes: 1048576 }
      ]
    }
  ]
  maxWaitMs: 500
  minBytes: 1
}

Redis Commands

XREAD COUNT <n> STREAMS korvet:storage:local:orders:0 <exclusive-start-id>
XGROUP CREATE korvet:storage:local:orders:0 my-group <start-id> MKSTREAM
XREADGROUP GROUP my-group consumer-1 COUNT <n> STREAMS korvet:storage:local:orders:0 >

Response Mapping

Redis returns entries with IDs and the record’s value/key/headers/timestamp fields
Reassemble each entry’s fields into Kafka record format (key, value, headers, timestamp)
Encode each entry ID to Kafka offset: offset = (timestamp << sequenceBits) | sequence
Build FetchResponse with records, partition metadata, and high watermark