krabka-client-consumer

Subscribe-style Kafka consumer client for Rust.

Part of Krabka, a Rust implementation of Apache Kafka-compatible infrastructure and clients.

Overview

krabka-client-consumer owns the high-level consumer behavior above krabka-client-core. It handles subscription, group membership, record polls, offset commits, position validation, cooperative shutdown, and share-group consumption. Use this crate for application-style consumers that join Kafka groups and process records by topic subscription.

Manual single-partition fetches and raw request dispatch stay in krabka-client-core. Admin group inspection stays in krabka-client-admin.

Capabilities

Classic consumer-group lifecycle with JoinGroup, SyncGroup, Heartbeat, Fetch, OffsetCommit, and LeaveGroup.
The consumer group protocol of KIP-848 with ConsumerGroupHeartbeat, server-side assignment and a broker-side pattern subscription.
Range, round-robin, sticky and cooperative-sticky partition assignors.
subscribe by topics or by pattern, unsubscribe, and manual assign with or without a group id.
poll, commit_sync, commit_async, seek, pause, offset queries, and graceful close APIs.
Leader-epoch position validation to handle truncation.
KIP-447 ConsumerGroupMetadata for transactional producers.
Share-group consumer support with explicit acknowledgement and commit.

Kafka Scope

The crate implements the classic consumer group protocol, the consumer group protocol of KIP-848 (group_protocol(GroupProtocol::Consumer)) and manual assignment. These flows cover KIP-429 cooperative sticky assignment, KIP-320 leader-epoch validation, KIP-516 offset wire shapes, KIP-447 consumer group metadata, and KIP-932 share groups.

A consumer built without a subscription cannot join a group later. Build it with subscribe or subscribe_pattern to use a group.

Install

cargo add krabka-client-consumer

For workspace development, use the path dependency from this repository.

Usage

Subscribe to a topic, poll records, and commit offsets:

use std::time::Duration;

use krabka_client_consumer::{AutoOffsetReset, Consumer};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let mut consumer = Consumer::builder()
    .bootstrap("127.0.0.1:9092")
    .group_id("orders-worker")
    .client_id("orders-worker-1")
    .subscribe(["orders".to_string()])
    .auto_offset_reset(AutoOffsetReset::Earliest)
    .build()
    .await?;

for record in consumer.poll(Duration::from_secs(1)).await? {
    println!("{}:{}@{}", record.topic, record.partition, record.offset);
}

consumer.commit_sync().await?;
consumer.close().await?;
# Ok(())
# }

Transport and startup policy

The consumer builder exposes dns_timeout, send_buffer, and receive_buffer, with defaults 10s, 128KiB, and 64KiB. None for a buffer selects the OS default. These overrides apply to topic-pattern resolution, the main client, and the coordinator client. Existing request, DNS selection, reconnect, setup, idle, frame, dispatch, and metadata recovery settings follow the same paths.

Use retry_policy(ConsumerRetryPolicy) to control internal startup attempts, the total startup deadline, startup backoff, and coordinator retries. Defaults remain 90s per attempt, five minutes total startup, startup backoff 500ms/max 1s, and coordinator retry 30s with 100ms/max 1s backoff. The total startup deadline now bounds all attempts and sleeps; a last attempt receives only the remaining time. Startup cancellation before completion does not leave a coordinator task running.

Transport settings are validated before I/O. Zero idle lifetime disables eviction; setup/reconnect maximum values must cover their initial values.

Documentation

License

Apache-2.0. Derivative work of Apache Kafka; see NOTICE.

krabka-client-consumer 0.7.1