kafkit-client 0.1.9

Kafka 4.0+ pure Rust client.
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

# Groups

`kafkit-client` supports two Kafka group styles:

- Modern consumer groups for partition-owned stream processing.
- Share groups for work-queue style record sharing.

Classic consumer group runtime membership is not implemented. The admin client
can list, describe, and delete classic groups.

## Modern Consumer Groups

Use modern consumer groups for normal Kafka consumption where each partition is
owned by one group member at a time.

Properties:

- Partition ordering is preserved for the assigned member.
- Membership and heartbeats use the modern Kafka group protocol.
- Topic-list and regex subscriptions participate in group assignment.
- Manual assignment is available when group membership is not needed.
- Existing committed offsets remain normal Kafka committed offsets.

```rust,no_run
use kafkit_client::{AutoOffsetReset, KafkaClient};

#[tokio::main]
async fn main() -> kafkit_client::Result<()> {
    let consumer = KafkaClient::new("localhost:9092")
        .topics(["orders", "payments"])
        .consumer("orders-reader")
        .with_auto_offset_reset(AutoOffsetReset::Earliest)
        .connect()
        .await?;

    let records = consumer.poll().await?;
    consumer.commit(&records).await?;
    consumer.shutdown().await?;
    Ok(())
}
```

## Share Groups

Use share groups for work-queue style processing where multiple members can
share records from the same partitions. Records are acquired, processed, then
acknowledged.

```rust,no_run
use kafkit_client::{AcknowledgeType, ConsumerConfig, KafkaShareConsumer};

#[tokio::main]
async fn main() -> kafkit_client::Result<()> {
    let mut consumer =
        KafkaShareConsumer::connect(ConsumerConfig::new("localhost:9092", "orders-share")).await?;

    consumer.subscribe(vec!["orders".to_owned()]).await?;

    let records = consumer.poll().await?;
    for record in records.iter() {
        println!("delivery count: {}", record.delivery_count);
        consumer.acknowledge(record, AcknowledgeType::Accept);
    }

    consumer.commit_sync().await?;
    consumer.shutdown().await?;
    Ok(())
}
```

Acknowledgement types:

- `Accept` marks a record as successfully processed.
- `Release` makes a record available for redelivery.
- `Reject` rejects a record.

## Choosing A Group Type

- Choose a modern consumer group for partition-ordered stream processing.
- Choose a share group for distributed task processing.
- Choose manual assignment for direct partition control without group
  coordination.
- Upgrade classic-group-only clusters before adopting `kafkit-client`
  consumers.