2aef8d4
use std::time::Duration;
use config::queue::AppQueueSettings;
use rdkafka::admin::{AdminClient, AdminOptions, NewTopic, TopicReplication};
use rdkafka::client::DefaultClientContext;
use rdkafka::config::ClientConfig;
use rdkafka::error::{KafkaError, RDKafkaErrorCode};
use crate::error::QueueError;
pub async fn ensure_topics(cfg: &AppQueueSettings, topics: &[String]) -> Result<(), QueueError> {
if topics.is_empty() {
return Ok(());
}
let mut client_cfg = ClientConfig::new();
client_cfg.set("bootstrap.servers", &cfg.brokers);
client_cfg.set("client.id", &cfg.client_id);
for (k, v) in &cfg.additional_properties {
client_cfg.set(k, v);
}
let admin: AdminClient<DefaultClientContext> = client_cfg.create().map_err(QueueError::Kafka)?;
let new_topics: Vec<NewTopic<'_>> = topics
.iter()
.map(|name| NewTopic::new(name.as_str(), 1, TopicReplication::Fixed(1)))
.collect();
let opts = AdminOptions::new().operation_timeout(Some(Duration::from_secs(10)));
let results = admin
.create_topics(new_topics.iter(), &opts)
.await
.map_err(QueueError::Kafka)?;
for result in results {
if let Err((_, code)) = result {
if code != RDKafkaErrorCode::TopicAlreadyExists {
return Err(QueueError::Kafka(KafkaError::AdminOp(code)));
}
}
}
Ok(())
}