kafrust 0.3.2

A pure Rust Kafka client with no librdkafka or C client binding dependency.
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

mod common;

use kafrust::{AdminClient, ClientConfig, DescribeQuorumTopic, Error};

#[tokio::main]
async fn main() -> kafrust::Result<()> {
    let mut config = ClientConfig::new(common::bootstrap_servers_from_env())
        .client_id("kafrust-admin-describe-quorum-example");
    if let Some(servers) = common::controller_bootstrap_servers_from_env() {
        config = config.controller_bootstrap_servers(servers);
    }
    let config = common::apply_security(config)?;
    let admin = AdminClient::new(config);
    let result = admin
        .describe_quorum(&[DescribeQuorumTopic::new("__cluster_metadata").partition(0)])
        .await?;
    if !result.is_success() {
        return Err(Error::Broker {
            code: result.error_code(),
            context: "describe metadata quorum".to_owned(),
        });
    }
    let partition = result
        .topics()
        .iter()
        .flat_map(|topic| topic.partitions())
        .next()
        .ok_or(Error::Unsupported(
            "DescribeQuorum returned no metadata partition",
        ))?;
    println!(
        "api_version={} leader={} epoch={} high_watermark={} voters={} observers={} nodes={}",
        result.api_version(),
        partition.leader_id(),
        partition.leader_epoch(),
        partition.high_watermark(),
        partition.current_voters().len(),
        partition.observers().len(),
        result.nodes().len(),
    );
    Ok(())
}