ene_kafka 0.2.0

Ene Kafka is an easy-to-use Rust client for Apache Kafka
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
120
121
122
123
124
125
126
127
128
129
130
131
132

use anyhow::anyhow;
use async_trait::async_trait;
use rdkafka::{
    admin::{
        AdminClient, AdminOptions, ConfigResource, NewTopic, OwnedResourceSpecifier,
        ResourceSpecifier,
    },
    client::DefaultClientContext,
    config::FromClientConfig,
    types::RDKafkaErrorCode,
};
use tracing::{error, info};

use super::KafkaAdminInterface;

#[async_trait]
impl KafkaAdminInterface for AdminClient<DefaultClientContext> {
    async fn verify_topic_existence(&self, topic: &str) -> anyhow::Result<bool> {
        let topic_config = ResourceSpecifier::Topic(topic);
        let admin_options =
            AdminOptions::new().request_timeout(Some(std::time::Duration::from_secs(1)));
        self.describe_configs(vec![&topic_config], &admin_options)
            .await
            .map(|res: Vec<Result<ConfigResource, RDKafkaErrorCode>>| {
                res.iter().any(|res| match res {
                    Ok(config_resource) => match &config_resource.specifier {
                        OwnedResourceSpecifier::Topic(recieved_topic) => {
                            let topic_exists =
                                (recieved_topic == topic) && !config_resource.entries.is_empty();
                            if topic_exists {
                                info!("Topic {} exists", topic);
                            };
                            topic_exists
                        }
                        something_else => {
                            error!("Expected topic, got {:?}", something_else);
                            false
                        }
                    },
                    Err(e) => {
                        error!("Error describing topic: {:?}", e);
                        false
                    }
                })
            })
            .map_err(|e| {
                error!("Error describing topic: {:?}", e);
                anyhow!(e.to_string())
            })
    }
    async fn create_topic_if_not_exists(
        &self,
        topic: &str,
        partitions: i32,
        replication_factor: i32,
    ) -> anyhow::Result<()> {
        let topic_exists = self.verify_topic_existence(topic).await?;
        if !topic_exists {
            let new_topic = NewTopic::new(
                topic,
                partitions,
                rdkafka::admin::TopicReplication::Fixed(replication_factor),
            );
            self.create_topics(
                &[new_topic],
                &AdminOptions::new()
                    .request_timeout(Some(std::time::Duration::from_secs(1)))
                    .operation_timeout(Some(std::time::Duration::from_secs(1))),
            )
            .await
            .map(|res| {
                res.iter().all(|res| match res {
                    Ok(_) => true,
                    Err(e) => {
                        error!("Error creating topic: {:?}", e);
                        false
                    }
                })
            })
            .map_err(|e| {
                error!("Error creating topic: {:?}", e);
                anyhow!(e.to_string())
            })?;
        }
        Ok(())
    }
    async fn check_topic_liveness(&self, topic: &str) -> anyhow::Result<bool> {
        let topic_config = ResourceSpecifier::Topic(topic);
        let admin_options = AdminOptions::new();
        self.describe_configs(vec![&topic_config], &admin_options)
            .await
            .map(|res: Vec<Result<ConfigResource, RDKafkaErrorCode>>| {
                res.iter().all(|res| res.is_ok())
                    && res.iter().any(|res| match res {
                        Ok(config_resource) => match &config_resource.specifier {
                            OwnedResourceSpecifier::Topic(recieved_topic) => {
                                (recieved_topic == topic) && !config_resource.entries.is_empty()
                            }
                            something_else => {
                                error!("Expected topic, got {:?}", something_else);
                                false
                            }
                        },
                        Err(e) => {
                            error!("Error describing topic: {:?}", e);
                            false
                        }
                    })
            })
            .map_err(|e| {
                error!("Error describing topic: {:?}", e);
                anyhow!(e.to_string())
            })
    }

    fn new(
        bootstrap_servers: String,
        request_time_out_ms: String,
        connection_max_idle_ms: String,
    ) -> Self {
        AdminClient::from_config(
            &rdkafka::ClientConfig::new()
                .set("bootstrap.servers", bootstrap_servers)
                .set("request.timeout.ms", &request_time_out_ms)
                .set(
                    "connections.max.idle.ms",
                    &connection_max_idle_ms.to_string(),
                ),
        )
        .expect("Failed to create admin client")
    }
}