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

mod common;

use kafrust::{ConsumerConfig, ConsumerGroupConfig, Error, OffsetResetPolicy};

#[tokio::main]
async fn main() -> kafrust::Result<()> {
    let bootstrap_servers = common::bootstrap_servers_from_env();
    let group_id =
        std::env::var("KAFRUST_GROUP_ID").unwrap_or_else(|_| "kafrust-position-control".to_owned());
    let topic = std::env::var("KAFRUST_TOPIC").unwrap_or_else(|_| "kafrust-smoke".to_owned());

    let mut consumer = common::apply_security(
        ConsumerConfig::new(bootstrap_servers.clone()).client_id("kafrust-position-control-direct"),
    )?
    .build()
    .await?;
    let direct_watermarks = consumer.fetch_watermarks(&topic, 0).await?;
    if direct_watermarks.high() < direct_watermarks.low() {
        return Err(Error::Unsupported(
            "direct consumer returned invalid watermarks",
        ));
    }
    consumer.assign(&topic, 0, 0);
    consumer.pause(&topic, 0)?;
    if !consumer.poll().await?.is_empty() {
        return Err(Error::Unsupported(
            "paused direct consumer returned records",
        ));
    }
    consumer.seek(&topic, 0, 0)?;
    consumer.resume(&topic, 0)?;
    let direct_records = consumer.poll().await?;
    if direct_records.is_empty()
        || !consumer
            .position(&topic, 0)
            .is_some_and(|offset| offset > 0)
    {
        return Err(Error::Unsupported(
            "resumed direct consumer did not advance position",
        ));
    }

    let mut group = common::apply_security(
        ConsumerGroupConfig::new(bootstrap_servers, group_id)
            .client_id("kafrust-position-control-group"),
    )?
    .offset_reset_policy(OffsetResetPolicy::Earliest)
    .subscribe(&topic)
    .join()
    .await?;
    let group_watermarks = group.fetch_watermarks(&topic, 0).await?;
    if group_watermarks.high() < group_watermarks.low() {
        return Err(Error::Unsupported(
            "consumer group returned invalid watermarks",
        ));
    }
    let assigned = group
        .assignments()
        .iter()
        .map(|assignment| (assignment.topic().to_owned(), assignment.partition()))
        .collect::<Vec<_>>();
    for (assigned_topic, partition) in &assigned {
        group.pause(assigned_topic, *partition)?;
    }
    if !group.poll().await?.is_empty() {
        return Err(Error::Unsupported("paused consumer group returned records"));
    }
    let (assigned_topic, partition) = assigned
        .first()
        .ok_or(Error::Unsupported("consumer group has no assignment"))?;
    group.seek(assigned_topic, *partition, 0)?;
    for (assigned_topic, partition) in &assigned {
        group.resume(assigned_topic, *partition)?;
    }
    let group_records = group.poll().await?;
    if group_records.is_empty()
        || !group
            .position(assigned_topic, *partition)
            .is_some_and(|offset| offset > 0)
    {
        return Err(Error::Unsupported(
            "resumed consumer group did not advance position",
        ));
    }
    group.leave().await?;

    println!("verified direct and group consumer watermarks, seek, pause, resume, and position");
    Ok(())
}