feat(kafka): Use the reloading producer for every topic

CHANGELOG.md

@@ -16,7 +16,7 @@
 **Internal**:
 
 - Make `spool.envelopes.max_unspool_envelopes_per_second` hot-reloadable. ([#6452](https://github.com/getsentry/relay/pull/6452))
-- Add `processing.reloading_producer_topics`, which lets a topic replace its Kafka client when `processing.kafka_config` changes instead of requiring a restart. Empty by default. ([#6529](https://github.com/getsentry/relay/pull/6529))
+- Kafka producers replace their client when `processing.kafka_config` or `processing.secondary_kafka_configs` changes, instead of requiring a restart. ([#6529](https://github.com/getsentry/relay/pull/6529), [#6549](https://github.com/getsentry/relay/pull/6549))
 
 ## 26.10.0
 

relay-config/src/config.rs

@@ -1,4 +1,5 @@
-use std::collections::{BTreeMap, BTreeSet, HashMap};
+use std::collections::btree_map::Entry;
+use std::collections::{BTreeMap, HashMap};
 use std::error::Error;
 use std::io::Write;
 use std::net::{IpAddr, SocketAddr};
@@ -1178,20 +1179,8 @@ pub struct Processing {
     pub topics: TopicAssignments,
     /// Whether to validate the supplied topics by calling Kafka's metadata endpoints.
     pub kafka_validate_topics: bool,
-    /// Logical topics whose Kafka client is replaced when the config changes, instead of being
-    /// fixed for the lifetime of the process.
-    ///
-    /// Named by logical topic, for example `ingest-events`. Note that `topics` additionally
-    /// accepts a short form (`events`), which this option does not: only the logical name
-    /// selects a topic here. Anything else is ignored.
-    ///
-    /// Empty by default, which is the behaviour of every topic before this option existed.
-    ///
-    /// This exists to roll the reloading producer out one topic at a time and is meant to be
-    /// removed once every topic uses it.
-    pub reloading_producer_topics: BTreeSet<String>,
-    /// Upper bound, in seconds, on the random delay before a reloading producer swaps its
-    /// client.
+    /// Upper bound, in seconds, on the random delay before a Kafka producer swaps its client
+    /// after the Kafka config changed.
     ///
     /// Spreads swaps out so that a config change does not have every pod in the fleet reconnect
     /// at once. Keep it well below how often the Kafka config is expected to change. Tests set
@@ -1240,7 +1229,6 @@ impl Default for Processing {
             secondary_kafka_configs: BTreeMap::new(),
             topics: TopicAssignments::default(),
             kafka_validate_topics: false,
-            reloading_producer_topics: BTreeSet::new(),
             reloading_producer_jitter_secs: 60,
             redis: None,
             attachment_chunk_size: ByteSize::mebibytes(1),
@@ -1921,8 +1909,7 @@ impl ConfigInner {
         // live, so a renamed topic would be keyed by its new name while the producer still
         // selects the old one.
         //
-        // This only reaches a running Kafka client for topics using the reloading producer;
-        // see `Processing::reloading_producer_topics`. Everything else picks it up on restart.
+        // Running Kafka clients pick this up via `ConfigReloadService`.
         if self.values.processing.kafka_config != other.values.processing.kafka_config
             || self.values.processing.secondary_kafka_configs
                 != other.values.processing.secondary_kafka_configs
@@ -2865,30 +2852,11 @@ impl ConfigSnapshot {
         self.inner.values.processing.kafka_validate_topics
     }
 
-    /// Whether this topic's Kafka client is replaced when the config changes.
-    pub fn is_reloading_producer_topic(&self, topic: KafkaTopic) -> bool {
-        self.inner
-            .values
-            .processing
-            .reloading_producer_topics
-            .contains(topic.logical_topic_name())
-    }
-
-    /// Upper bound on the random delay before a reloading producer swaps its client.
+    /// Upper bound on the random delay before a Kafka producer swaps its client.
     pub fn reloading_producer_jitter(&self) -> Duration {
         Duration::from_secs(self.inner.values.processing.reloading_producer_jitter_secs)
     }
 
-    /// Whether any topic at all uses the reloading producer.
-    pub fn has_reloading_producer_topics(&self) -> bool {
-        !self
-            .inner
-            .values
-            .processing
-            .reloading_producer_topics
-            .is_empty()
-    }
-
     /// Builds the Kafka config blob describing every configured topic.
     ///
     /// # This is a stand-in
@@ -2928,10 +2896,18 @@ impl ConfigSnapshot {
                 // Arroyo keys the blob by logical topic and maps it to one physical topic. Relay's
                 // sharded topics are one logical topic to many physical ones, which Arroyo cannot
                 // express yet, so each physical topic is its own entry and maps to itself.
-                topics.insert(
-                    params.topic_name.clone(),
-                    serde_json::json!({ "physical_topic": { "cluster": cluster_name } }),
-                );
+                //
+                // That also means one topic name cannot point at two clusters.
+                let entry = serde_json::json!({ "physical_topic": { "cluster": cluster_name } });
+                match topics.entry(params.topic_name.clone()) {
+                    Entry::Vacant(vacant) => {
+                        vacant.insert(entry);
+                    }
+                    Entry::Occupied(occupied) if *occupied.get() != entry => {
+                        return Err(KafkaConfigError::AmbiguousTopic(params.topic_name.clone()));
+                    }
+                    Entry::Occupied(_) => {}
+                }
             }
         }
 
@@ -3304,6 +3280,34 @@ processing:
         assert_eq!(after, "other:9092");
     }
 
+    /// The blob is keyed by topic name, so one name on two clusters cannot be expressed. That
+    /// must fail loudly instead of sending one of the two to the wrong cluster.
+    #[cfg(feature = "processing")]
+    #[test]
+    fn test_kafka_config_blob_rejects_topic_on_two_clusters() {
+        let config = processing_config(
+            r#"
+processing:
+  enabled: true
+  kafka_config:
+    - {name: "bootstrap.servers", value: "primary:9092"}
+  secondary_kafka_configs:
+    other:
+      - {name: "bootstrap.servers", value: "other:9092"}
+  topics:
+    outcomes:
+      name: "outcomes-topic"
+      config: "other"
+    outcomes-billing: "outcomes-topic"
+"#,
+        );
+
+        assert!(matches!(
+            config.current().kafka_config_blob(),
+            Err(KafkaConfigError::AmbiguousTopic(topic)) if topic == "outcomes-topic"
+        ));
+    }
+
     /// Broker settings reload; topic assignments do not, since producers hold theirs from
     /// startup.
     #[test]

relay-kafka/src/config.rs

@@ -14,6 +14,9 @@ pub enum ConfigError {
     /// The user did not configure 0 shard
     #[error("invalid kafka shard configuration: must have shard with index 0")]
     InvalidShard,
+    /// Two topic assignments produce to the same Kafka topic name on different clusters.
+    #[error("kafka topic {0:?} is configured on more than one kafka config")]
+    AmbiguousTopic(String),
 }
 
 /// Define the topics over which Relay communicates with Sentry.