kafka_client 0.4.0

A pure Rust Kafka client library with SASL authentication support
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

//! Basic connection example
//!
//! Demonstrates the simplest way to connect to a Kafka cluster
//! and discover broker metadata.
//!
//! # Usage
//!
//! ```bash
//! # Default: connects to localhost:9092
//! cargo run --example basic_connect
//!
//! # Custom bootstrap server
//! KAFKA_BOOTSTRAP=192.168.1.100:9092 cargo run --example basic_connect
//! ```

use kafka_client::Client;

fn get_bootstrap_addrs() -> Vec<String> {
    let bootstrap =
        std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".to_string());
    bootstrap.split(',').map(|s| s.trim().to_string()).collect()
}

#[tokio::main]
async fn main() {
    // Initialize logging (optional, but helpful for debugging)
    let _ = tracing_subscriber::fmt()
        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
        .try_init();

    let addrs = get_bootstrap_addrs();
    println!("=== Connecting to Kafka at {:?} ===", addrs);

    // Build and connect the client
    let client = match Client::builder(addrs)
        .with_client_id("basic-connect-example")
        .build()
        .await
    {
        Ok(c) => c,
        Err(e) => {
            eprintln!("ERROR: Failed to connect to Kafka: {}", e);
            std::process::exit(1);
        }
    };

    println!("SUCCESS: Connected!");

    // Query cluster metadata
    let brokers = client.metadata().get_all_brokers().await;
    println!("Discovered {} brokers:", brokers.len());
    for b in &brokers {
        println!("  Broker {}: {}:{}", b.node_id, b.host, b.port);
    }

    // Clean shutdown
    if let Err(e) = client.close().await {
        eprintln!("WARNING: Error during shutdown: {}", e);
    }
    println!("Connection closed.");
}