Fixed tests

@@ -42,21 +42,26 @@ import org.thingsboard.server.common.msg.queue.RuleEngineException;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.dao.model.ModelConstants;
 import org.thingsboard.server.dao.tenant.TenantService;
+import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
 import scala.concurrent.duration.Duration;
 
+import java.util.HashSet;
 import java.util.Optional;
+import java.util.Set;
 
 public class AppActor extends ContextAwareActor {
 
     private static final TenantId SYSTEM_TENANT = new TenantId(ModelConstants.NULL_UUID);
     private final TenantService tenantService;
     private final BiMap<TenantId, ActorRef> tenantActors;
+    private final Set<TenantId> deletedTenants;
     private boolean ruleChainsInitialized;
 
     private AppActor(ActorSystemContext systemContext) {
         super(systemContext);
         this.tenantService = systemContext.getTenantService();
         this.tenantActors = HashBiMap.create();
+        this.deletedTenants = new HashSet<>();
     }
 
     @Override
@@ -139,7 +144,11 @@ public class AppActor extends ContextAwareActor {
         if (SYSTEM_TENANT.equals(msg.getTenantId())) {
             msg.getTbMsg().getCallback().onFailure(new RuleEngineException("Message has system tenant id!"));
         } else {
-            getOrCreateTenantActor(msg.getTenantId()).tell(msg, self());
+            if (!deletedTenants.contains(msg.getTenantId())) {
+                getOrCreateTenantActor(msg.getTenantId()).tell(msg, self());
+            } else {
+                msg.getTbMsg().getCallback().onSuccess();
+            }
         }
     }
 
@@ -154,8 +163,10 @@ public class AppActor extends ContextAwareActor {
         } else {
             if (msg.getEntityId().getEntityType() == EntityType.TENANT
                     && msg.getEvent() == ComponentLifecycleEvent.DELETED) {
-                log.debug("[{}] Handling tenant deleted notification: {}", msg.getTenantId(), msg);
-                ActorRef tenantActor = tenantActors.remove(new TenantId(msg.getEntityId().getId()));
+                log.info("[{}] Handling tenant deleted notification: {}", msg.getTenantId(), msg);
+                TenantId tenantId = new TenantId(msg.getEntityId().getId());
+                deletedTenants.add(tenantId);
+                ActorRef tenantActor = tenantActors.get(tenantId);
                 if (tenantActor != null) {
                     log.debug("[{}] Deleting tenant actor: {}", msg.getTenantId(), tenantActor);
                     context().stop(tenantActor);
@@ -172,16 +183,22 @@ public class AppActor extends ContextAwareActor {
     }
 
     private void onToDeviceActorMsg(TenantAwareMsg msg) {
-        getOrCreateTenantActor(msg.getTenantId()).tell(msg, ActorRef.noSender());
+        if (!deletedTenants.contains(msg.getTenantId())) {
+            getOrCreateTenantActor(msg.getTenantId()).tell(msg, ActorRef.noSender());
+        } else {
+            if (msg instanceof TransportToDeviceActorMsgWrapper) {
+                ((TransportToDeviceActorMsgWrapper) msg).getCallback().onSuccess();
+            }
+        }
     }
 
     private ActorRef getOrCreateTenantActor(TenantId tenantId) {
         return tenantActors.computeIfAbsent(tenantId, k -> {
-            log.debug("[{}] Creating tenant actor.", tenantId);
+            log.info("[{}] Creating tenant actor.", tenantId);
             ActorRef tenantActor = context().actorOf(Props.create(new TenantActor.ActorCreator(systemContext, tenantId))
                     .withDispatcher(DefaultActorService.CORE_DISPATCHER_NAME), tenantId.toString());
             context().watch(tenantActor);
-            log.debug("[{}] Created tenant actor: {}.", tenantId, tenantActor);
+            log.info("[{}] Created tenant actor: {}.", tenantId, tenantActor);
             return tenantActor;
         });
     }

@@ -66,6 +66,8 @@ public class TenantActor extends RuleChainManagerActor {
         return strategy;
     }
 
+    boolean cantFindTenant = false;
+
     @Override
     public void preStart() {
         log.info("[{}] Starting tenant actor.", tenantId);
@@ -78,10 +80,14 @@ public class TenantActor extends RuleChainManagerActor {
             isCore = systemContext.getServiceInfoProvider().isService(ServiceType.TB_CORE);
 
             if (isRuleEngineForCurrentTenant) {
-                if (isolatedTenantId.map(id -> id.equals(tenantId)).orElseGet(() -> !tenant.isIsolatedTbRuleEngine())) {
-                    initRuleChains();
-                } else {
-                    isRuleEngineForCurrentTenant = false;
+                try {
+                    if (isolatedTenantId.map(id -> id.equals(tenantId)).orElseGet(() -> !tenant.isIsolatedTbRuleEngine())) {
+                        initRuleChains();
+                    } else {
+                        isRuleEngineForCurrentTenant = false;
+                    }
+                } catch (Exception e) {
+                    cantFindTenant = true;
                 }
             }
             log.info("[{}] Tenant actor started.", tenantId);
@@ -97,6 +103,9 @@ public class TenantActor extends RuleChainManagerActor {
 
     @Override
     protected boolean process(TbActorMsg msg) {
+        if (cantFindTenant) {
+            log.info("Missing Tenant msg: {}", msg);
+        }
         switch (msg.getMsgType()) {
             case PARTITION_CHANGE_MSG:
                 PartitionChangeMsg partitionChangeMsg = (PartitionChangeMsg) msg;

@@ -34,6 +34,7 @@ import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.EntityView;
 import org.thingsboard.server.common.data.HasName;
+import org.thingsboard.server.common.data.HasTenantId;
 import org.thingsboard.server.common.data.Tenant;
 import org.thingsboard.server.common.data.User;
 import org.thingsboard.server.common.data.alarm.Alarm;
@@ -667,7 +668,13 @@ public abstract class BaseController {
                     }
                 }
                 TbMsg tbMsg = TbMsg.newMsg(msgType, entityId, metaData, TbMsgDataType.JSON, json.writeValueAsString(entityNode));
-                tbClusterService.pushMsgToRuleEngine(user.getTenantId(), entityId, tbMsg, null);
+                TenantId tenantId = user.getTenantId();
+                if (tenantId.isNullUid()) {
+                    if (entity instanceof HasTenantId) {
+                        tenantId = ((HasTenantId) entity).getTenantId();
+                    }
+                }
+                tbClusterService.pushMsgToRuleEngine(tenantId, entityId, tbMsg, null);
             } catch (Exception e) {
                 log.warn("[{}] Failed to push entity action to rule engine: {}", entityId, actionType, e);
             }

@@ -325,6 +325,8 @@ public class DefaultDeviceStateService implements DeviceStateService {
                 });
             });
 
+            addedPartitions.forEach(tpi -> partitionedDevices.computeIfAbsent(tpi, key -> ConcurrentHashMap.newKeySet()));
+
             //TODO 3.0: replace this dummy search with new functionality to search by partitions using SQL capabilities.
             // Adding only devices that are in new partitions
             List<Tenant> tenants = tenantService.findTenants(new TextPageLink(Integer.MAX_VALUE)).getData();

@@ -27,6 +27,7 @@ import org.thingsboard.server.common.data.kv.JsonDataEntry;
 import org.thingsboard.server.common.data.kv.LongDataEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.dao.asset.AssetService;
+import org.thingsboard.server.dao.exception.DataValidationException;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.queue.util.TbRuleEngineComponent;
 import org.thingsboard.server.service.queue.TbRuleEngineConsumerStats;
@@ -76,13 +77,19 @@ public class DefaultRuleEngineStatisticsService implements RuleEngineStatisticsS
         String queueName = ruleEngineStats.getQueueName();
         ruleEngineStats.getTenantStats().forEach((id, stats) -> {
             TenantId tenantId = new TenantId(id);
-            AssetId serviceAssetId = getServiceAssetId(tenantId, queueName);
-            if (stats.getTotalMsgCounter().get() > 0) {
-                List<TsKvEntry> tsList = stats.getCounters().entrySet().stream()
-                        .map(kv -> new BasicTsKvEntry(ts, new LongDataEntry(kv.getKey(), (long) kv.getValue().get())))
-                        .collect(Collectors.toList());
-                if (!tsList.isEmpty()) {
-                    tsService.saveAndNotify(tenantId, serviceAssetId, tsList, CALLBACK);
+            try {
+                AssetId serviceAssetId = getServiceAssetId(tenantId, queueName);
+                if (stats.getTotalMsgCounter().get() > 0) {
+                    List<TsKvEntry> tsList = stats.getCounters().entrySet().stream()
+                            .map(kv -> new BasicTsKvEntry(ts, new LongDataEntry(kv.getKey(), (long) kv.getValue().get())))
+                            .collect(Collectors.toList());
+                    if (!tsList.isEmpty()) {
+                        tsService.saveAndNotify(tenantId, serviceAssetId, tsList, CALLBACK);
+                    }
+                }
+            } catch (DataValidationException e) {
+                if (!e.getMessage().equalsIgnoreCase("Asset is referencing to non-existent tenant!")) {
+                    throw e;
                 }
             }
         });

@@ -27,7 +27,6 @@
 
     <logger name="org.thingsboard.server" level="INFO" />
     <logger name="akka" level="INFO" />
-    <logger name="org.springframework.boot.autoconfigure.logging" level="DEBUG" />
 
 <!--    <logger name="org.thingsboard.server.service.queue" level="TRACE" />-->
 <!--    <logger name="org.thingsboard.server.service.transport" level="TRACE" />-->

@@ -31,7 +31,7 @@ server:
     key-store-type: "${SSL_KEY_STORE_TYPE:PKCS12}"
     # Alias that identifies the key in the key store
     key-alias: "${SSL_KEY_ALIAS:tomcat}"
-  log_controller_error_stack_trace: "${HTTP_LOG_CONTROLLER_ERROR_STACK_TRACE:true}"
+  log_controller_error_stack_trace: "${HTTP_LOG_CONTROLLER_ERROR_STACK_TRACE:false}"
   ws:
     send_timeout: "${TB_SERVER_WS_SEND_TIMEOUT:5000}"
     limits:
@@ -412,7 +412,7 @@ audit-log:
 state:
   defaultInactivityTimeoutInSec: "${DEFAULT_INACTIVITY_TIMEOUT:10}"
   defaultStateCheckIntervalInSec: "${DEFAULT_STATE_CHECK_INTERVAL:10}"
-  persistToTelemetry: "${PERSIST_STATE_TO_TELEMETRY:true}"
+  persistToTelemetry: "${PERSIST_STATE_TO_TELEMETRY:false}"
 
 js:
   evaluator: "${JS_EVALUATOR:local}" # local/remote
@@ -513,7 +513,7 @@ swagger:
   version: "${SWAGGER_VERSION:2.0}"
 
 queue:
-  type: "${TB_QUEUE_TYPE:in-memory}" # kafka or in-memory or aws-sqs or pubsub or service-bus
+  type: "${TB_QUEUE_TYPE:in-memory}" # in-memory or kafka (Apache Kafka) or aws-sqs (AWS SQS) or pubsub (PubSub) or service-bus (Azure Service Bus) or rabbitmq (RabbitMQ)
   kafka:
     bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}"
     acks: "${TB_KAFKA_ACKS:all}"
@@ -588,7 +588,7 @@ queue:
       enabled: "${TB_QUEUE_RULE_ENGINE_STATS_ENABLED:true}"
       print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}"
     queues:
-      - name: "Main"
+      - name: "${TB_QUEUE_RE_MAIN_QUEUE_NAME:Main}"
         topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb_rule_engine.main}"
         poll-interval: "${TB_QUEUE_RE_MAIN_POLL_INTERVAL_MS:25}"
         partitions: "${TB_QUEUE_RE_MAIN_PARTITIONS:10}"

@@ -33,6 +33,7 @@ import org.junit.rules.TestRule;
 import org.junit.rules.TestWatcher;
 import org.junit.runner.Description;
 import org.junit.runner.RunWith;
+import org.mockito.Mockito;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootContextLoader;
 import org.springframework.boot.test.context.SpringBootTest;
@@ -197,6 +198,7 @@ public abstract class AbstractControllerTest {
         createUserAndLogin(customerUser, CUSTOMER_USER_PASSWORD);
 
         logout();
+
         log.info("Executed setup");
     }
 

@@ -17,6 +17,7 @@ package org.thingsboard.server.controller;
 
 import com.datastax.driver.core.utils.UUIDs;
 import com.fasterxml.jackson.core.type.TypeReference;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
@@ -24,6 +25,7 @@ import org.eclipse.paho.client.mqttv3.MqttMessage;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.thingsboard.server.common.data.Customer;
 import org.thingsboard.server.common.data.Device;
@@ -46,6 +48,7 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import static org.hamcrest.Matchers.containsString;
 import static org.junit.Assert.assertEquals;
@@ -55,6 +58,7 @@ import static org.junit.Assert.assertTrue;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID;
 
+@Slf4j
 public abstract class BaseEntityViewControllerTest extends AbstractControllerTest {
 
     private IdComparator<EntityView> idComparator;
@@ -417,12 +421,22 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
         MqttConnectOptions options = new MqttConnectOptions();
         options.setUserName(accessToken);
         client.connect(options);
-        Thread.sleep(3000);
-
+        awaitConnected(client, TimeUnit.SECONDS.toMillis(30));
         MqttMessage message = new MqttMessage();
         message.setPayload(strKvs.getBytes());
         client.publish("v1/devices/me/telemetry", message);
         Thread.sleep(1000);
+//        client.disconnect();
+    }
+
+    private void awaitConnected(MqttAsyncClient client, long ms) throws InterruptedException {
+        long start = System.currentTimeMillis();
+        while (!client.isConnected()) {
+            Thread.sleep(100);
+            if (start + ms < System.currentTimeMillis()) {
+                throw new RuntimeException("Client is not connected!");
+            }
+        }
     }
 
     private Set<String> getTelemetryKeys(String type, String id) throws Exception {

@@ -80,7 +80,7 @@ public abstract class AbstractMqttTelemetryIntegrationTest extends AbstractContr
         String deviceId = savedDevice.getId().getId().toString();
 
         Thread.sleep(1000);
-        List<String> actualKeys = doGetAsync("/api/plugins/telemetry/DEVICE/" + deviceId +  "/keys/timeseries", List.class);
+        List<String> actualKeys = doGetAsync("/api/plugins/telemetry/DEVICE/" + deviceId + "/keys/timeseries", List.class);
         Set<String> actualKeySet = new HashSet<>(actualKeys);
 
         List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4");
@@ -88,7 +88,7 @@ public abstract class AbstractMqttTelemetryIntegrationTest extends AbstractContr
 
         assertEquals(expectedKeySet, actualKeySet);
 
-        String getTelemetryValuesUrl = "/api/plugins/telemetry/DEVICE/" + deviceId +  "/values/timeseries?keys=" + String.join(",", actualKeySet);
+        String getTelemetryValuesUrl = "/api/plugins/telemetry/DEVICE/" + deviceId + "/values/timeseries?keys=" + String.join(",", actualKeySet);
         Map<String, List<Map<String, String>>> values = doGetAsync(getTelemetryValuesUrl, Map.class);
 
         assertEquals("value1", values.get("key1").get(0).get("value"));
@@ -104,13 +104,17 @@ public abstract class AbstractMqttTelemetryIntegrationTest extends AbstractContr
 
         MqttConnectOptions options = new MqttConnectOptions();
         options.setUserName(accessToken);
-        client.connect(options).waitForCompletion(3000);
         CountDownLatch latch = new CountDownLatch(1);
         TestMqttCallback callback = new TestMqttCallback(client, latch);
         client.setCallback(callback);
+        client.connect(options).waitForCompletion(3000);
         client.subscribe("v1/devices/me/attributes", MqttQoS.AT_MOST_ONCE.value());
         String payload = "{\"key\":\"value\"}";
-        String result = doPostAsync("/api/plugins/telemetry/" + savedDevice.getId() + "/SHARED_SCOPE", payload, String.class, status().isOk());
+//        TODO 3.1: we need to acknowledge subscription only after it is processed by device actor and not when the message is pushed to queue.
+//        MqttClient -> SUB REQUEST -> Transport -> Kafka -> Device Actor (subscribed)
+//        MqttClient <- SUB_ACK <- Transport
+        Thread.sleep(1000);
+        doPostAsync("/api/plugins/telemetry/" + savedDevice.getId() + "/SHARED_SCOPE", payload, String.class, status().isOk());
         latch.await(10, TimeUnit.SECONDS);
         assertEquals(payload, callback.getPayload());
         assertEquals(MqttQoS.AT_MOST_ONCE.value(), callback.getQoS());
@@ -120,8 +124,8 @@ public abstract class AbstractMqttTelemetryIntegrationTest extends AbstractContr
 
         private final MqttAsyncClient client;
         private final CountDownLatch latch;
-        private Integer qoS;
-        private String payload;
+        private volatile Integer qoS;
+        private volatile String payload;
 
         String getPayload() {
             return payload;
@@ -138,6 +142,7 @@ public abstract class AbstractMqttTelemetryIntegrationTest extends AbstractContr
 
         @Override
         public void connectionLost(Throwable throwable) {
+            log.error("Client connection lost", throwable);
         }
 
         @Override

@@ -15,14 +15,17 @@
  */
 package org.thingsboard.server.rules.flow;
 
+import akka.actor.ActorRef;
 import com.datastax.driver.core.utils.UUIDs;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.Mockito;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.thingsboard.rule.engine.metadata.TbGetAttributesNodeConfiguration;
+import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.service.ActorService;
 import org.thingsboard.server.common.data.*;
 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
@@ -35,6 +38,8 @@ import org.thingsboard.server.common.data.security.Authority;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgDataType;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
+import org.thingsboard.server.common.msg.queue.TbMsgCallback;
 import org.thingsboard.server.controller.AbstractRuleEngineControllerTest;
 import org.thingsboard.server.dao.attributes.AttributesService;
 
@@ -55,7 +60,7 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
     protected User tenantAdmin;
 
     @Autowired
-    protected ActorService actorService;
+    protected ActorSystemContext actorSystem;
 
     @Autowired
     protected AttributesService attributesService;
@@ -142,15 +147,12 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
 
         Thread.sleep(1000);
 
+        TbMsgCallback tbMsgCallback = Mockito.mock(TbMsgCallback.class);
+        TbMsg tbMsg = TbMsg.newMsg("CUSTOM", device.getId(), new TbMsgMetaData(), "{}", tbMsgCallback);
+        QueueToRuleEngineMsg qMsg = new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg, null, null);
         // Pushing Message to the system
-        TbMsg tbMsg = TbMsg.newMsg(
-                "CUSTOM",
-                device.getId(),
-                new TbMsgMetaData(), TbMsgDataType.JSON, "{}");
-        //TODO 2.5
-//        actorService.onMsg(new SendToClusterMsg(device.getId(), new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg)));
-
-        Thread.sleep(3000);
+        actorSystem.tell(qMsg, ActorRef.noSender());
+        Mockito.verify(tbMsgCallback, Mockito.timeout(3000)).onSuccess();
 
         TimePageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
         List<Event> events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
@@ -257,17 +259,13 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
 
         Thread.sleep(1000);
 
+        TbMsgCallback tbMsgCallback = Mockito.mock(TbMsgCallback.class);
+        TbMsg tbMsg = TbMsg.newMsg("CUSTOM", device.getId(), new TbMsgMetaData(), "{}", tbMsgCallback);
+        QueueToRuleEngineMsg qMsg = new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg, null, null);
         // Pushing Message to the system
-        TbMsg tbMsg = TbMsg.newMsg(
-                "CUSTOM",
-                device.getId(),
-                new TbMsgMetaData(),
-                TbMsgDataType.JSON,
-                "{}");
-        //TODO 2.5
-//        actorService.onMsg(new SendToClusterMsg(device.getId(), new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg)));
-
-        Thread.sleep(3000);
+        actorSystem.tell(qMsg, ActorRef.noSender());
+
+        Mockito.verify(tbMsgCallback, Mockito.timeout(3000)).onSuccess();
 
         TimePageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), rootRuleChain.getFirstRuleNodeId(), 1000);
         List<Event> events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());

@@ -15,14 +15,17 @@
  */
 package org.thingsboard.server.rules.lifecycle;
 
+import akka.actor.ActorRef;
 import com.datastax.driver.core.utils.UUIDs;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.Mockito;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.thingsboard.rule.engine.metadata.TbGetAttributesNodeConfiguration;
+import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.service.ActorService;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.Device;
@@ -39,6 +42,8 @@ import org.thingsboard.server.common.data.security.Authority;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgDataType;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
+import org.thingsboard.server.common.msg.queue.TbMsgCallback;
 import org.thingsboard.server.controller.AbstractRuleEngineControllerTest;
 import org.thingsboard.server.dao.attributes.AttributesService;
 
@@ -58,7 +63,7 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac
     protected User tenantAdmin;
 
     @Autowired
-    protected ActorService actorService;
+    protected ActorSystemContext actorSystem;
 
     @Autowired
     protected AttributesService attributesService;
@@ -133,17 +138,13 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac
 
         Thread.sleep(1000);
 
+        TbMsgCallback tbMsgCallback = Mockito.mock(TbMsgCallback.class);
+        TbMsg tbMsg = TbMsg.newMsg("CUSTOM", device.getId(), new TbMsgMetaData(), "{}", tbMsgCallback);
+        QueueToRuleEngineMsg qMsg = new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg, null, null);
         // Pushing Message to the system
