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 diff --git a/relay-config/src/config.rs b/relay-config/src/config.rs index 6f44ec579b3..8452d03784d 100644 --- a/relay-config/src/config.rs +++ b/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, - /// 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] 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/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_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() 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, },