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(())
# }

Documentation

License

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

krabka-client-consumer 0.6.0