-        TbMsg tbMsg = TbMsg.newMsg(
-                "CUSTOM",
-                device.getId(),
-                new TbMsgMetaData(),
-                TbMsgDataType.JSON,
-                "{}");
-        //TODO 2.5
-//        actorService.onMsg(new SendToClusterMsg(device.getId(), new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg)));
-
-        Thread.sleep(3000);
+        actorSystem.tell(qMsg, ActorRef.noSender());
+        Mockito.verify(tbMsgCallback, Mockito.timeout(3000)).onSuccess();
+
 
         TimePageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
         List<Event> events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());

@@ -7,7 +7,7 @@
         </encoder>
     </appender>
 
-    <logger name="org.thingsboard.server" level="WARN"/>
+    <logger name="org.thingsboard.server" level="INFO"/>
     <logger name="org.springframework" level="WARN"/>
     <logger name="org.springframework.boot.test" level="WARN"/>
     <logger name="org.apache.cassandra" level="WARN"/>

@@ -65,6 +65,10 @@ public final class TbMsg implements Serializable {
         return new TbMsg(UUID.randomUUID(), type, originator, metaData.copy(), dataType, data, ruleChainId, ruleNodeId, TbMsgCallback.EMPTY);
     }
 
+    public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data, TbMsgCallback callback) {
+        return new TbMsg(UUID.randomUUID(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, null, null, callback);
+    }
+
     public static 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.getRuleChainId(), origMsg.getRuleNodeId(), origMsg.getCallback());

