Improvements and bugfix to QueueProviders

application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@@ -35,6 +35,7 @@ import org.springframework.context.annotation.Lazy;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Service;
 import org.thingsboard.rule.engine.api.MailService;
 import org.thingsboard.rule.engine.api.RuleChainTransactionService;
 import org.thingsboard.server.actors.service.ActorService;
@@ -64,6 +65,7 @@ import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.dao.tenant.TenantService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.user.UserService;
+import org.thingsboard.server.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.kafka.TbNodeIdProvider;
 import org.thingsboard.server.service.component.ComponentDiscoveryService;
 import org.thingsboard.server.service.encoding.DataDecodingEncodingService;
@@ -102,6 +104,11 @@ public class ActorSystemContext {
         return debugPerTenantLimits;
     }
 
+    @Autowired
+    @Getter
+    @Setter
+    private TbServiceInfoProvider serviceInfoProvider;
+
     @Getter
     @Setter
     private ActorService actorService;
@@ -233,10 +240,6 @@ public class ActorSystemContext {
     @Getter
     private RuleChainTransactionService ruleChainTransactionService;
 
-    @Value("${cluster.partition_id}")
-    @Getter
-    private long queuePartitionId;
-
     @Value("${actors.session.max_concurrent_sessions_per_device:1}")
     @Getter
     private long maxConcurrentSessionsPerDevice;
@@ -359,7 +362,7 @@ public class ActorSystemContext {
         event.setEntityId(entityId);
         event.setType(DataConstants.ERROR);
         //TODO 2.5
-//        event.setBody(toBodyJson(discoveryService.getCurrentServer().getServerAddress(), method, toString(e)));
+        event.setBody(toBodyJson(serviceInfoProvider.getServiceInfo().getServiceId(), method, toString(e)));
         persistEvent(event);
     }
 
@@ -369,7 +372,7 @@ public class ActorSystemContext {
         event.setEntityId(entityId);
         event.setType(DataConstants.LC_EVENT);
         //TODO 2.5
-//        event.setBody(toBodyJson(discoveryService.getCurrentServer().getServerAddress(), lcEvent, Optional.ofNullable(e)));
+        event.setBody(toBodyJson(serviceInfoProvider.getServiceInfo().getServiceId(), lcEvent, Optional.ofNullable(e)));
         persistEvent(event);
     }
 
@@ -383,8 +386,8 @@ public class ActorSystemContext {
         return sw.toString();
     }
 
-    private JsonNode toBodyJson(ServerAddress server, ComponentLifecycleEvent event, Optional<Exception> e) {
+    private JsonNode toBodyJson(String serviceId, ComponentLifecycleEvent event, Optional<Exception> e) {
-        ObjectNode node = mapper.createObjectNode().put("server", server.toString()).put("event", event.name());
+        ObjectNode node = mapper.createObjectNode().put("server", serviceId).put("event", event.name());
         if (e.isPresent()) {
             node = node.put("success", false);
             node = node.put("error", toString(e.get()));
@@ -394,8 +397,8 @@ public class ActorSystemContext {
         return node;
     }
 
-    private JsonNode toBodyJson(ServerAddress server, String method, String body) {
+    private JsonNode toBodyJson(String serviceId, String method, String body) {
-        return mapper.createObjectNode().put("server", server.toString()).put("method", method).put("error", body);
+        return mapper.createObjectNode().put("server", serviceId).put("method", method).put("error", body);
     }
 
 

application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java

@@ -146,12 +146,12 @@ class DefaultTbContext implements TbContext {
 
     @Override
     public TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data) {
-        return new TbMsg(UUIDs.timeBased(), type, originator, metaData.copy(), data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId(), mainCtx.getQueuePartitionId());
+        return new TbMsg(UUIDs.timeBased(), type, originator, metaData.copy(), data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId(), 0);
     }
 
     @Override
     public TbMsg transformMsg(TbMsg origMsg, String type, EntityId originator, TbMsgMetaData metaData, String data) {
-        return new TbMsg(origMsg.getId(), type, originator, metaData.copy(), origMsg.getDataType(), data, origMsg.getTransactionData(), origMsg.getRuleChainId(), origMsg.getRuleNodeId(), mainCtx.getQueuePartitionId());
+        return new TbMsg(origMsg.getId(), type, originator, metaData.copy(), origMsg.getDataType(), data, origMsg.getTransactionData(), origMsg.getRuleChainId(), origMsg.getRuleNodeId(), 0);
     }
 
     @Override

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java

@@ -331,6 +331,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
 
     private TbMsg enrichWithRuleChainId(TbMsg tbMsg) {
         // We don't put firstNodeId because it may change over time;
-        return new TbMsg(tbMsg.getId(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.getMetaData().copy(), tbMsg.getData(), entityId, null, systemContext.getQueuePartitionId());
+        return new TbMsg(tbMsg.getId(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.getMetaData().copy(), tbMsg.getData(), entityId, null, 0);
     }
 }

application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java

@@ -27,6 +27,8 @@ import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.TbQueueConsumer;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.common.TbProtoQueueMsg;
+import org.thingsboard.server.discovery.PartitionChangeEvent;
+import org.thingsboard.server.discovery.ServiceType;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg;
 import org.thingsboard.server.provider.TbCoreQueueProvider;
@@ -69,10 +71,17 @@ public class DefaultTbCoreConsumerService implements TbCoreConsumerService {
 
     @PostConstruct
     public void init() {
-        this.consumer.subscribe();
         this.mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-core-consumer"));
     }
 
+    @Override
+    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+        if (partitionChangeEvent.getServiceKey().getServiceType() == ServiceType.TB_CORE) {
+            log.info("Subscribing to partitions: {}", partitionChangeEvent.getPartitions());
+            this.consumer.subscribe(partitionChangeEvent.getPartitions());
+        }
+    }
+
     @EventListener(ApplicationReadyEvent.class)
     public void onApplicationEvent(ApplicationReadyEvent event) {
         mainConsumerExecutor.execute(() -> {

application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -15,7 +15,6 @@
  */
 package org.thingsboard.server.service.queue;
 
-import akka.actor.ActorRef;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
@@ -27,10 +26,10 @@ import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.TbQueueConsumer;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.common.TbProtoQueueMsg;
+import org.thingsboard.server.discovery.PartitionChangeEvent;
+import org.thingsboard.server.discovery.ServiceType;
 import org.thingsboard.server.gen.transport.TransportProtos;
-import org.thingsboard.server.provider.TbCoreQueueProvider;
 import org.thingsboard.server.provider.TbRuleEngineQueueProvider;
-import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
@@ -70,10 +69,17 @@ public class DefaultTbRuleEngineConsumerService implements TbRuleEngineConsumerS
 
     @PostConstruct
     public void init() {
-        this.consumer.subscribe();
         this.mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-core-consumer"));
     }
 
+    @Override
+    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+        if (partitionChangeEvent.getServiceKey().getServiceType() == ServiceType.TB_RULE_ENGINE) {
+            log.info("Subscribing to partitions: {}", partitionChangeEvent.getPartitions());
+            this.consumer.subscribe(partitionChangeEvent.getPartitions());
+        }
+    }
+
     @EventListener(ApplicationReadyEvent.class)
     public void onApplicationEvent(ApplicationReadyEvent event) {
         mainConsumerExecutor.execute(() -> {

application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerService.java

@@ -15,10 +15,12 @@
  */
 package org.thingsboard.server.service.queue;
 
+import org.springframework.context.ApplicationListener;
+import org.thingsboard.server.discovery.PartitionChangeEvent;
 import org.thingsboard.server.gen.transport.TransportProtos;
 
 import java.util.function.Consumer;
 
-public interface TbCoreConsumerService {
+public interface TbCoreConsumerService extends ApplicationListener<PartitionChangeEvent> {
 
 }

application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerService.java

@@ -15,6 +15,9 @@
  */
 package org.thingsboard.server.service.queue;
 
-public interface TbRuleEngineConsumerService {
+import org.springframework.context.ApplicationListener;
+import org.thingsboard.server.discovery.PartitionChangeEvent;
+
+public interface TbRuleEngineConsumerService extends ApplicationListener<PartitionChangeEvent> {
 
 }

application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

application/src/main/java/org/thingsboard/server/service/transport/DefaultTbCoreToTransportService.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -40,7 +40,7 @@ public class DefaultTbCoreToTransportService implements TbCoreToTransportService
 
     private final TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> tbTransportProducer;
 
-    @Value("${queue.notifications.topic}")
+    @Value("${queue.transport.notifications_topic}")
     private String notificationsTopic;
 
     public DefaultTbCoreToTransportService(TbCoreQueueProvider tbCoreQueueProvider) {

application/src/main/resources/thingsboard.yml

@@ -70,16 +70,7 @@ zk:
   # Name of the directory in zookeeper 'filesystem'
   zk_dir: "${ZOOKEEPER_NODES_DIR:/thingsboard}"
 
-# Clustering properties related to consistent-hashing. See architecture docs for more details.
 cluster:
-  # Unique id for this node (autogenerated if empty)
-  node_id: "${CLUSTER_NODE_ID:}"
-  # Name of hash function used for consistent hash ring.
-  hash_function_name: "${CLUSTER_HASH_FUNCTION_NAME:murmur3_128}"
-  # Amount of virtual nodes in consistent hash ring.
-  vitrual_nodes_size: "${CLUSTER_VIRTUAL_NODES_SIZE:16}"
-  # Queue partition id for current node
-  partition_id: "${QUEUE_PARTITION_ID:0}"
   stats:
     enabled: "${TB_CLUSTER_STATS_ENABLED:false}"
     print_interval_ms: "${TB_CLUSTER_STATS_PRINT_INTERVAL_MS:10000}"
@@ -536,6 +527,9 @@ queue:
     batch.size: "${TB_KAFKA_BATCH_SIZE:16384}"
     linger.ms: "${TB_KAFKA_LINGER_MS:1}"
     buffer.memory: "${TB_BUFFER_MEMORY:33554432}"
+  partitions:
+    hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}"
+    virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}"
   transport_api:
     requests_topic: "${TB_QUEUE_TRANSPORT_API_REQUEST_TOPIC:tb.transport.api.requests}"
     responses_topic: "${TB_QUEUE_TRANSPORT_API_RESPONSE_TOPIC:tb.transport.api.responses}"
@@ -547,7 +541,7 @@ queue:
   core:
     topic: "${TB_QUEUE_CORE_TOPIC:tb.core}"
     poll_interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}"
-    partitions: "${TB_QUEUE_CORE_PARTITIONS:100}"
+    partitions: "${TB_QUEUE_CORE_PARTITIONS:10}"
     pack_processing_timeout: "${TB_QUEUE_CORE_PACK_PROCESSING_TIMEOUT_MS:60000}"
     stats:
       enabled: "${TB_QUEUE_CORE_STATS_ENABLED:false}"
@@ -555,13 +549,14 @@ queue:
   rule_engine:
     topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb.rule-engine}"
     poll_interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}"
-    partitions: "${TB_QUEUE_RULE_ENGINE_PARTITIONS:100}"
+    partitions: "${TB_QUEUE_RULE_ENGINE_PARTITIONS:10}"
     pack_processing_timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}"
     stats:
       enabled: "${TB_QUEUE_RULE_ENGINE_STATS_ENABLED:false}"
       print_interval_ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}"
-  notifications:
+  transport:
-    topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb.transport.notifications}"
+    notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb.transport.notifications}"
+    poll_interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}"
 
 service:
   type: "${TB_SERVICE_TYPE:monolith}" # monolith or tb-core or tb-rule-engine or tb-transport

application/src/test/java/org/thingsboard/server/service/cluster/routing/ConsistentClusterRoutingServiceTest.java → application/src/test/java/org/thingsboard/server/service/cluster/routing/ConsistentHashParitionServiceTest.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -21,11 +21,10 @@ import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.runners.MockitoJUnitRunner;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.test.util.ReflectionTestUtils;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.common.msg.cluster.ServerAddress;
-import org.thingsboard.server.common.msg.cluster.ServerType;
 import org.thingsboard.server.discovery.ConsistentHashPartitionService;
 import org.thingsboard.server.discovery.ServiceType;
 import org.thingsboard.server.discovery.TbServiceInfoProvider;
@@ -44,12 +43,13 @@ import static org.mockito.Mockito.mock;
 
 @Slf4j
 @RunWith(MockitoJUnitRunner.class)
-public class ConsistentClusterRoutingServiceTest {
+public class ConsistentHashParitionServiceTest {
 
     public static final int ITERATIONS = 1000000;
     private ConsistentHashPartitionService clusterRoutingService;
 
     private TbServiceInfoProvider discoveryService;
+    private ApplicationEventPublisher applicationEventPublisher;
 
     private String hashFunctionName = "murmur3_128";
     private Integer virtualNodesSize = 16;
@@ -58,7 +58,8 @@ public class ConsistentClusterRoutingServiceTest {
     @Before
     public void setup() throws Exception {
         discoveryService = mock(TbServiceInfoProvider.class);
-        clusterRoutingService = new ConsistentHashPartitionService(discoveryService);
+        applicationEventPublisher = mock(ApplicationEventPublisher.class);
+        clusterRoutingService = new ConsistentHashPartitionService(discoveryService, applicationEventPublisher);
         ReflectionTestUtils.setField(clusterRoutingService, "coreTopic", "tb.core");
         ReflectionTestUtils.setField(clusterRoutingService, "corePartitions", 3);
         ReflectionTestUtils.setField(clusterRoutingService, "ruleEngineTopic", "tb.rule-engine");
@@ -103,7 +104,7 @@ public class ConsistentClusterRoutingServiceTest {
         List<Map.Entry<Integer, Integer>> data = map.entrySet().stream().sorted(Comparator.comparingInt(Map.Entry::getValue)).collect(Collectors.toList());
         long end = System.currentTimeMillis();
         double diff = (data.get(data.size() - 1).getValue() - data.get(0).getValue());
-        System.out.println("Size: " + virtualNodesSize + " Time: " + (end - start) + " Diff: " + diff + "(" + String.format("%f", (diff/ITERATIONS) * 100.0) + "%)");
+        System.out.println("Size: " + virtualNodesSize + " Time: " + (end - start) + " Diff: " + diff + "(" + String.format("%f", (diff / ITERATIONS) * 100.0) + "%)");
 
         for (Map.Entry<Integer, Integer> entry : data) {
             System.out.println(entry.getKey() + ": " + entry.getValue());

common/queue/src/main/java/org/thingsboard/server/TbQueueConsumer.java

@@ -23,6 +23,8 @@ public interface TbQueueConsumer<T extends TbQueueMsg> {
 
     void subscribe();
 
+    void subscribe(List<Integer> partitions);
+
     void unsubscribe();
 
     List<T> poll(long durationInMillis);

common/queue/src/main/java/org/thingsboard/server/TbQueueProducer.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

common/queue/src/main/java/org/thingsboard/server/TbQueueRequestTemplate.java

@@ -19,6 +19,10 @@ import com.google.common.util.concurrent.ListenableFuture;
 
 public interface TbQueueRequestTemplate<Request extends TbQueueMsg, Response extends TbQueueMsg> {
 
+    void init();
+
     ListenableFuture<Response> send(Request request);
 
+    void stop();
+
 }

common/queue/src/main/java/org/thingsboard/server/TbQueueRuleEngineSettings.java

@@ -0,0 +1,31 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server;
+
+import lombok.Data;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+@Data
+@Component
+public class TbQueueRuleEngineSettings {
+
+    @Value("${queue.rule_engine.topic}")
+    private String topic;
+
+    @Value("${queue.rule_engine.partitions}")
+    private int partitions;
+}

common/queue/src/main/java/org/thingsboard/server/TbQueueTransportApiSettings.java

@@ -22,6 +22,7 @@ import org.springframework.stereotype.Component;
 @Data
 @Component
 public class TbQueueTransportApiSettings {
+
     @Value("${queue.transport_api.requests_topic}")
     private String requestsTopic;
 
@@ -34,6 +35,12 @@ public class TbQueueTransportApiSettings {
     @Value("${queue.transport_api.max_requests_timeout}")
     private int maxRequestsTimeout;
 
+    @Value("${queue.transport_api.max_callback_threads}")
+    private int maxCallbackThreads;
+
+    @Value("${queue.transport_api.request_poll_interval}")
+    private long requestPollInterval;
+
     @Value("${queue.transport_api.response_poll_interval}")
     private long responsePollInterval;
 

common/queue/src/main/java/org/thingsboard/server/TbQueueTransportNotificationSettings.java

@@ -0,0 +1,32 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server;
+
+import lombok.Data;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+@Data
+@Component
+public class TbQueueTransportNotificationSettings {
+
+    @Value("${queue.transport.notifications_topic}")
+    private String notificationsTopic;
+
+    @Value("${queue.transport.poll_interval}")
+    private long transportPollInterval;
+
+}

common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueRequestTemplate.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -79,6 +79,7 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
         }
     }
 
+    @Override
     public void init() {
         queueAdmin.createTopicIfNotExists(responseTemplate.getTopic());
         this.requestTemplate.init();
@@ -141,6 +142,7 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
         });
     }
 
+    @Override
     public void stop() {
         stopped = true;
         if (internalExecutor) {

common/queue/src/main/java/org/thingsboard/server/discovery/ConsistentHashCircle.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

common/queue/src/main/java/org/thingsboard/server/discovery/ConsistentHashPartitionService.java

@@ -1,12 +1,27 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.thingsboard.server.discovery;
 
 import com.google.common.hash.HashCode;
 import com.google.common.hash.HashFunction;
 import com.google.common.hash.Hashing;
-import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
@@ -20,7 +35,6 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
@@ -39,11 +53,12 @@ public class ConsistentHashPartitionService implements PartitionService {
     private String ruleEngineTopic;
     @Value("${queue.rule_engine.partitions:100}")
     private Integer ruleEnginePartitions;
-    @Value("${queue.partitions.hash_function_name:murmur3_32}")
+    @Value("${queue.partitions.hash_function_name:murmur3_128}")
     private String hashFunctionName;
     @Value("${queue.partitions.virtual_nodes_size:16}")
     private Integer virtualNodesSize;
 
+    private final ApplicationEventPublisher applicationEventPublisher;
     private final TbServiceInfoProvider serviceInfoProvider;
     private final ConcurrentMap<ServiceType, String> partitionTopics = new ConcurrentHashMap<>();
     private final ConcurrentMap<ServiceType, Integer> partitionSizes = new ConcurrentHashMap<>();
@@ -53,8 +68,9 @@ public class ConsistentHashPartitionService implements PartitionService {
 
     private HashFunction hashFunction;
 
-    public ConsistentHashPartitionService(TbServiceInfoProvider serviceInfoProvider) {
+    public ConsistentHashPartitionService(TbServiceInfoProvider serviceInfoProvider, ApplicationEventPublisher applicationEventPublisher) {
         this.serviceInfoProvider = serviceInfoProvider;
+        this.applicationEventPublisher = applicationEventPublisher;
     }
 
     @PostConstruct
@@ -128,13 +144,15 @@ public class ConsistentHashPartitionService implements PartitionService {
             for (int i = 0; i < size; i++) {
                 ServiceInfo serviceInfo = resolveByPartitionIdx(newCircles.get(type), i);
                 if (currentService.equals(serviceInfo)) {
-                    myPartitions.putIfAbsent(new ServiceKey(type, getTenantId(serviceInfo)), new ArrayList<>());
+                    ServiceKey serviceKey = new ServiceKey(type, getTenantId(serviceInfo));
+                    myPartitions.computeIfAbsent(serviceKey, key -> new ArrayList<>()).add(i);
                 }
             }
         });
         myPartitions.forEach((serviceKey, partitions) -> {
             if (!partitions.equals(oldPartitions.get(serviceKey))) {
                 log.info("[{}] NEW PARTITIONS: {}", serviceKey, partitions);
+                applicationEventPublisher.publishEvent(new PartitionChangeEvent(this, serviceKey, partitions));
             }
         });
     }
@@ -178,32 +196,6 @@ public class ConsistentHashPartitionService implements PartitionService {
         return hashFunction.newHasher().putString(instance.getServiceId(), StandardCharsets.UTF_8).putInt(i).hash();
     }
 
-    private static class ServiceKey {
-        @Getter
-        private final ServiceType serviceType;
-        @Getter
-        private final TenantId tenantId;
-
-        public ServiceKey(ServiceType serviceType, TenantId tenantId) {
-            this.serviceType = serviceType;
-            this.tenantId = tenantId;
-        }
-
-        @Override
-        public boolean equals(Object o) {
-            if (this == o) return true;
-            if (o == null || getClass() != o.getClass()) return false;
-            ServiceKey that = (ServiceKey) o;
-            return serviceType == that.serviceType &&
-                    Objects.equals(tenantId, that.tenantId);
-        }
-
-        @Override
-        public int hashCode() {
-            return Objects.hash(serviceType, tenantId);
-        }
-    }
-
     public static HashFunction forName(String name) {
         switch (name) {
             case "murmur3_32":

common/queue/src/main/java/org/thingsboard/server/discovery/DefaultTbServiceInfoProvider.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

common/queue/src/main/java/org/thingsboard/server/discovery/DummyDiscoveryService.java

@@ -1,3 +1,18 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.thingsboard.server.discovery;
 
 import lombok.extern.slf4j.Slf4j;

common/queue/src/main/java/org/thingsboard/server/discovery/PartitionChangeEvent.java

@@ -16,6 +16,7 @@
 package org.thingsboard.server.discovery;
 
 import lombok.Getter;
+import lombok.Setter;
 import org.springframework.context.ApplicationEvent;
 
 import java.util.List;
@@ -23,11 +24,14 @@ import java.util.List;
 
 public class PartitionChangeEvent extends ApplicationEvent {
 
+    @Getter
+    private final ServiceKey serviceKey;
     @Getter
     private final List<Integer> partitions;
 
-    public PartitionChangeEvent(Object source, List<Integer> partitions) {
+    public PartitionChangeEvent(Object source, ServiceKey serviceKey, List<Integer> partitions) {
         super(source);
+        this.serviceKey = serviceKey;
         this.partitions = partitions;
     }
 }

common/queue/src/main/java/org/thingsboard/server/discovery/PartitionService.java

@@ -1,3 +1,18 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.thingsboard.server.discovery;
 
 import org.thingsboard.server.common.data.id.EntityId;

common/queue/src/main/java/org/thingsboard/server/discovery/ServiceKey.java

@@ -0,0 +1,47 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.discovery;
+
+import lombok.Getter;
+import org.thingsboard.server.common.data.id.TenantId;
+
+import java.util.Objects;
+
+public class ServiceKey {
+    @Getter
+    private final ServiceType serviceType;
+    @Getter
+    private final TenantId tenantId;
+
+    public ServiceKey(ServiceType serviceType, TenantId tenantId) {
+        this.serviceType = serviceType;
+        this.tenantId = tenantId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ServiceKey that = (ServiceKey) o;
+        return serviceType == that.serviceType &&
+                Objects.equals(tenantId, that.tenantId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(serviceType, tenantId);
+    }
+}

common/queue/src/main/java/org/thingsboard/server/discovery/ZkPartitionDiscoveryService.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaConsumerTemplate.java

@@ -31,6 +31,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Properties;
+import java.util.stream.Collectors;
 
 /**
  * Created by ashvayka on 24.09.18.
@@ -40,6 +41,7 @@ public class TBKafkaConsumerTemplate<T extends TbQueueMsg> implements TbQueueCon
 
     private final KafkaConsumer<String, byte[]> consumer;
     private final TbKafkaDecoder<T> decoder;
+    private volatile boolean subscribed;
 
     @Getter
     private final String topic;
@@ -69,23 +71,37 @@ public class TBKafkaConsumerTemplate<T extends TbQueueMsg> implements TbQueueCon
     @Override
     public void subscribe() {
         consumer.subscribe(Collections.singletonList(topic));
+        subscribed = true;
+    }
+
+    @Override
+    public void subscribe(List<Integer> partitions) {
+        consumer.subscribe(partitions.stream().map(partition -> topic + "." + partition).collect(Collectors.toList()));
+        subscribed = true;
     }
 
     @Override
     public List<T> poll(long durationInMillis) {
-        ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(durationInMillis));
+        if (!subscribed) {
-        if (records.count() > 0) {
+            try {
-            List<T> result = new ArrayList<>();
+                Thread.sleep(durationInMillis);
-            records.forEach(record -> {
+            } catch (InterruptedException e) {
-                try {
+                log.debug("Failed to await subscription", e);
-                    result.add(decode(record));
+            }
-                } catch (IOException e) {
+        } else {
-                    log.error("Failed decode record: [{}]", record);
+            ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(durationInMillis));
-                }
+            if (records.count() > 0) {
-            });
+                List<T> result = new ArrayList<>();
-            return result;
+                records.forEach(record -> {
+                    try {
+                        result.add(decode(record));
+                    } catch (IOException e) {
+                        log.error("Failed decode record: [{}]", record);
+                    }
+                });
+                return result;
+            }
         }
-
         return Collections.emptyList();
     }
 

common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

common/queue/src/main/java/org/thingsboard/server/memory/InMemoryStorage.java

@@ -15,16 +15,24 @@
  */
 package org.thingsboard.server.memory;
 
+import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.TbQueueMsg;
 
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 import java.util.Queue;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 
+@Slf4j
 public final class InMemoryStorage {
     private static InMemoryStorage instance;
-    private final Map<String, Queue<TbQueueMsg>> storage;
+    private final Map<String, BlockingQueue<TbQueueMsg>> storage;
 
     private InMemoryStorage() {
         storage = new ConcurrentHashMap<>();
@@ -42,19 +50,38 @@ public final class InMemoryStorage {
     }
 
     public boolean put(String topic, TbQueueMsg msg) {
-        return storage.computeIfAbsent(topic, (t) -> new LinkedList<>()).add(msg);
+        return storage.computeIfAbsent(topic, (t) -> new LinkedBlockingQueue<>()).add(msg);
     }
 
-    public TbQueueMsg get(String topic) {
+    public <T extends TbQueueMsg> List<T> get(String topic, long durationInMillis) {
         if (storage.containsKey(topic)) {
-            return storage.get(topic).peek();
+            try {
+                List<T> entities;
+                T first = (T) storage.get(topic).poll(durationInMillis, TimeUnit.MILLISECONDS);
+                if (first != null) {
+                    entities = new ArrayList<>();
+                    entities.add(first);
+                } else {
+                    entities = Collections.emptyList();
+                    List<TbQueueMsg> otherList = new ArrayList<>();
+                    storage.get(topic).drainTo(otherList, 100);
+                    for (TbQueueMsg other : otherList) {
+                        entities.add((T) other);
+                    }
+                }
+                return entities;
+            } catch (InterruptedException e) {
+                log.warn("Queue was interrupted", e);
+                return Collections.emptyList();
+            }
         }
-        return null;
+        return Collections.emptyList();
     }
 
     public void commit(String topic) {
+        //TODO: 2.5 Until someone calls commit you should not allow to poll new elements.
         if (storage.containsKey(topic)) {
-            storage.get(topic).remove();
+//            storage.get(topic).remove();
         }
     }
 }

common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueConsumer.java

@@ -18,7 +18,6 @@ package org.thingsboard.server.memory;
 import org.thingsboard.server.TbQueueConsumer;
 import org.thingsboard.server.TbQueueMsg;
 
-import java.util.Collections;
 import java.util.List;
 
 public class InMemoryTbQueueConsumer<T extends TbQueueMsg> implements TbQueueConsumer<T> {
@@ -40,6 +39,11 @@ public class InMemoryTbQueueConsumer<T extends TbQueueMsg> implements TbQueueCon
 
     }
 
+    @Override
+    public void subscribe(List<Integer> partitions) {
+
+    }
+
     @Override
     public void unsubscribe() {
 
@@ -47,7 +51,7 @@ public class InMemoryTbQueueConsumer<T extends TbQueueMsg> implements TbQueueCon
 
     @Override
     public List<T> poll(long durationInMillis) {
-        return Collections.singletonList((T)storage.get(topic));
+        return storage.get(topic, durationInMillis);
     }
 
     @Override

common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueProducer.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -41,10 +41,13 @@ public class InMemoryTbQueueProducer<T extends TbQueueMsg> implements TbQueuePro
     public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) {
         boolean result = storage.put(tpi.getTopic(), msg);
         if (result) {
-            callback.onSuccess(null);
+            if (callback != null) {
+                callback.onSuccess(null);
+            }
         } else {
-            Exception e = new RuntimeException("Failure add msg to InMemoryQueue");
+            if (callback != null) {
-            callback.onFailure(e);
+                callback.onFailure(new RuntimeException("Failure add msg to InMemoryQueue"));
+            }
         }
     }
 }

common/queue/src/main/java/org/thingsboard/server/provider/InMemoryMonolithQueueProvider.java

@@ -21,7 +21,11 @@ import org.springframework.stereotype.Component;
 import org.thingsboard.server.TbQueueConsumer;
 import org.thingsboard.server.TbQueueCoreSettings;
 import org.thingsboard.server.TbQueueProducer;
+import org.thingsboard.server.TbQueueRuleEngineSettings;
+import org.thingsboard.server.TbQueueTransportApiSettings;
+import org.thingsboard.server.TbQueueTransportNotificationSettings;
 import org.thingsboard.server.common.TbProtoQueueMsg;
+import org.thingsboard.server.discovery.PartitionChangeEvent;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
@@ -36,50 +40,52 @@ import org.thingsboard.server.memory.InMemoryTbQueueProducer;
 public class InMemoryMonolithQueueProvider implements TbCoreQueueProvider, TbRuleEngineQueueProvider {
 
     private final TbQueueCoreSettings coreSettings;
+    private final TbQueueRuleEngineSettings ruleEngineSettings;
+    private final TbQueueTransportApiSettings transportApiSettings;
+    private final TbQueueTransportNotificationSettings notificationSettings;
 
-    public InMemoryMonolithQueueProvider(TbQueueCoreSettings coreSettings) {
+    public InMemoryMonolithQueueProvider(TbQueueCoreSettings coreSettings,
+                                         TbQueueRuleEngineSettings ruleEngineSettings,
+                                         TbQueueTransportApiSettings transportApiSettings,
+                                         TbQueueTransportNotificationSettings notificationSettings) {
         this.coreSettings = coreSettings;
+        this.ruleEngineSettings = ruleEngineSettings;
+        this.transportApiSettings = transportApiSettings;
+        this.notificationSettings = notificationSettings;
     }
 
     @Override
     public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportMsgProducer() {
-        InMemoryTbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic());
+        return new InMemoryTbQueueProducer<>(notificationSettings.getNotificationsTopic());
-        return producer;
     }
 
     @Override
     public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() {
-        InMemoryTbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic());
+        return new InMemoryTbQueueProducer<>(ruleEngineSettings.getTopic());
-        return producer;
     }
 
     @Override
     public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() {
-        InMemoryTbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic());
+        return new InMemoryTbQueueProducer<>(coreSettings.getTopic());
-        return producer;
     }
 
     @Override
     public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> getToRuleEngineMsgConsumer() {
-        InMemoryTbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer = new InMemoryTbQueueConsumer<>(coreSettings.getTopic());
+        return new InMemoryTbQueueConsumer<>(ruleEngineSettings.getTopic());
-        return consumer;
     }
 
     @Override
     public TbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> getToCoreMsgConsumer() {
-        InMemoryTbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> consumer = new InMemoryTbQueueConsumer<>(coreSettings.getTopic());
+        return new InMemoryTbQueueConsumer<>(coreSettings.getTopic());
-        return consumer;
     }
 
     @Override
     public TbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> getTransportApiRequestConsumer() {
-        InMemoryTbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> consumer = new InMemoryTbQueueConsumer<>(coreSettings.getTopic());
+        return new InMemoryTbQueueConsumer<>(transportApiSettings.getRequestsTopic());
-        return consumer;
     }
 
     @Override
     public TbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> getTransportApiResponseProducer() {
-        InMemoryTbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic());
+        return new InMemoryTbQueueProducer<>(transportApiSettings.getResponsesTopic());
-        return producer;
     }
 }

common/queue/src/main/java/org/thingsboard/server/provider/InMemoryTransportQueueProvider.java

@@ -15,13 +15,19 @@
  */
 package org.thingsboard.server.provider;
 
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Component;
+import org.thingsboard.server.TbQueueAdmin;
 import org.thingsboard.server.TbQueueConsumer;
+import org.thingsboard.server.TbQueueCoreSettings;
 import org.thingsboard.server.TbQueueProducer;
 import org.thingsboard.server.TbQueueRequestTemplate;
+import org.thingsboard.server.TbQueueRuleEngineSettings;
 import org.thingsboard.server.TbQueueTransportApiSettings;
+import org.thingsboard.server.TbQueueTransportNotificationSettings;
 import org.thingsboard.server.common.DefaultTbQueueRequestTemplate;
 import org.thingsboard.server.common.TbProtoQueueMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
@@ -37,20 +43,29 @@ import org.thingsboard.server.memory.InMemoryTbQueueProducer;
 @Slf4j
 public class InMemoryTransportQueueProvider implements TransportQueueProvider {
 
+    private final TbQueueCoreSettings coreSettings;
+    private final TbQueueRuleEngineSettings ruleEngineSettings;
     private final TbQueueTransportApiSettings transportApiSettings;
+    private final TbQueueTransportNotificationSettings notificationSettings;
 
-    public InMemoryTransportQueueProvider(TbQueueTransportApiSettings transportApiSettings) {
+    public InMemoryTransportQueueProvider(TbQueueCoreSettings coreSettings,
+                                          TbQueueRuleEngineSettings ruleEngineSettings,
+                                          TbQueueTransportApiSettings transportApiSettings,
+                                          TbQueueTransportNotificationSettings notificationSettings) {
+        this.coreSettings = coreSettings;
+        this.ruleEngineSettings = ruleEngineSettings;
         this.transportApiSettings = transportApiSettings;
+        this.notificationSettings = notificationSettings;
     }
 
     @Override
     public TbQueueRequestTemplate<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> getTransportApiRequestTemplate() {
         InMemoryTbQueueProducer<TbProtoQueueMsg<TransportApiRequestMsg>> producer = new InMemoryTbQueueProducer<>(transportApiSettings.getRequestsTopic());
-
         InMemoryTbQueueConsumer<TbProtoQueueMsg<TransportApiResponseMsg>> consumer = new InMemoryTbQueueConsumer<>(transportApiSettings.getResponsesTopic());
 
         DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder
                 <TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> templateBuilder = DefaultTbQueueRequestTemplate.builder();
+        templateBuilder.queueAdmin(topic -> Futures.immediateFuture(null));
         templateBuilder.requestTemplate(producer);
         templateBuilder.responseTemplate(consumer);
         templateBuilder.maxPendingRequests(transportApiSettings.getMaxPendingRequests());
@@ -61,19 +76,16 @@ public class InMemoryTransportQueueProvider implements TransportQueueProvider {
 
     @Override
     public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() {
-        InMemoryTbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> producer = new InMemoryTbQueueProducer<>(transportApiSettings.getRequestsTopic());
+        return new InMemoryTbQueueProducer<>(ruleEngineSettings.getTopic());
-        return producer;
     }
 
     @Override
     public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() {
-        InMemoryTbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> producer = new InMemoryTbQueueProducer<>(transportApiSettings.getRequestsTopic());
+        return new InMemoryTbQueueProducer<>(coreSettings.getTopic());
-        return producer;
     }
 
     @Override
     public TbQueueConsumer<TbProtoQueueMsg<ToTransportMsg>> getTransportNotificationsConsumer() {
-        InMemoryTbQueueConsumer<TbProtoQueueMsg<ToTransportMsg>> consumer = new InMemoryTbQueueConsumer<>(transportApiSettings.getResponsesTopic());
+        return new InMemoryTbQueueConsumer<>(notificationSettings.getNotificationsTopic());
-        return consumer;
     }
 }

common/queue/src/main/java/org/thingsboard/server/provider/KafkaMonolithQueueProvider.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -21,6 +21,8 @@ import org.thingsboard.server.TbQueueConsumer;
 import org.thingsboard.server.TbQueueCoreSettings;
 import org.thingsboard.server.TbQueueProducer;
 import org.thingsboard.server.common.TbProtoQueueMsg;
+import org.thingsboard.server.discovery.PartitionChangeEvent;
+import org.thingsboard.server.discovery.ServiceType;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;

common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbCoreQueueProvider.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -21,6 +21,8 @@ import org.thingsboard.server.TbQueueConsumer;
 import org.thingsboard.server.TbQueueCoreSettings;
 import org.thingsboard.server.TbQueueProducer;
 import org.thingsboard.server.common.TbProtoQueueMsg;
+import org.thingsboard.server.discovery.PartitionChangeEvent;
+import org.thingsboard.server.discovery.ServiceType;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;

common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbRuleEngineQueueProvider.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

common/queue/src/main/java/org/thingsboard/server/provider/KafkaTransportQueueProvider.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

common/queue/src/main/java/org/thingsboard/server/provider/TbCoreQueueProvider.java

@@ -15,9 +15,11 @@
  */
 package org.thingsboard.server.provider;
 
+import org.springframework.context.ApplicationListener;
 import org.thingsboard.server.TbQueueConsumer;
 import org.thingsboard.server.TbQueueProducer;
 import org.thingsboard.server.common.TbProtoQueueMsg;
+import org.thingsboard.server.discovery.PartitionChangeEvent;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;

common/queue/src/main/java/org/thingsboard/server/provider/TbRuleEngineQueueProvider.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -78,13 +78,12 @@ public class DefaultTransportService implements TransportService {
     private long sessionInactivityTimeout;
     @Value("${transport.sessions.report_timeout}")
     private long sessionReportTimeout;
+    @Value("${queue.transport.poll_interval}")
+    private int notificationsPollDuration;
 
     private final TransportQueueProvider queueProvider;
     private final PartitionService partitionService;
 
-    @Value("${kafka.notifications.poll_interval}")
-    private int notificationsPollDuration;
-
     protected TbQueueRequestTemplate<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> transportApiRequestTemplate;
     protected TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> ruleEngineMsgProducer;
     protected TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> tbCoreMsgProducer;
@@ -120,7 +119,7 @@ public class DefaultTransportService implements TransportService {
         ruleEngineMsgProducer = queueProvider.getRuleEngineMsgProducer();
         tbCoreMsgProducer = queueProvider.getTbCoreMsgProducer();
         transportNotificationsConsumer = queueProvider.getTransportNotificationsConsumer();
-
+        transportApiRequestTemplate.init();
         mainConsumerExecutor.execute(() -> {
             while (!stopped) {
                 try {
@@ -163,6 +162,9 @@ public class DefaultTransportService implements TransportService {
         if (mainConsumerExecutor != null) {
             mainConsumerExecutor.shutdownNow();
         }
+        if (transportApiRequestTemplate != null) {
+            transportApiRequestTemplate.stop();
+        }
     }
 
     @Override