kafrust 0.2.29

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

mod common;

use kafrust::{
    AclBinding, AclFilter, AclOperation, AclPatternType, AclPermissionType, AclResourceType,
    AdminClient, ClientConfig, Error,
};
use std::time::{Duration, Instant};

#[tokio::main]
async fn main() -> kafrust::Result<()> {
    let bootstrap_servers = common::bootstrap_servers_from_env();
    let topic =
        std::env::var("KAFRUST_ACL_TOPIC").unwrap_or_else(|_| "kafrust-acl-smoke".to_owned());
    let principal =
        std::env::var("KAFRUST_ACL_PRINCIPAL").unwrap_or_else(|_| "User:ANONYMOUS".to_owned());
    let config = common::apply_security(
        ClientConfig::new(bootstrap_servers).client_id("kafrust-acl-example"),
    )?;
    let admin = AdminClient::new(config);
    let binding = AclBinding::new(
        AclResourceType::Topic,
        &topic,
        AclPatternType::Literal,
        &principal,
        "*",
        AclOperation::Read,
        AclPermissionType::Allow,
    );

    let created = admin.create_acls(std::slice::from_ref(&binding)).await?;
    if !created.is_success() {
        return Err(Error::Broker {
            code: created
                .results()
                .first()
                .map(|result| result.error_code())
                .unwrap_or(-1),
            context: format!("create ACL for topic {topic}"),
        });
    }
    println!("created ACL for {principal} on topic {topic}");

    let filter = AclFilter::any()
        .resource_type(AclResourceType::Topic)
        .resource_name(&topic)
        .pattern_type(AclPatternType::Literal)
        .principal(&principal)
        .host("*")
        .operation(AclOperation::Read)
        .permission_type(AclPermissionType::Allow);
    wait_for_acl_binding(&admin, &filter, &binding, &topic).await?;
    println!("described ACL for {principal} on topic {topic}");

    let deleted = admin.delete_acls(&[filter]).await?;
    if !deleted.is_success() {
        return Err(Error::Broker {
            code: deleted
                .filter_results()
                .first()
                .map(|result| result.error_code())
                .unwrap_or(-1),
            context: format!("delete ACL for topic {topic}"),
        });
    }
    println!("deleted ACL for {principal} on topic {topic}");

    Ok(())
}

async fn wait_for_acl_binding(
    admin: &AdminClient,
    filter: &AclFilter,
    binding: &AclBinding,
    topic: &str,
) -> kafrust::Result<()> {
    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        let described = admin.describe_acls(filter).await?;
        if !described.is_success() {
            return Err(Error::Broker {
                code: described.error_code(),
                context: format!("describe ACL for topic {topic}"),
            });
        }
        if described
            .bindings()
            .iter()
            .any(|candidate| candidate == binding)
        {
            return Ok(());
        }
        if Instant::now() >= deadline {
            return Err(Error::Broker {
                code: 0,
                context: format!("ACL for topic {topic} was not visible before timeout"),
            });
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
}