@@ -95,14 +95,13 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
                         continue;
                     }
                     responses.forEach(response -> {
-                        log.trace("Received response to Queue Template request: {}", response);
                         byte[] requestIdHeader = response.getHeaders().get(REQUEST_ID_HEADER);
                         UUID requestId;
                         if (requestIdHeader == null) {
                             log.error("[{}] Missing requestId in header and body", response);
                         } else {
                             requestId = bytesToUuid(requestIdHeader);
-                            log.trace("[{}] Response received", requestId);
+                            log.trace("[{}] Response received: {}", requestId, response);
                             ResponseMetaData<Response> expectedResponse = pendingRequests.remove(requestId);
                             if (expectedResponse == null) {
                                 log.trace("[{}] Invalid or stale request", requestId);

@@ -41,7 +41,7 @@ public class TBKafkaAdmin implements TbQueueAdmin {
         client = AdminClient.create(settings.toProps());
     }
 
-    //TODO 2.5
+    //TODO 2.5 - ybondarenko Need to pass not only settings but also properties for topic creation. Somewhere in thingsboard.yml, in KV format.
     @Override
     public void createTopicIfNotExists(String topic) {
         try {
@@ -57,29 +57,6 @@ public class TBKafkaAdmin implements TbQueueAdmin {
             log.warn("[{}] Failed to create topic", topic, e);
             throw new RuntimeException(e);
         }
-//
-//        KafkaFuture<TopicDescription> topicDescriptionFuture = client.describeTopics(Collections.singleton(topic)).values().get(topic);
-//
-//        ListenableFuture<TopicDescription> topicFuture = JdkFutureAdapters.listenInPoolThread(topicDescriptionFuture);
-//
-//        return Futures.transformAsync(topicFuture, topicDescription -> {
-//            KafkaFuture<Void> resultFuture = createTopic(new NewTopic(topic, 1, (short) 1)).values().get(topic);
-//            return JdkFutureAdapters.listenInPoolThread(resultFuture);
-//        });
-    }
-
-    public void waitForTopic(String topic, long timeout, TimeUnit timeoutUnit) throws InterruptedException, TimeoutException {
-        synchronized (this) {
-            long timeoutExpiredMs = System.currentTimeMillis() + timeoutUnit.toMillis(timeout);
-            while (!topicExists(topic)) {
-                long waitMs = timeoutExpiredMs - System.currentTimeMillis();
-                if (waitMs <= 0) {
-                    throw new TimeoutException("Timeout occurred while waiting for topic [" + topic + "] to be available!");
-                } else {
-                    wait(1000);
-                }
-            }
-        }
     }
 
     public CreateTopicsResult createTopic(NewTopic topic) {

@@ -60,16 +60,13 @@ public final class InMemoryStorage {
                 if (first != null) {
                     entities = new ArrayList<>();
                     entities.add(first);
-                } else {
-                    entities = Collections.emptyList();
                     List<TbQueueMsg> otherList = new ArrayList<>();
-                    storage.get(topic).drainTo(otherList, 100);
+                    storage.get(topic).drainTo(otherList, 999);
                     for (TbQueueMsg other : otherList) {
                         entities.add((T) other);
                     }
-                }
-                if (entities.size() > 0) {
-                    storage.computeIfAbsent(topic, (t) -> new LinkedBlockingQueue<>()).addAll(entities);
+                } else {
+                    entities = Collections.emptyList();
                 }
                 return entities;
             } catch (InterruptedException e) {

@@ -36,11 +36,12 @@ public class TbQueueRuleEngineSettings {
     private String topic;
     private List<TbRuleEngineQueueConfiguration> queues;
 
-    //TODO 2.5 ybondarenko: make sure the queue names are valid to all queue providers. See how ther are used in TbRuleEngineQueueFactory.createToRuleEngineMsgConsumer and all producers
+    //TODO 2.5 ybondarenko: make sure the queue names are valid to all queue providers.
+    // See how they are used in TbRuleEngineQueueFactory.createToRuleEngineMsgConsumer and all producers
     @PostConstruct
     public void validate() {
         queues.stream().filter(queue -> queue.getName().equals("Main")).findFirst().orElseThrow(() -> {
-            log.warn("Main queue is not configured in thingsboard.yml");
+            log.error("Main queue is not configured in thingsboard.yml");
             return new RuntimeException("No \"Main\" queue configured!");
         });
     }

@@ -400,11 +400,11 @@ message ToRuleEngineNotificationMsg {
 
 /* Messages that are handled by ThingsBoard Transport Service */
 message ToTransportMsg {
-    int64 sessionIdMSB = 1;
-    int64 sessionIdLSB = 2;
-    SessionCloseNotificationProto sessionCloseNotification = 3;
-    GetAttributeResponseMsg getAttributesResponse = 4;
-    AttributeUpdateNotificationMsg attributeUpdateNotification = 5;
-    ToDeviceRpcRequestMsg toDeviceRequest = 6;
-    ToServerRpcResponseMsg toServerResponse = 7;
+  int64 sessionIdMSB = 1;
+  int64 sessionIdLSB = 2;
+  SessionCloseNotificationProto sessionCloseNotification = 3;
+  GetAttributeResponseMsg getAttributesResponse = 4;
+  AttributeUpdateNotificationMsg attributeUpdateNotification = 5;
+  ToDeviceRpcRequestMsg toDeviceRequest = 6;
+  ToServerRpcResponseMsg toServerResponse = 7;
 }

@@ -60,3 +60,4 @@ cassandra.query.tenant_rate_limits.enabled=false
 cassandra.query.tenant_rate_limits.configuration=5000:1,100000:60
 cassandra.query.tenant_rate_limits.print_tenant_names=false
 
+service.type=monolith
\ No newline at end of file

@@ -16,6 +16,8 @@ spring.datasource.url=jdbc:hsqldb:file:/tmp/testDb;sql.enforce_size=false
 spring.datasource.driverClassName=org.hsqldb.jdbc.JDBCDriver
 spring.datasource.hikari.maximumPoolSize = 50
 
+service.type=monolith
+
 #database.ts.type=timescale
 #database.ts.type=sql
 #database.entities.type=sql

@@ -25,6 +25,7 @@ import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
 import org.mockito.stubbing.Answer;
@@ -47,6 +48,7 @@ import org.thingsboard.server.dao.alarm.AlarmService;
 import javax.script.ScriptException;
 import java.io.IOException;
 import java.util.concurrent.Callable;
+import java.util.function.Consumer;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotSame;
@@ -82,6 +84,11 @@ public class TbAlarmNodeTest {
     @Mock
     private ScriptEngine detailsJs;
 
+    @Captor
+    private ArgumentCaptor<Runnable> successCaptor;
+    @Captor
+    private ArgumentCaptor<Consumer<Throwable>> failureCaptor;
+
     private RuleChainId ruleChainId = new RuleChainId(UUIDs.timeBased());
     private RuleNodeId ruleNodeId = new RuleNodeId(UUIDs.timeBased());
 
@@ -119,11 +126,12 @@ public class TbAlarmNodeTest {
 
         when(detailsJs.executeJsonAsync(msg)).thenReturn(Futures.immediateFuture(null));
         when(alarmService.findLatestByOriginatorAndType(tenantId, originator, "SomeType")).thenReturn(Futures.immediateFuture(null));
-
         doAnswer((Answer<Alarm>) invocationOnMock -> (Alarm) (invocationOnMock.getArguments())[0]).when(alarmService).createOrUpdateAlarm(any(Alarm.class));
 
         node.onMsg(ctx, msg);
 
+        verify(ctx).enqueue(any(), successCaptor.capture(), failureCaptor.capture());
+        successCaptor.getValue().run();
         verify(ctx).tellNext(any(), eq("Created"));
 
         ArgumentCaptor<TbMsg> msgCaptor = ArgumentCaptor.forClass(TbMsg.class);
@@ -191,6 +199,8 @@ public class TbAlarmNodeTest {
 
         node.onMsg(ctx, msg);
 
+        verify(ctx).enqueue(any(), successCaptor.capture(), failureCaptor.capture());
+        successCaptor.getValue().run();
         verify(ctx).tellNext(any(), eq("Created"));
 
         ArgumentCaptor<TbMsg> msgCaptor = ArgumentCaptor.forClass(TbMsg.class);

@@ -256,7 +256,7 @@ public class TbGetCustomerAttributeNodeTest {
                 .thenReturn(Futures.immediateFuture(timeseries));
 
         node.onMsg(ctx, msg);
-        verify(ctx).tellNext(msg, SUCCESS);
+        verify(ctx).tellSuccess(msg);
         assertEquals(msg.getMetaData().getValue("tempo"), "highest");
     }
 
@@ -268,7 +268,7 @@ public class TbGetCustomerAttributeNodeTest {
                 .thenReturn(Futures.immediateFuture(attributes));
 
         node.onMsg(ctx, msg);
-        verify(ctx).tellNext(msg, SUCCESS);
+        verify(ctx).tellSuccess(msg);
         assertEquals(msg.getMetaData().getValue("tempo"), "high");
     }
 }
\ No newline at end of file

@@ -77,7 +77,7 @@ public class TbTransformMsgNodeTest {
         node.onMsg(ctx, msg);
         verify(ctx).getDbCallbackExecutor();
         ArgumentCaptor<TbMsg> captor = ArgumentCaptor.forClass(TbMsg.class);
-        verify(ctx).tellNext(captor.capture(), eq(SUCCESS));
+        verify(ctx).tellSuccess(captor.capture());
         TbMsg actualMsg = captor.getValue();
         assertEquals(transformedMsg, actualMsg);
     }