Round-robin multiple Kafka topics (#299)

round-robin multiple topics when topic contains comma

---------

Co-authored-by: Daryl Coburn <daryl_coburn@cargill.com>
Co-authored-by: Vincent Ruello <5345986+vruello@users.noreply.github.com>

Author: 3 people

doc/outputs.md

@@ -73,7 +73,9 @@ $ openwec subscriptions edit <subscription> outputs add --format <format> files
 
 The Kafka driver sends events in a Kafka topic.
 
-For a given subscription, all events will be sent in the configured Kafka topic. You may want to add additionnal options to the inner Kafka client, such as `bootstrap.servers`. This options will be directly given to `librdkafka` (available options are listed here: https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html).
+For a given subscription, all events will be sent in the configured Kafka topic. If multiple topics are provided, delimited by commas, then the events are evenly distributed across the listed topics.
+
+You may want to add additionnal options to the inner Kafka client, such as `bootstrap.servers`. This options will be directly given to `librdkafka` (available options are listed here: https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html).
 
 > [!TIP]
 > If multiple outputs use the Kafka driver and connect to the same Kafka cluster, it is recommended to configure the additional options in OpenWEC settings (`outputs.kafka.options`) **and** to omit the `options` parameter in Kafka output configuration. This way, only one Kafka client will be used by all the outputs, which is more resource efficient.

server/src/drivers/kafka.rs

@@ -41,8 +41,8 @@ impl OutputKafkaContext {
 }
 
 pub struct OutputKafka {
-    config: KafkaConfiguration,
     producer: FutureProducer,
+    topics: Vec<String>,
 }
 
 impl OutputKafka {
@@ -72,9 +72,15 @@ impl OutputKafka {
             );
             client_config.create()?
         };
+
         Ok(OutputKafka {
-            config: config.clone(),
             producer,
+            topics: config
+                .topic()
+                .split(',')
+                .map(|s| s.trim().to_string())
+                .filter(|s| !s.is_empty())
+                .collect(),
         })
     }
 }
@@ -86,13 +92,17 @@ impl OutputDriver for OutputKafka {
         _metadata: Arc<EventMetadata>,
         events: Arc<Vec<Arc<String>>>,
     ) -> Result<()> {
+        let mut l_topic_index: usize = 0;
         let mut futures = Vec::new();
         for event in events.iter() {
+            // Get current topic
+            let topic = self.topics[l_topic_index].as_ref();
             // We need to explicitly assign the Key type as ()
             futures.push(self.producer.send::<(), _, _>(
-                FutureRecord::to(self.config.topic()).payload(event.as_ref()),
+                FutureRecord::to(topic).payload(event.as_ref()),
                 Timeout::After(Duration::from_secs(30)),
             ));
+            l_topic_index = (l_topic_index + 1) % self.topics.len();
         }
 
         // Wait for all events to be sent and ack