krabka-client-consumer 0.7.1

Subscribe-style consumer client for Apache Kafka in Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

# krabka-client-consumer

[![Crates.io](https://img.shields.io/crates/v/krabka-client-consumer.svg)](https://crates.io/crates/krabka-client-consumer)
[![Docs.rs](https://docs.rs/krabka-client-consumer/badge.svg)](https://docs.rs/krabka-client-consumer)
[![CI](https://github.com/krabka-io/krabka-client-rs/actions/workflows/ci.yml/badge.svg)](https://github.com/krabka-io/krabka-client-rs/actions/workflows/ci.yml)

Subscribe-style Kafka consumer client for Rust.

Part of [Krabka](https://github.com/krabka-io/krabka-client-rs), 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

```sh
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:

```rust,no_run
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

- [API documentation](https://docs.rs/krabka-client-consumer)
- [krabka-client-rs repository](https://github.com/krabka-io/krabka-client-rs)
- [Kafka compatibility matrix](https://github.com/krabka-io/krabka-broker/blob/main/docs/KIP_MATRIX.md)

## License

Apache-2.0. Derivative work of [Apache Kafka](https://kafka.apache.org); see
[NOTICE](https://github.com/krabka-io/krabka-client-rs/blob/main/NOTICE).