refactor: prefix kafka group with application name

drogue-iot · Oct 10, 2022 · e7df7b4 · e7df7b4
1 parent e7347c4
commit e7df7b4
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/mqtt-integration/src/service/session.rs b/mqtt-integration/src/service/session.rs
@@ -190,7 +190,7 @@ impl Session {
                 .kafka_target(KafkaEventType::Events, &self.config.kafka)
                 .map(|target| target.into())
                 .map_err(|_| v5::codec::SubscribeAckReason::UnspecifiedError)?,
-            consumer_group: group_id.map(|s| s.to_string()),
+            consumer_group: group_id.map(|s| format!("{app}.{s}")),
         };
         let event_stream = EventStream::<CustomAck>::new(stream_config).map_err(|err| {
             log::info!("Failed to subscribe to Kafka topic: {}", err);

diff --git a/websocket-integration/src/service.rs b/websocket-integration/src/service.rs
@@ -147,7 +147,7 @@ impl Service {
                 .kafka_target(KafkaEventType::Events, kafka_config)
                 .map_err(|_| ServiceError::InternalError("This should be infallible".into()))?
                 .into(),
-            consumer_group: group_id,
+            consumer_group: group_id.map(|group_id| format!("{application}.{group_id}")),
         })
         .map_err(|err| {
             log::info!("Failed to subscribe to Kafka topic: {}", err);