kafkit-client 0.1.6

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
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
114
115
116
117
118
119
120
121
122
123

# Admin

`KafkaAdmin` provides Kafka cluster, topic, group, config, feature, and SCRAM
credential operations through the Kafka protocol.

Use admin calls for application setup, deployment checks, tooling, and
operational workflows. Keep topic creation and config mutation explicit; these
operations affect cluster state.

## Connect

```rust,no_run
use kafkit_client::{AdminConfig, KafkaAdmin};

#[tokio::main]
async fn main() -> kafkit_client::Result<()> {
    let admin = KafkaAdmin::connect(AdminConfig::new("localhost:9092")).await?;

    let cluster = admin.describe_cluster().await?;
    println!("cluster id: {}", cluster.cluster_id);
    println!("controller id: {}", cluster.controller_id);

    Ok(())
}
```

## Topics

```rust,no_run
use kafkit_client::{AdminConfig, KafkaAdmin, NewPartitions, NewTopic};

#[tokio::main]
async fn main() -> kafkit_client::Result<()> {
    let admin = KafkaAdmin::connect(AdminConfig::new("localhost:9092")).await?;

    admin
        .create_topics([
            NewTopic::new("orders", 3, 1)
                .with_config("cleanup.policy", "compact,delete"),
        ])
        .await?;

    admin
        .create_partitions([("orders", NewPartitions::increase_to(6))])
        .await?;

    let topics = admin.list_topics().await?;
    println!("topics: {}", topics.len());

    let descriptions = admin.describe_topics(["orders"]).await?;
    for topic in descriptions {
        println!("{} has {} partitions", topic.name, topic.partitions.len());
    }

    Ok(())
}
```

`delete_topics(...)` is also available. Treat deletes as irreversible from the
client's point of view.

## Consumer Groups

```rust,no_run
use kafkit_client::{AdminConfig, KafkaAdmin};

#[tokio::main]
async fn main() -> kafkit_client::Result<()> {
    let admin = KafkaAdmin::connect(AdminConfig::new("localhost:9092")).await?;

    let groups = admin.list_consumer_groups().await?;
    for group in &groups {
        println!("group: {}", group.group_id);
    }

    let descriptions = admin.describe_consumer_groups(["orders-reader"]).await?;
    for group in descriptions {
        println!("{} members: {}", group.group_id, group.members.len());
    }

    Ok(())
}
```

## Configs

Describe configs before mutating them so tooling can show the current value and
source.

```rust,no_run
use kafkit_client::{
    AdminConfig, AlterConfigOp, ConfigResource, KafkaAdmin,
};

#[tokio::main]
async fn main() -> kafkit_client::Result<()> {
    let admin = KafkaAdmin::connect(AdminConfig::new("localhost:9092")).await?;

    let configs = admin
        .describe_configs([ConfigResource::topic("orders")])
        .await?;
    for config in &configs {
        if let Some(entry) = config.entry("cleanup.policy") {
            println!("cleanup.policy = {:?}", entry.value);
        }
    }

    admin
        .incremental_alter_configs([(
            ConfigResource::topic("orders"),
            vec![AlterConfigOp::set("retention.ms", "604800000")],
        )])
        .await?;

    Ok(())
}
```

## Security Administration

The admin client includes SCRAM credential upsert helpers. Use them only from
trusted operational tooling, and protect passwords before they reach logs or
tracing subscribers.