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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

mod common;

use std::io::{self, Write};
use std::time::Duration;

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

#[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 partition = partition_from_env()?;
    let pause = pause_from_env()?;

    let mut producer = common::apply_security(
        ProducerConfig::new(bootstrap_servers).client_id("kafrust-producer-failover-example"),
    )?
    .enable_idempotence(true)
    .compression(common::compression_from_env()?)
    .max_retries(5)
    .build()
    .await?;

    let before = producer
        .send(failover_record(&topic, partition, "before"))
        .await?;
    print_metadata("failover before produced", &before)?;

    if !pause.is_zero() {
        println!("failover pause {}ms", pause.as_millis());
        flush_stdout()?;
        tokio::time::sleep(pause).await;
    }

    let after = producer
        .send(failover_record(&topic, partition, "after"))
        .await?;
    print_metadata("failover after produced", &after)?;

    Ok(())
}

fn failover_record(topic: &str, partition: i32, marker: &str) -> ProducerRecord {
    ProducerRecord::to(topic.to_owned())
        .partition(partition)
        .key(format!("kafrust-failover-{marker}-key"))
        .value(format!("hello from kafrust failover {marker}"))
}

fn print_metadata(label: &str, metadata: &RecordMetadata) -> kafrust::Result<()> {
    println!(
        "{label} {}-{}@{}",
        metadata.topic(),
        metadata.partition(),
        metadata.offset()
    );
    flush_stdout()
}

fn partition_from_env() -> kafrust::Result<i32> {
    std::env::var("KAFRUST_PARTITION")
        .map(|value| parse_partition(&value))
        .unwrap_or(Ok(0))
}

fn parse_partition(value: &str) -> kafrust::Result<i32> {
    value
        .trim()
        .parse::<i32>()
        .map_err(|_| Error::Unsupported("KAFRUST_PARTITION must be a partition index"))
}

fn pause_from_env() -> kafrust::Result<Duration> {
    std::env::var("KAFRUST_FAILOVER_PAUSE_MS")
        .map(|value| parse_pause(&value))
        .unwrap_or(Ok(Duration::from_millis(0)))
}

fn parse_pause(value: &str) -> kafrust::Result<Duration> {
    value
        .trim()
        .parse::<u64>()
        .map(Duration::from_millis)
        .map_err(|_| Error::Unsupported("KAFRUST_FAILOVER_PAUSE_MS must be milliseconds"))
}

fn flush_stdout() -> kafrust::Result<()> {
    io::stdout().flush().map_err(Error::from)
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::{parse_partition, parse_pause};

    #[test]
    fn parses_partition() {
        assert_eq!(parse_partition(" 2 ").expect("partition should parse"), 2);
    }

    #[test]
    fn rejects_invalid_partition() {
        assert!(parse_partition("not-a-partition").is_err());
    }

    #[test]
    fn parses_pause() {
        assert_eq!(
            parse_pause(" 1500 ").expect("pause should parse"),
            Duration::from_millis(1500)
        );
    }

    #[test]
    fn rejects_invalid_pause() {
        assert!(parse_pause("one second").is_err());
    }
}