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

Consuming Messages

This guide shows how to consume messages from Korvet using Kafka clients.

Using kafka-console-consumer

The simplest way to consume messages:

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic my-topic \
  --from-beginning

Java Consumer

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n",
                         record.offset(), record.key(), record.value());
    }
}

Python Consumer

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'my-topic',
    bootstrap_servers='localhost:9092',
    group_id='my-consumer-group',
    auto_offset_reset='earliest'
)

for message in consumer:
    print(f"Offset: {message.offset}, Value: {message.value}")

Offset Management

Korvet tracks consumer offsets to ensure messages are not lost or duplicated:

  • Auto-commit: Offsets are automatically committed periodically

  • Manual commit: You can control when offsets are committed

Offset Commit Reliability

Korvet ensures that committed offsets are durable and safe:

  • Persist-before-acknowledge ordering: When you commit an offset, Korvet first persists the committed offset to Redis, then acknowledges the messages in the stream. This ensures that if a broker crashes during the commit operation, the committed offset is never lost, preventing message loss.

  • Automatic reconciliation: When a consumer group starts fetching from a partition, Korvet automatically reconciles any pending messages that were already committed but not yet acknowledged (due to historical crashes). This cleanup runs transparently and ensures that the stream’s pending entries list stays consistent with committed offsets.

Out-of-Range Offsets and auto.offset.reset

When a consumer attempts to fetch from an offset outside the retained log range, Korvet returns OFFSET_OUT_OF_RANGE with the actual logStartOffset and highWatermark bounds derived from the stream’s first and last entry IDs. This triggers the client’s auto.offset.reset policy:

  • auto.offset.reset=earliest: The consumer resets to the real log start and reads only the records that survived retention.

  • auto.offset.reset=latest: The consumer skips to the current high watermark.

  • auto.offset.reset=none: The consumer surfaces OffsetOutOfRangeException, requiring manual intervention.

Fetches below the log start (after retention trim or segment eviction) or above the high watermark fail with OFFSET_OUT_OF_RANGE. Gaps within the retained range (in-place compaction XDELs or sparse offsets) are still readable — only reads outside the retained range fail.

Because Korvet offsets are timestamp-derived, a partition’s log start is never 0. Calling consumer.seek(partition, 0) now correctly returns OFFSET_OUT_OF_RANGE and defers to your auto.offset.reset setting. Use consumer.seekToBeginning(…​) (which resolves the real earliest offset via ListOffsets) as the portable idiom for reading from the beginning.

Reading from Specific Offset

You can start reading from a specific offset:

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic my-topic \
  --partition 0 \
  --offset 100

Consumer Groups

Consumer groups enable parallel message processing with automatic partition assignment and load balancing.

Stale Entry Recovery

When a consumer group member crashes or disconnects before committing its offsets, Korvet automatically recovers unacknowledged messages:

  • Automatic reclaim: On each fetch, Korvet checks for pending entries (messages delivered but not acknowledged) that have been idle longer than the configured threshold

  • Redis XAUTOCLAIM: Uses Redis Streams' native pending entry list (PEL) recovery to reclaim stale messages

  • Redelivery: Reclaimed messages are delivered to active group members in the next fetch, ensuring no messages are lost

Configuration:

Property Description Default

korvet.broker.group-reclaim-min-idle

Minimum time a pending entry must be idle before it can be reclaimed from another consumer. Should be comfortably above typical commit intervals to avoid stealing in-flight records from live members.

1m (60 seconds)

Set via environment variable:

export KORVET_BROKER_GROUP_RECLAIM_MIN_IDLE=2m

Or in application.yml:

korvet:
  broker:
    group-reclaim-min-idle: 2m
Setting group-reclaim-min-idle too low risks duplicate delivery if messages are reclaimed from live members that haven’t yet committed their offsets.

Session Timeout Management

Korvet includes a background reaper that automatically evicts timed-out consumer group members:

  • Periodic sweep: Runs every 3 seconds (half the minimum session timeout)

  • Automatic rebalance: When a member is evicted, surviving members trigger a rebalance to redistribute partitions

  • No manual intervention: Dead members are removed even when the group is idle with no new joiners

This ensures partitions assigned to crashed consumers are quickly reassigned to healthy members.

Fetch Behavior

Long Polling with fetch.min.bytes

Korvet supports fetch.min.bytes for efficient long-polling, reducing network overhead when message rates are low:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("fetch.min.bytes", "10000");  // Wait for 10KB
props.put("fetch.max.wait.ms", "500");   // Or 500ms timeout

Behavior:

  • Default (≤1 byte): Returns as soon as any messages are available

  • With threshold (>1 byte): Accumulates messages across multiple reads until:

    • The byte threshold is reached, or

    • All partition record budgets are exhausted, or

    • The fetch.max.wait.ms deadline passes

  • Consumer groups: Merges deliveries across multiple polling rounds; all delivered messages enter the pending list

  • Standalone fetches: Re-reads from advanced positions, merging results until the threshold is met

This matches standard Kafka fetch behavior for both standalone and group consumers.

Next Steps