From c949af5fdc97dc4d61da23610129b038d53b60aa Mon Sep 17 00:00:00 2001 From: Leonardo Rodoni Date: Thu, 13 Aug 2026 13:24:27 +0200 Subject: [PATCH] fix(collector): subscribe to flow-recv actor only when needed for Move the extra supervisor.subscribe() call for FlowOptionsActorHandle inside the enrichment input branch instead of creating it unconditionally for every FlowKafkaAvro/FlowKafkaJson publisher. Previously an unused subscription (and its buffered channel) was created even when no flow_options input was configured, wasting memory and adding a needless subscriber to the flow supervisor. --- crates/collector/src/lib.rs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/crates/collector/src/lib.rs b/crates/collector/src/lib.rs index 50412619..2f3b9af1 100644 --- a/crates/collector/src/lib.rs +++ b/crates/collector/src/lib.rs @@ -180,16 +180,16 @@ pub async fn init_flow_collection( } } - let (flow_recv, _) = supervisor_handle - .subscribe(publisher_config.buffer_size) - .await?; - if let Some(enrichment_config) = publisher_config.enrichment.as_ref() { if let Some(flow_options_config) = enrichment_config .inputs .as_ref() .and_then(|i| i.flow_options.as_ref()) { + let (flow_recv, _) = supervisor_handle + .subscribe(publisher_config.buffer_size) + .await?; + let (flow_options_join, flow_options_handle) = FlowOptionsActorHandle::from_config( flow_options_config, @@ -279,16 +279,16 @@ pub async fn init_flow_collection( } } - let (flow_recv, _) = supervisor_handle - .subscribe(publisher_config.buffer_size) - .await?; - if let Some(enrichment_config) = publisher_config.enrichment.as_ref() { if let Some(flow_options_config) = enrichment_config .inputs .as_ref() .and_then(|i| i.flow_options.as_ref()) { + let (flow_recv, _) = supervisor_handle + .subscribe(publisher_config.buffer_size) + .await?; + let (flow_options_join, flow_options_handle) = FlowOptionsActorHandle::from_config( flow_options_config,