fail init if kafka cluster unreachable

capture-server/src/main.rs

@@ -35,6 +35,9 @@ async fn shutdown() {
 
 #[tokio::main]
 async fn main() {
+    // initialize tracing
+    tracing_subscriber::fmt::init();
+
     let config = Config::init_from_env().expect("Invalid configuration:");
 
     let redis_client =
@@ -63,9 +66,6 @@ async fn main() {
         )
     };
 
-    // initialize tracing
-    tracing_subscriber::fmt::init();
-
     // run our app with hyper
     // `axum::Server` is a re-export of `hyper::Server`
 

capture/src/sink.rs

@@ -1,11 +1,15 @@
 use async_trait::async_trait;
 use metrics::{absolute_counter, counter, gauge, histogram};
+use std::time::Duration;
 use tokio::task::JoinSet;
 
 use crate::api::CaptureError;
 use rdkafka::config::ClientConfig;
 use rdkafka::error::RDKafkaErrorCode;
 use rdkafka::producer::future_producer::{FutureProducer, FutureRecord};
+use rdkafka::producer::Producer;
+use rdkafka::util::Timeout;
+use tracing::info;
 
 use crate::event::ProcessedEvent;
 
@@ -102,11 +106,19 @@ pub struct KafkaSink {
 
 impl KafkaSink {
     pub fn new(topic: String, brokers: String) -> anyhow::Result<KafkaSink> {
+        info!("connecting to Kafka brokers at {}...", brokers);
         let producer: FutureProducer<KafkaContext> = ClientConfig::new()
             .set("bootstrap.servers", &brokers)
             .set("statistics.interval.ms", "10000")
             .create_with_context(KafkaContext)?;
 
+        // Ping the cluster to make sure we can reach brokers
+        _ = producer.client().fetch_metadata(
+            Some("__consumer_offsets"),
+            Timeout::After(Duration::new(10, 0)),
+        )?;
+        info!("connected to Kafka brokers");
+
         Ok(KafkaSink { producer, topic })
     }
 }