diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java index 6ed49d21fc..8e2ab69915 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java @@ -164,7 +164,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.clientId("monolith-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); - consumerBuilder.groupId("monolith-rule-engine-notifications-consumer"); + consumerBuilder.groupId("monolith-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(notificationAdmin); return consumerBuilder.build(); @@ -188,7 +188,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.clientId("monolith-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); - consumerBuilder.groupId("monolith-core-notifications-consumer"); + consumerBuilder.groupId("monolith-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(notificationAdmin); return consumerBuilder.build(); @@ -229,7 +229,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi responseBuilder.settings(kafkaSettings); responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); - responseBuilder.groupId("rule-engine-node"); + responseBuilder.groupId("rule-engine-node-" + serviceInfoProvider.getServiceId()); responseBuilder.decoder(msg -> { JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java index 7ca2f9196d..c6acc0fd6a 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java @@ -158,7 +158,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.clientId("tb-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); - consumerBuilder.groupId("tb-core-notifications-node"); + consumerBuilder.groupId("tb-core-notifications-node-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(notificationAdmin); return consumerBuilder.build(); @@ -199,7 +199,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { responseBuilder.settings(kafkaSettings); responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); - responseBuilder.groupId("rule-engine-node"); + responseBuilder.groupId("rule-engine-node-" + serviceInfoProvider.getServiceId()); responseBuilder.decoder(msg -> { JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java index 3bab6d1866..ad8d84b4b6 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java @@ -153,7 +153,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.clientId("tb-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); - consumerBuilder.groupId("tb-rule-engine-notifications-node"); + consumerBuilder.groupId("tb-rule-engine-notifications-node-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(notificationAdmin); return consumerBuilder.build(); @@ -172,7 +172,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { responseBuilder.settings(kafkaSettings); responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); - responseBuilder.groupId("rule-engine-node"); + responseBuilder.groupId("rule-engine-node-" + serviceInfoProvider.getServiceId()); responseBuilder.decoder(msg -> { JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java index 58623a8f0b..505aa203d7 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java @@ -91,7 +91,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { responseBuilder.settings(kafkaSettings); responseBuilder.topic(transportApiSettings.getResponsesTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("transport-api-response-" + serviceInfoProvider.getServiceId()); - responseBuilder.groupId("transport-node"); + responseBuilder.groupId("transport-node-" + serviceInfoProvider.getServiceId()); responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders())); responseBuilder.admin(transportApiAdmin); @@ -132,7 +132,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { responseBuilder.settings(kafkaSettings); responseBuilder.topic(transportNotificationSettings.getNotificationsTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("transport-api-notifications-" + serviceInfoProvider.getServiceId()); - responseBuilder.groupId("transport-node"); + responseBuilder.groupId("transport-node-" + serviceInfoProvider.getServiceId()); responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); responseBuilder.admin(notificationAdmin); return responseBuilder.build();