fix: wire up kafka topic refresh loop and make auto-discovery opt-in

- Connect the existing TopicRefreshHandle to KafkaTransport so periodic
  topic re-resolution actually runs (was implemented but never wired up)
- Add auto_discover field to KafkaConfig (default false) so producers
  with empty topics don't trigger unwanted broker auto-discovery
- Default topic_exclude to ["^__"] to filter Kafka internal topics
  like __consumer_offsets from auto-discovery results
- Demote worker pool scaling log from info to debug to reduce log spam

Author: kazmosahebi

src/transport/kafka/config.rs

@@ -349,14 +349,22 @@ pub struct KafkaConfig {
     #[serde(default)]
     pub topics: Vec<String>,
 
+    /// Enable auto-discovery when `topics` is empty.
+    /// When false (default), empty `topics` means no subscription (producer-only).
+    /// When true, empty `topics` triggers broker auto-discovery with
+    /// `topic_include`/`topic_exclude` filters and suppression rules.
+    #[serde(default)]
+    pub auto_discover: bool,
+
     /// Regex patterns for topic include filtering (empty = accept all).
     /// Topics must match at least one pattern (OR logic).
     #[serde(default)]
     pub topic_include: Vec<String>,
 
     /// Regex patterns for topic exclude filtering.
     /// Topics matching any pattern are excluded. Exclude wins over include.
-    #[serde(default)]
+    /// Default: `["^__"]` (excludes Kafka internal topics like `__consumer_offsets`).
+    #[serde(default = "default_topic_exclude")]
     pub topic_exclude: Vec<String>,
 
     /// Periodic topic refresh interval in seconds (0 = disabled).
@@ -466,6 +474,10 @@ pub struct KafkaConfig {
     pub extra_config: HashMap<String, String>,
 }
 
+fn default_topic_exclude() -> Vec<String> {
+    vec!["^__".to_string()]
+}
+
 fn default_topic_refresh_secs() -> u64 {
     60
 }
@@ -534,8 +546,9 @@ impl Default for KafkaConfig {
             group: default_group(),
             client_id: default_client_id(),
             topics: Vec::new(),
+            auto_discover: false,
             topic_include: Vec::new(),
-            topic_exclude: Vec::new(),
+            topic_exclude: default_topic_exclude(),
             topic_refresh_secs: default_topic_refresh_secs(),
             topic_suppression_rules: default_topic_suppression_rules(),
             security_protocol: default_security_protocol(),
@@ -935,7 +948,8 @@ mod tests {
     fn kafka_config_topic_resolution_defaults() {
         let config = KafkaConfig::default();
         assert!(config.topic_include.is_empty());
-        assert!(config.topic_exclude.is_empty());
+        assert_eq!(config.topic_exclude, vec!["^__".to_string()]);
+        assert!(!config.auto_discover);
         assert_eq!(config.topic_refresh_secs, 60);
         assert_eq!(config.topic_suppression_rules.len(), 1);
         assert_eq!(config.topic_suppression_rules[0].preferred_suffix, "_load");

src/transport/kafka/mod.rs

@@ -122,6 +122,11 @@ pub struct KafkaTransport {
     healthy: Arc<AtomicBool>,
     /// Topics we're subscribed to (for cache warming).
     subscribed_topics: Vec<String>,
+    /// Shutdown token — cancelled on close() to stop background tasks.
+    shutdown_token: tokio_util::sync::CancellationToken,
+    /// Periodic topic refresh handle (auto-discovery mode only).
+    /// Checked on each recv() call to detect new/removed topics.
+    topic_refresh: Option<std::sync::Mutex<TopicRefreshHandle>>,
 }
 
 impl KafkaTransport {
@@ -217,20 +222,45 @@ impl KafkaTransport {
             .create_with_context(StatsContext::new())
             .map_err(|e| TransportError::Connection(format!("Failed to create consumer: {e}")))?;
 
-        // Resolve effective topics: use explicit list or auto-discover from broker
-        let effective_topics = if config.topics.is_empty() {
-            tracing::info!("Topics empty — auto-discovering from broker");
-            let resolver = topic_resolver::TopicResolver::new(config)?;
-            let discovered = resolver.resolve()?;
-            if discovered.is_empty() {
-                return Err(TransportError::Config(
-                    "Auto-discovery found no matching topics".into(),
-                ));
-            }
-            discovered
-        } else {
-            config.topics.clone()
-        };
+        // Resolve effective topics:
+        // - Explicit list → subscribe to those
+        // - Empty + auto_discover → auto-discover from broker
+        // - Empty + !auto_discover → no subscription (producer-only)
+        let (effective_topics, topic_refresh, shutdown_token) =
+            if config.topics.is_empty() && config.auto_discover {
+                tracing::info!("Topics empty — auto-discovering from broker");
+                let resolver = topic_resolver::TopicResolver::new(config)?;
+                let discovered = resolver.resolve()?;
+                if discovered.is_empty() {
+                    return Err(TransportError::Config(
+                        "Auto-discovery found no matching topics".into(),
+                    ));
+                }
+
+                let token = tokio_util::sync::CancellationToken::new();
+                let refresh = if config.topic_refresh_secs > 0 {
+                    let refresh_resolver = topic_resolver::TopicResolver::new(config)?;
+                    let handle = refresh_resolver.start_refresh_loop(
+                        Duration::from_secs(config.topic_refresh_secs),
+                        token.clone(),
+                    );
+                    tracing::info!(
+                        interval_secs = config.topic_refresh_secs,
+                        "Started periodic topic refresh"
+                    );
+                    Some(std::sync::Mutex::new(handle))
+                } else {
+                    None
+                };
+
+                (discovered, refresh, token)
+            } else {
+                (
+                    config.topics.clone(),
+                    None,
+                    tokio_util::sync::CancellationToken::new(),
+                )
+            };
 
         // Subscribe to topics
         let subscribed_topics = effective_topics;
@@ -273,6 +303,8 @@ impl KafkaTransport {
             closed: AtomicBool::new(false),
             healthy,
             subscribed_topics,
+            shutdown_token,
+            topic_refresh,
         })
     }
 
@@ -290,6 +322,7 @@ impl TransportBase for KafkaTransport {
     async fn close(&self) -> TransportResult<()> {
         self.closed.store(true, Ordering::Relaxed);
         self.healthy.store(false, Ordering::Relaxed);
+        self.shutdown_token.cancel();
         // rdkafka handles cleanup on drop
         Ok(())
     }
@@ -386,6 +419,23 @@ impl TransportReceiver for KafkaTransport {
             return Err(TransportError::Closed);
         }
 
+        // Check for topic changes from the background refresh loop
+        if let Some(ref refresh) = self.topic_refresh {
+            if let Ok(mut handle) = refresh.lock() {
+                if let Some(new_topics) = handle.check_changed() {
+                    let topics: Vec<&str> = new_topics.iter().map(String::as_str).collect();
+                    match self.consumer.subscribe(&topics) {
+                        Ok(()) => {
+                            tracing::info!(?new_topics, "Re-subscribed after topic refresh");
+                        }
+                        Err(e) => {
+                            tracing::warn!(error = %e, "Failed to re-subscribe after topic refresh");
+                        }
+                    }
+                }
+            }
+        }
+
         let timeout = Duration::from_millis(tuning::POLL_TIMEOUT_MS);
         let max_msgs = max;
 
@@ -542,6 +592,7 @@ impl std::fmt::Debug for KafkaTransport {
             .field("subscribed_topics", &self.subscribed_topics)
             .field("closed", &self.closed.load(Ordering::Relaxed))
             .field("healthy", &self.healthy.load(Ordering::Relaxed))
+            .field("topic_refresh_active", &self.topic_refresh.is_some())
             .finish_non_exhaustive()
     }
 }

src/worker/scaler.rs

@@ -157,7 +157,7 @@ impl ScalingController {
                 "Worker pool steady"
             );
         } else {
-            tracing::info!(
+            tracing::debug!(
                 cpu = format!("{cpu_util:.2}"),
                 mem = format!("{effective_memory_pressure:.2}"),
                 current = current_permits,