feat(queue): ensure kafka topics exist on email worker startup
Create the gta.email_events_* topics via the Kafka AdminClient before subscribers start, tolerating TopicAlreadyExists, so consumers no longer loop on UnknownTopicOrPartition.
zhenyi committed
Modified lib/email/runtime.rs lib/email/runtime.rs +15 −0 View lib/email/runtime.rs
@@ -8,6 +8,9 @@
88use tracing::{info, warn};
99
1010use crate::error::EmailError;
11use crate::events::{
12 EmailFailed, EmailSent, RenderAndSendEmailRequest, SendBatchEmailRequest, SendEmailRequest,
13};
1114use crate::service::{EmailService, EmailServiceBuilder, EmailSubscribers};
1215
1316#[derive(Debug, Default, Clone)]
@@ -39,6 +42,8 @@
3942 "email service disabled at startup".to_string(),
4043 ))?;
4144
45 ensure_email_topics(&queue).await?;
46
4247 let subscribers = service.start_subscribers()?;
4348 info!("email worker subscribers started");
4449
@@ -49,6 +54,16 @@
4954 })
5055}
5156
57async fn ensure_email_topics(queue: &Queue) -> Result<(), EmailError> {
58 queue.ensure_typed_topic::<SendEmailRequest>().await?;
59 queue.ensure_typed_topic::<SendBatchEmailRequest>().await?;
60 queue.ensure_typed_topic::<RenderAndSendEmailRequest>().await?;
61 queue.ensure_typed_topic::<EmailSent>().await?;
62 queue.ensure_typed_topic::<EmailFailed>().await?;
63 info!("email topics ensured");
64 Ok(())
65}
66
5267pub async fn wait_for_shutdown() {
5368 let ctrl_c = async {
5469 if let Err(e) = signal::ctrl_c().await {
Added lib/queue/admin.rs lib/queue/admin.rs +43 −0 View lib/queue/admin.rs
@@ -0,0 +1,43 @@
1use std::time::Duration;
2
3use config::queue::AppQueueSettings;
4use rdkafka::admin::{AdminClient, AdminOptions, NewTopic, TopicReplication};
5use rdkafka::client::DefaultClientContext;
6use rdkafka::config::ClientConfig;
7use rdkafka::error::{KafkaError, RDKafkaErrorCode};
8
9use crate::error::QueueError;
10
11pub async fn ensure_topics(cfg: &AppQueueSettings, topics: &[String]) -> Result<(), QueueError> {
12 if topics.is_empty() {
13 return Ok(());
14 }
15
16 let mut client_cfg = ClientConfig::new();
17 client_cfg.set("bootstrap.servers", &cfg.brokers);
18 client_cfg.set("client.id", &cfg.client_id);
19 for (k, v) in &cfg.additional_properties {
20 client_cfg.set(k, v);
21 }
22 let admin: AdminClient<DefaultClientContext> = client_cfg.create().map_err(QueueError::Kafka)?;
23
24 let new_topics: Vec<NewTopic<'_>> = topics
25 .iter()
26 .map(|name| NewTopic::new(name.as_str(), 1, TopicReplication::Fixed(1)))
27 .collect();
28
29 let opts = AdminOptions::new().operation_timeout(Some(Duration::from_secs(10)));
30 let results = admin
31 .create_topics(new_topics.iter(), &opts)
32 .await
33 .map_err(QueueError::Kafka)?;
34
35 for result in results {
36 if let Err((_, code)) = result {
37 if code != RDKafkaErrorCode::TopicAlreadyExists {
38 return Err(QueueError::Kafka(KafkaError::AdminOp(code)));
39 }
40 }
41 }
42 Ok(())
43}
Modified lib/queue/lib.rs lib/queue/lib.rs +13 −0 View lib/queue/lib.rs
@@ -1,3 +1,4 @@
1pub mod admin;
12pub mod consumer;
23pub mod error;
34pub mod metrics;
@@ -78,6 +79,18 @@
7879 TypedQueue::new(self)
7980 }
8081
82 pub async fn ensure_topics(&self, topics: &[String]) -> Result<(), QueueError> {
83 admin::ensure_topics(&self.inner.cfg, topics).await
84 }
85
86 pub async fn ensure_typed_topic<T>(&self) -> Result<(), QueueError>
87 where
88 T: 'static,
89 {
90 let topic = topic::topic_for_type::<T>(&self.topic_prefix());
91 self.ensure_topics(&[topic]).await
92 }
93
8194 pub async fn shutdown(&self) -> Result<(), QueueError> {
8295 Ok(())
8396 }