From 67823b73a13ce3b0b67121ddf4e537bb279339fe Mon Sep 17 00:00:00 2001 From: Markus Unterwaditzer Date: Thu, 22 Oct 2026 16:09:42 +0200 Subject: [PATCH 1/3] feat(kafka): Use the reloading producer for every topic Removes the static Kafka client and the processing.reloading_producer_topics rollout option. Every physical topic now gets its own ReloadingKafkaProducer, which is replaced when the Kafka config changes. ref STREAM-1939 --- relay-config/src/config.rs | 43 +-- relay-kafka/src/producer/mod.rs | 314 +++++++-------------- relay-kafka/src/producer/utils.rs | 2 +- relay-kafka/src/statsd.rs | 6 +- relay-server/src/services/config_reload.rs | 3 +- relay-server/src/services/store.rs | 24 +- tests/integration/test_config.py | 8 +- tests/integration/test_store.py | 5 +- 8 files changed, 122 insertions(+), 283 deletions(-) diff --git a/relay-config/src/config.rs b/relay-config/src/config.rs index 6f44ec579b3..5eba7785b15 100644 --- a/relay-config/src/config.rs +++ b/relay-config/src/config.rs @@ -1,4 +1,4 @@ -use std::collections::{BTreeMap, BTreeSet, HashMap}; +use std::collections::{BTreeMap, HashMap}; use std::error::Error; use std::io::Write; use std::net::{IpAddr, SocketAddr}; @@ -1178,20 +1178,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, - /// 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 +1228,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 +1908,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 +2851,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 diff --git a/relay-kafka/src/producer/mod.rs b/relay-kafka/src/producer/mod.rs index 9e56a1ede80..c1d1dc0df76 100644 --- a/relay-kafka/src/producer/mod.rs +++ b/relay-kafka/src/producer/mod.rs @@ -7,13 +7,11 @@ use std::sync::Arc; use std::sync::atomic::{AtomicU32, Ordering}; use std::time::{Duration, Instant}; -use rdkafka::message::{Header, ToBytes}; +use rdkafka::message::Header; use rdkafka::producer::BaseRecord; use relay_statsd::metric; use sentry_arroyo::backends::ProducerError; -use sentry_arroyo::backends::kafka::config::KafkaConfig; use sentry_arroyo::backends::kafka::config_blob::ProducerSelector; -use sentry_arroyo::backends::kafka::producer::KafkaProducer as ArroyoKafkaProducer; use sentry_arroyo::backends::kafka::reloading_producer::{ ReloadConfig, ReloadError, ReloadingKafkaProducer, }; @@ -50,10 +48,6 @@ pub enum ClientError { #[error("failed to find producer for the requested kafka topic")] InvalidTopicName, - /// Failed to create a kafka producer because of the invalid configuration. - #[error("failed to create kafka producer: invalid kafka config: {0}")] - InvalidConfig(#[source] rdkafka::error::KafkaError), - /// Failed to serialize the message. #[error("failed to serialize kafka message")] InvalidMsgPack(#[source] rmp_serde::encode::Error), @@ -75,8 +69,9 @@ pub enum ClientError { #[error("failed to validate the topic with name {0}: {1}")] TopicError(String, #[source] rdkafka::error::KafkaError), - /// Failed to build a producer that reloads its Kafka client. - #[error("failed to create reloading kafka producer")] + /// Failed to create a kafka producer, for example because the config blob does not resolve + /// for the topic or the resolved config is invalid. + #[error("failed to create kafka producer")] Reload(#[from] ReloadError), /// Failed to encode the protobuf into the buffer @@ -167,14 +162,14 @@ impl TopicProducers { struct TopicProducer { pub topic_name: String, - pub producer: Arc, + pub producer: Arc>, pub rate_limiter: Option, } -/// Asks for a topic to be produced through a client that can be reloaded. +/// Where producers take their initial Kafka config from, and how they reload it. #[derive(Debug, Clone, Copy)] -pub struct ReloadingTopic<'a> { - /// The Kafka config blob the initial client is resolved from. +pub struct KafkaConfigSource<'a> { + /// The Kafka config blob the initial clients are resolved from. pub blob: &'a [u8], /// Which application's overrides in the blob apply. For Relay, `"relay"`. pub app: &'a str, @@ -197,74 +192,6 @@ fn reload_config(jitter: Duration) -> ReloadConfig { } } -/// The Kafka client behind a topic. -/// -/// Selects a fixed or a reloading client per topic. Both take the same -/// [`BaseRecord`] and report through the same [`Context`]. Delete this, and the -/// config option that selects it, once every topic is on [`Backend::Reloading`]. -enum Backend { - /// A client fixed for the lifetime of the process. - Static(ArroyoKafkaProducer), - /// A client that is replaced when a new Kafka config is pushed in. - Reloading(ReloadingKafkaProducer), -} - -impl Backend { - fn context(&self) -> &Context { - match self { - Self::Static(producer) => producer.context(), - Self::Reloading(producer) => producer.context(), - } - } - - fn in_flight_count(&self) -> i32 { - match self { - Self::Static(producer) => producer.in_flight_count(), - Self::Reloading(producer) => producer.in_flight_count(), - } - } - - /// Number of times the client was replaced. Always zero for a static one. - fn generation(&self) -> u64 { - match self { - Self::Static(_) => 0, - Self::Reloading(producer) => producer.generation(), - } - } - - fn produce_record(&self, record: BaseRecord<'_, K, P>) -> Result<(), ProducerError> - where - K: ToBytes + ?Sized, - P: ToBytes + ?Sized, - { - match self { - Self::Static(producer) => producer.produce_record(record), - Self::Reloading(producer) => producer.produce_record(record), - } - } - - fn validate_topic( - &self, - topic: Topic, - timeout: Duration, - ) -> Result<(), rdkafka::error::KafkaError> { - match self { - Self::Static(producer) => producer.validate_topic(topic, timeout), - Self::Reloading(producer) => producer.validate_topic(topic, timeout), - } - } - - /// Hands a new Kafka config to a reloading client. A static one ignores it. - /// - /// Returns `Ok(false)` when there is nothing to push to. - fn push_config(&self, blob: &[u8]) -> Result { - match self { - Self::Static(_) => Ok(false), - Self::Reloading(producer) => producer.push_config(blob).map(|_| true), - } - } -} - /// Single kafka producer config with assigned topic. struct Producer { /// Topic to producer and rate limiter mappings for sharding. @@ -477,11 +404,10 @@ impl KafkaClient { producer.send(key, headers, variant, payload) } - /// Hands a new Kafka config blob to every topic on the reloading producer. + /// Hands a new Kafka config blob to every topic's producer. /// - /// Topics still on a static client ignore it. A blob that does not resolve - /// for a topic is logged and counted rather than returned; that topic keeps - /// the client it is running. + /// A blob that does not resolve for a topic is logged and counted rather + /// than returned; that topic keeps the client it is running. /// /// A successful push only means the rollout started. Arroyo probes a /// broker before swapping; watch `kafka.reload_generation` for the swap. @@ -491,13 +417,11 @@ impl KafkaClient { let producer_name = tp.producer.context().producer_name(); match tp.producer.push_config(blob) { - Ok(true) => metric!( + Ok(_) => metric!( counter(KafkaCounters::ConfigReloadPushed) += 1, topic = &tp.topic_name, producer_name = producer_name ), - // Not a reloading producer, nothing to push to. - Ok(false) => {} Err(error) => { relay_log::error!( error = &error as &dyn std::error::Error, @@ -516,26 +440,14 @@ impl KafkaClient { } } -/// Identifies a producer that can be shared between topics. -/// -/// A static client is defined purely by its broker config, so everything naming the same one -/// shares it. A reloading client additionally resolves its config out of the blob by physical -/// topic name, so it can only be shared by topics producing to that same name. Sharing it any -/// wider would point some of them at the wrong topic's config. -/// -/// This also keeps a reloading and a static topic that name the same Kafka config apart, which -/// would otherwise collide and leave one of them on the wrong backend. -#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] -struct ProducerKey { - config_name: Option, - /// The physical topic, for reloading producers only. - topic_name: Option, -} - /// Helper structure responsible for building the actual [`KafkaClient`]. #[derive(Default)] pub struct KafkaClientBuilder { - reused_producers: BTreeMap>, + /// Producers by physical topic name. + /// + /// A producer resolves its config out of the blob by physical topic, so it can only be + /// shared by logical topics that produce to that same physical topic. + reused_producers: BTreeMap>>, producers: HashMap, } @@ -548,21 +460,18 @@ impl KafkaClientBuilder { /// Adds topic configuration to the current [`KafkaClientBuilder`], which in return assigns /// dedicates producer to the topic which can will be used to send the messages. /// - /// When `reloading` is set, the topic's client is resolved from the config blob it carries - /// and is replaced whenever a newer blob is pushed in through - /// [`KafkaClient::push_kafka_config`]. Otherwise the client is fixed for the lifetime of - /// the process. + /// Each physical topic's client is resolved from the config blob in `source`, and is + /// replaced whenever a newer blob is pushed in through [`KafkaClient::push_kafka_config`]. /// /// # Errors - /// Returns [`ClientError::InvalidConfig`] error if the provided configuration is wrong and - /// the producer could not be created, or [`ClientError::Reload`] if a reloading producer - /// could not resolve its config out of the blob. + /// Returns [`ClientError::Reload`] if a producer could not be created, for example because + /// the blob has no entry for one of the topic's physical topics. pub fn add_kafka_topic_config( mut self, topic: KafkaTopic, topic_config: &KafkaTopicConfig<'_>, validate_topic: bool, - reloading: Option<&ReloadingTopic<'_>>, + source: &KafkaConfigSource<'_>, ) -> Result { let mut topic_producers = TopicProducers::new(); @@ -584,14 +493,8 @@ impl KafkaClientBuilder { ) }); - let config_name = config_name.map(str::to_owned); - let key = ProducerKey { - config_name: config_name.clone(), - topic_name: reloading.is_some().then(|| topic_name.clone()), - }; - - // Get or create producer for this broker config - let kafka_producer = if let Some(producer) = self.reused_producers.get(&key) { + // Get or create the producer for this physical topic + let kafka_producer = if let Some(producer) = self.reused_producers.get(topic_name) { Arc::clone(producer) } else { // Extract producer name from client.id, fallback to config name, then "unknown" @@ -599,35 +502,18 @@ impl KafkaClientBuilder { .iter() .find(|p| p.name == "client.id") .map(|p| p.value.clone()) - .or_else(|| config_name.clone()) + .or_else(|| config_name.map(str::to_owned)) .unwrap_or_else(|| "unknown".to_owned()); - let context = Context::new(producer_name); - let producer = match reloading { - Some(reloading) => { - let selector = ProducerSelector::new(topic_name.clone(), reloading.app); - Backend::Reloading(ReloadingKafkaProducer::new_with_context( - reloading.blob, - selector, - reload_config(reloading.jitter), - context, - )?) - } - None => { - let params = config_params - .iter() - .map(|param| (param.name.clone(), param.value.clone())) - .collect(); - let config = KafkaConfig::new_config(Vec::new(), Some(params)); - Backend::Static( - ArroyoKafkaProducer::new_with_context(config, context) - .map_err(ClientError::InvalidConfig)?, - ) - } - }; - let producer = Arc::new(producer); + let producer = Arc::new(ReloadingKafkaProducer::new_with_context( + source.blob, + ProducerSelector::new(topic_name.clone(), source.app), + reload_config(source.jitter), + Context::new(producer_name), + )?); - self.reused_producers.insert(key, Arc::clone(&producer)); + self.reused_producers + .insert(topic_name.clone(), Arc::clone(&producer)); producer }; @@ -694,6 +580,14 @@ mod tests { .unwrap() } + fn source(blob: &[u8]) -> KafkaConfigSource<'_> { + KafkaConfigSource { + blob, + app: "relay", + jitter: Duration::ZERO, + } + } + fn params(servers: &str) -> Vec { vec![KafkaConfigParam { name: "bootstrap.servers".to_owned(), @@ -701,22 +595,11 @@ mod tests { }] } - fn build( - assignment: &TopicAssignment, - params: &Vec, - reloading: Option<&ReloadingTopic<'_>>, - ) -> KafkaClient { - let secondary = BTreeMap::new(); - let configs = assignment.kafka_configs(params, &secondary).unwrap(); - - KafkaClient::builder() - .add_kafka_topic_config(KafkaTopic::Events, &configs, false, reloading) - .unwrap() - .build() - } - - fn backends(client: &KafkaClient) -> Vec<&Arc> { - client.producers[&KafkaTopic::Events] + fn producers( + client: &KafkaClient, + topic: KafkaTopic, + ) -> Vec<&Arc>> { + client.producers[&topic] .topic_producers .producers .iter() @@ -724,86 +607,89 @@ mod tests { .collect() } - /// Every shard needs its own reloading producer, since each is a separate - /// physical topic with its own entry in the blob. + /// Every shard needs its own producer, since each is a separate physical + /// topic with its own entry in the blob. #[test] - fn test_sharded_reloading_topic_gets_one_producer_per_shard() { + fn test_sharded_topic_gets_one_producer_per_shard() { let cluster = MockCluster::new(1).unwrap(); let servers = cluster.bootstrap_servers(); let assignment: TopicAssignment = serde_yaml::from_str(r#"["ingest-events-1", "ingest-events-2"]"#).unwrap(); let blob = blob(&servers, &["ingest-events-1", "ingest-events-2"]); + let params = params(&servers); + let secondary = BTreeMap::new(); + let configs = assignment.kafka_configs(¶ms, &secondary).unwrap(); - let client = build( - &assignment, - ¶ms(&servers), - Some(&ReloadingTopic { - blob: &blob, - app: "relay", - jitter: Duration::ZERO, - }), - ); + let client = KafkaClient::builder() + .add_kafka_topic_config(KafkaTopic::Events, &configs, false, &source(&blob)) + .unwrap() + .build(); - let backends = backends(&client); - assert_eq!(backends.len(), 2); - assert!( - backends - .iter() - .all(|backend| matches!(***backend, Backend::Reloading(_))) - ); + let producers = producers(&client, KafkaTopic::Events); + assert_eq!(producers.len(), 2); // Sharding routes by index, so two shards sharing one producer would // send both to whichever topic that producer resolved. assert!( - !Arc::ptr_eq(backends[0], backends[1]), + !Arc::ptr_eq(producers[0], producers[1]), "shards share a producer, so one of them produces to the wrong topic" ); } - /// A reloading and a static topic naming the same Kafka config must not - /// share a client, or one of them silently gets the wrong backend. + /// Logical topics producing to the same physical topic share a producer, + /// while different physical topics on the same cluster do not. #[test] - fn test_static_and_reloading_topics_do_not_share_a_client() { + fn test_producers_are_shared_by_physical_topic() { let cluster = MockCluster::new(1).unwrap(); let servers = cluster.bootstrap_servers(); let params = params(&servers); + let secondary = BTreeMap::new(); + let blob = blob(&servers, &["outcomes", "ingest-events"]); + let outcomes = TopicAssignment::from("outcomes".to_owned()); let events = TopicAssignment::from("ingest-events".to_owned()); - let transactions = TopicAssignment::from("ingest-transactions".to_owned()); - let blob = blob(&servers, &["ingest-events", "ingest-transactions"]); + let outcomes = outcomes.kafka_configs(¶ms, &secondary).unwrap(); + let events = events.kafka_configs(¶ms, &secondary).unwrap(); - // Both use the default broker config, so only the backend keeps their cache keys apart. let client = KafkaClient::builder() - .add_kafka_topic_config( - KafkaTopic::Events, - &events.kafka_configs(¶ms, &BTreeMap::new()).unwrap(), - false, - Some(&ReloadingTopic { - blob: &blob, - app: "relay", - jitter: Duration::ZERO, - }), - ) + .add_kafka_topic_config(KafkaTopic::Outcomes, &outcomes, false, &source(&blob)) .unwrap() .add_kafka_topic_config( - KafkaTopic::Transactions, - &transactions - .kafka_configs(¶ms, &BTreeMap::new()) - .unwrap(), + KafkaTopic::OutcomesBilling, + &outcomes, false, - None, + &source(&blob), ) .unwrap() + .add_kafka_topic_config(KafkaTopic::Events, &events, false, &source(&blob)) + .unwrap() .build(); - let events = &client.producers[&KafkaTopic::Events] - .topic_producers - .producers[0] - .producer; - let transactions = &client.producers[&KafkaTopic::Transactions] - .topic_producers - .producers[0] - .producer; + let outcomes = producers(&client, KafkaTopic::Outcomes)[0]; + let billing = producers(&client, KafkaTopic::OutcomesBilling)[0]; + let events = producers(&client, KafkaTopic::Events)[0]; + + assert!(Arc::ptr_eq(outcomes, billing)); + assert!(!Arc::ptr_eq(outcomes, events)); + } + + /// A topic missing from the blob cannot get a producer, so building fails + /// instead of producing with a config nobody asked for. + #[test] + fn test_topic_missing_from_blob_is_an_error() { + let cluster = MockCluster::new(1).unwrap(); + let servers = cluster.bootstrap_servers(); + let blob = blob(&servers, &["ingest-events"]); + let params = params(&servers); + let secondary = BTreeMap::new(); + let assignment = TopicAssignment::from("ingest-transactions".to_owned()); + let configs = assignment.kafka_configs(¶ms, &secondary).unwrap(); + + let result = KafkaClient::builder().add_kafka_topic_config( + KafkaTopic::Transactions, + &configs, + false, + &source(&blob), + ); - assert!(matches!(**events, Backend::Reloading(_))); - assert!(matches!(**transactions, Backend::Static(_))); + assert!(matches!(result, Err(ClientError::Reload(_)))); } } diff --git a/relay-kafka/src/producer/utils.rs b/relay-kafka/src/producer/utils.rs index b4fe90a681f..8d290f2c45c 100644 --- a/relay-kafka/src/producer/utils.rs +++ b/relay-kafka/src/producer/utils.rs @@ -75,7 +75,7 @@ where /// Kafka client and producer context that logs statistics and producer errors. /// -/// Cloned once per client a [`super::Backend::Reloading`] producer builds, so that the +/// Cloned once per client a producer builds, so that the /// callbacks keep reporting across a config reload. #[derive(Clone)] pub struct Context { diff --git a/relay-kafka/src/statsd.rs b/relay-kafka/src/statsd.rs index d736b824a76..8274143790e 100644 --- a/relay-kafka/src/statsd.rs +++ b/relay-kafka/src/statsd.rs @@ -79,7 +79,7 @@ pub enum KafkaCounters { /// - `producer_name`: The configured producer name/deployment identifier. ProduceStatusError, - /// Number of Kafka config blobs handed to a topic's reloading producer. + /// Number of Kafka config blobs handed to a topic's producer. /// /// Counts pushes that were accepted for rollout, which includes blobs that resolve to the /// config already running. It does not mean the client was replaced; watch @@ -90,7 +90,7 @@ pub enum KafkaCounters { /// - `producer_name`: The configured producer name/deployment identifier. ConfigReloadPushed, - /// Number of Kafka config blobs a topic's reloading producer refused. + /// Number of Kafka config blobs a topic's producer refused. /// /// Rejection leaves the running client in place. Sustained non-zero values mean this topic /// is no longer applying config changes. @@ -146,7 +146,7 @@ pub enum KafkaGauges { /// The number of times this topic's Kafka client has been replaced by a config reload. /// - /// Always `0` for topics not using the reloading producer. Arroyo reports the same number + /// Arroyo reports the same number /// under `arroyo.producer.config_reload_generation`; this one is tagged the way Relay tags /// the rest of its produce metrics, so it lines up with them. /// diff --git a/relay-server/src/services/config_reload.rs b/relay-server/src/services/config_reload.rs index 76660ff2f7a..d141b041867 100644 --- a/relay-server/src/services/config_reload.rs +++ b/relay-server/src/services/config_reload.rs @@ -12,8 +12,7 @@ pub struct ConfigReloadService { config: Arc, /// The store's Kafka client, when this Relay produces to Kafka. /// - /// Topics using the reloading producer take their new broker settings from here; see - /// `processing.reloading_producer_topics`. Everything else picks them up on restart. + /// Every topic's producer takes its new broker settings from here. #[cfg(feature = "processing")] kafka_client: Option>, } diff --git a/relay-server/src/services/store.rs b/relay-server/src/services/store.rs index 1db824c83fc..5194e0f5a77 100644 --- a/relay-server/src/services/store.rs +++ b/relay-server/src/services/store.rs @@ -27,7 +27,7 @@ use relay_common::time::UnixTimestamp; use relay_config::{Config, ConfigSnapshot}; use relay_event_schema::protocol::{Event, EventId, SpanV2, datetime_to_timestamp}; use relay_kafka::{ - ClientError, KafkaClient, KafkaTopic, Message, ReloadingTopic, SerializationOutput, + ClientError, KafkaClient, KafkaConfigSource, KafkaTopic, Message, SerializationOutput, }; use relay_metrics::{ Bucket, BucketView, BucketViewValue, BucketsView, ByNamespace, GaugeValue, MetricName, @@ -101,30 +101,22 @@ impl Producer { pub fn create(config: &ConfigSnapshot) -> anyhow::Result { let mut client_builder = KafkaClient::builder(); - // Only needed by topics on the reloading producer, and building it walks every topic, - // so skip it entirely when no topic asked for one. - let blob = config - .has_reloading_producer_topics() - .then(|| config.kafka_config_blob()) - .transpose()?; + let blob = config.kafka_config_blob()?; + let source = KafkaConfigSource { + blob: &blob, + app: KAFKA_CONFIG_APP, + jitter: config.reloading_producer_jitter(), + }; for topic in KafkaTopic::iter() { let kafka_configs = config.kafka_configs(*topic)?; - let reloading = blob - .as_ref() - .filter(|_| config.is_reloading_producer_topic(*topic)) - .map(|blob| ReloadingTopic { - blob, - app: KAFKA_CONFIG_APP, - jitter: config.reloading_producer_jitter(), - }); client_builder = client_builder .add_kafka_topic_config( *topic, &kafka_configs, config.kafka_validate_topics(), - reloading.as_ref(), + &source, ) .map_err(|e| ServiceError::Kafka(e.to_string()))?; } diff --git a/tests/integration/test_config.py b/tests/integration/test_config.py index 0b191205044..adbad304b93 100644 --- a/tests/integration/test_config.py +++ b/tests/integration/test_config.py @@ -105,14 +105,12 @@ def test_variable_loaded_from_file_is_reloaded(mini_sentry, relay): time.sleep(0.5) -def test_unknown_reloading_producer_topic_is_ignored( +def test_removed_reloading_producer_topics_option_is_ignored( mini_sentry, relay_with_processing ): """ - Names in `reloading_producer_topics` that are not topics must not stop Relay from starting. - - The option names logical topics and is set per environment, so a name that is wrong, or - belongs to a topic a future Relay removed, has to be inert rather than fatal. + `reloading_producer_topics` was removed, but deployments still set it until their config + catches up. It must not stop Relay from starting. """ relay = relay_with_processing( options={ diff --git a/tests/integration/test_store.py b/tests/integration/test_store.py index a7086859983..d6265e20c5c 100644 --- a/tests/integration/test_store.py +++ b/tests/integration/test_store.py @@ -1226,9 +1226,7 @@ def test_reloading_producer_swaps_the_client_on_a_kafka_config_change( Rewriting `kafka_config` must replace the Kafka client underneath a running Relay, without interrupting produce. - This is the whole point of the feature: with a static client the config change is simply - ignored until a restart, so the test asserts the swap actually happened rather than just - that events still arrive. + The test asserts the swap actually happened rather than just that events still arrive. """ events_consumer = events_consumer() @@ -1242,7 +1240,6 @@ def test_reloading_producer_swaps_the_client_on_a_kafka_config_change( "relay": {"config_reload_interval": 1}, "metrics": {"statsd": f"{host}:{port}", "prefix": "relay"}, "processing": { - "reloading_producer_topics": ["ingest-events"], # Swap as soon as the config lands, instead of spreading it out. "reloading_producer_jitter_secs": 0, }, From dd303958abb83a8912316efb93fd499d45103cd9 Mon Sep 17 00:00:00 2001 From: Markus Unterwaditzer Date: Thu, 22 Oct 2026 16:10:00 +0200 Subject: [PATCH 2/3] ref(changelog): Describe the reloading producer as always on --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ed22b9c196a..47161f6f316 100644 --- a/CHANGELOG.md +++ b/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 From 70775ed0ad2577a2b33fea247ca7117fe1fcc9e2 Mon Sep 17 00:00:00 2001 From: Markus Unterwaditzer Date: Thu, 22 Oct 2026 16:49:26 +0200 Subject: [PATCH 3/3] fix(kafka): Reject one Kafka topic name on two clusters The config blob is keyed by topic name, so this case silently sent one of the topics to the other's cluster. No deployment uses it; the integration test that did now moves both outcome topics. ref STREAM-1939 --- relay-config/src/config.rs | 45 ++++++++++++++++++++++++++++--- relay-kafka/src/config.rs | 3 +++ tests/integration/test_outcome.py | 12 +++++---- 3 files changed, 51 insertions(+), 9 deletions(-) diff --git a/relay-config/src/config.rs b/relay-config/src/config.rs index 5eba7785b15..8452d03784d 100644 --- a/relay-config/src/config.rs +++ b/relay-config/src/config.rs @@ -1,3 +1,4 @@ +use std::collections::btree_map::Entry; use std::collections::{BTreeMap, HashMap}; use std::error::Error; use std::io::Write; @@ -2895,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(_) => {} + } } } @@ -3271,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] diff --git a/relay-kafka/src/config.rs b/relay-kafka/src/config.rs index c5cbef20f74..6cc492f30f7 100644 --- a/relay-kafka/src/config.rs +++ b/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. diff --git a/tests/integration/test_outcome.py b/tests/integration/test_outcome.py index 5ffd300027a..70272319ddd 100644 --- a/tests/integration/test_outcome.py +++ b/tests/integration/test_outcome.py @@ -136,11 +136,13 @@ def test_outcomes_custom_topic( options["processing"]["secondary_kafka_configs"] = {} options["processing"]["secondary_kafka_configs"]["foo"] = kafka_config - # ...however, we use a custom topic config to make it work again - options["processing"]["topics"]["outcomes"] = { - "name": get_topic_name("outcomes"), - "config": "foo", - } + # ...however, we use a custom topic config to make it work again. Billing outcomes share the + # topic, so they have to move along with it. + for topic in ("outcomes", "outcomes_billing"): + options["processing"]["topics"][topic] = { + "name": get_topic_name("outcomes"), + "config": "foo", + } outcomes_consumer = outcomes_consumer()