krafka 0.27.0

An async Apache Kafka client in pure Rust: producer, transactions, consumer groups, share groups and admin
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

//! Cluster administration: describe the cluster, then create, describe and
//! delete a topic.
//!
//! Needs a broker at `KAFKA_BOOTSTRAP_SERVERS` (default `localhost:9092`) on
//! which the client may create and delete topics.
//!
//! Run with:
//! ```sh
//! cargo run --example admin
//! ```

use krafka::Kafka;
use krafka::admin::NewTopic;

const TOPIC: &str = "krafka-admin-example";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let bootstrap =
        std::env::var("KAFKA_BOOTSTRAP_SERVERS").unwrap_or_else(|_| "localhost:9092".into());
    let kafka = Kafka::builder(bootstrap)
        .client_id("krafka-admin-example")
        .connect()
        .await?;
    let admin = kafka.admin();

    let cluster = admin.describe_cluster(Default::default()).await?;
    println!(
        "cluster {} (controller {})",
        cluster.cluster_id, cluster.controller_id
    );
    for broker in &cluster.brokers {
        println!("  broker {} at {}:{}", broker.id, broker.host, broker.port);
    }

    // Results are per topic: one topic failing does not fail the call.
    let created = admin
        .create_topics([NewTopic::new(TOPIC, 3, 1)?], Default::default())
        .await?;
    for (name, result) in created {
        match result {
            Ok(()) => println!("created {name}"),
            Err(error) => println!("create {name}: {error}"),
        }
    }

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

    for (name, description) in admin.describe_topics([TOPIC], Default::default()).await? {
        let description = description?;
        println!("{name}:");
        for partition in &description.partitions {
            println!(
                "  partition {} leader={:?} replicas={:?} isr={:?}",
                partition.partition, partition.leader, partition.replicas, partition.isr
            );
        }
    }

    for (name, result) in admin.delete_topics([TOPIC], Default::default()).await? {
        result?;
        println!("deleted {name}");
    }

    admin.close().await?;
    Ok(())
}