krafka 0.7.0

A pure Rust, async-native Apache Kafka 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

//! Admin client example.
//!
//! Demonstrates how to use the Krafka admin client for cluster management.
//!
//! Run with:
//! ```
//! cargo run --example admin
//! ```

use krafka::admin::AdminClient;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize tracing
    tracing_subscriber::fmt()
        .with_env_filter("krafka=debug")
        .init();

    // Create admin client
    let admin = AdminClient::builder()
        .bootstrap_servers("localhost:9092")
        .client_id("krafka-admin-example")
        .build()
        .await?;

    println!("Admin client connected to Kafka!");

    // List all topics
    println!("\n=== Topics ===");
    let topics = admin.list_topics().await?;
    for topic in &topics {
        println!("  - {}", topic);
    }

    // Describe cluster
    println!("\n=== Cluster ===");
    let cluster = admin.describe_cluster().await?;
    println!("Controller ID: {:?}", cluster.controller_id);
    println!("Brokers:");
    for broker in &cluster.brokers {
        println!(
            "  - ID: {}, Address: {}:{}",
            broker.broker_id, broker.host, broker.port
        );
    }

    // Describe specific topics
    if !topics.is_empty() {
        println!("\n=== Topic Details ===");
        let topic_infos = admin
            .describe_topics(&topics[..1.min(topics.len())])
            .await?;
        for info in topic_infos {
            println!("Topic: {}", info.name);
            println!("  Partitions: {}", info.partition_count());
            for partition in &info.partitions {
                println!(
                    "    Partition {}: leader={}, replicas={:?}, isr={:?}",
                    partition.partition, partition.leader, partition.replicas, partition.isr
                );
            }
        }
    }

    Ok(())
}