kafrust 0.2.28

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
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

mod common;

use kafrust::{Acks, Error, ProducerConfig, ProducerRecord};

#[tokio::main]
async fn main() -> kafrust::Result<()> {
    let bootstrap_servers = common::bootstrap_servers_from_env();
    let topic = std::env::var("KAFRUST_TOPIC").unwrap_or_else(|_| "kafrust-smoke".to_owned());
    let expected_partitions = expected_partitions_from_env()?;
    let count = expected_partitions.as_ref().map_or(2, Vec::len).max(1);

    let mut producer = common::apply_security(
        ProducerConfig::new(bootstrap_servers).client_id("kafrust-keyless-producer-example"),
    )?
    .acks(Acks::Leader)
    .build()
    .await?;

    let mut actual_partitions = Vec::with_capacity(count);
    for index in 0..count {
        let metadata = producer
            .send(
                ProducerRecord::to(topic.clone())
                    .value(format!("hello from kafrust keyless {index}")),
            )
            .await?;
        actual_partitions.push(metadata.partition());
        println!(
            "keyless produced {}-{}@{}",
            metadata.topic(),
            metadata.partition(),
            metadata.offset()
        );
    }

    if expected_partitions
        .as_ref()
        .is_some_and(|expected| expected != &actual_partitions)
    {
        return Err(Error::Unsupported(
            "keyless partition sequence did not match KAFRUST_EXPECT_PARTITIONS",
        ));
    }

    Ok(())
}

fn expected_partitions_from_env() -> kafrust::Result<Option<Vec<i32>>> {
    std::env::var("KAFRUST_EXPECT_PARTITIONS")
        .ok()
        .map(|value| parse_partitions(&value))
        .transpose()
}

fn parse_partitions(value: &str) -> kafrust::Result<Vec<i32>> {
    value
        .split(',')
        .map(str::trim)
        .filter(|partition| !partition.is_empty())
        .map(|partition| {
            partition.parse::<i32>().map_err(|_| {
                Error::Unsupported(
                    "KAFRUST_EXPECT_PARTITIONS must be comma-separated partition indexes",
                )
            })
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parses_expected_partitions() {
        assert_eq!(
            parse_partitions(" 0,1,,2 ").expect("partition list should parse"),
            vec![0, 1, 2]
        );
    }

    #[test]
    fn rejects_invalid_expected_partition() {
        assert!(parse_partitions("0,invalid").is_err());
    }
}