Fixed tests

application/src/main/java/org/thingsboard/server/service/stats/DefaultRuleEngineStatisticsService.java

@@ -47,6 +47,7 @@ import java.util.stream.Collectors;
 @Slf4j
 public class DefaultRuleEngineStatisticsService implements RuleEngineStatisticsService {
 
+    public static final String TB_SERVICE_QUEUE = "TbServiceQueue";
     public static final FutureCallback<Void> CALLBACK = new FutureCallback<Void>() {
         @Override
         public void onSuccess(@Nullable Void result) {
@@ -95,7 +96,13 @@ public class DefaultRuleEngineStatisticsService implements RuleEngineStatisticsS
         });
         ruleEngineStats.getTenantExceptions().forEach((tenantId, e) -> {
             TsKvEntry tsKv = new BasicTsKvEntry(ts, new JsonDataEntry("ruleEngineException", e.toJsonString()));
+            try {
                 tsService.saveAndNotify(tenantId, getServiceAssetId(tenantId, queueName), Collections.singletonList(tsKv), CALLBACK);
+            } catch (DataValidationException e2) {
+                if (!e2.getMessage().equalsIgnoreCase("Asset is referencing to non-existent tenant!")) {
+                    throw e2;
+                }
+            }
         });
         ruleEngineStats.reset();
     }
@@ -113,7 +120,7 @@ public class DefaultRuleEngineStatisticsService implements RuleEngineStatisticsS
                         asset = new Asset();
                         asset.setTenantId(tenantId);
                         asset.setName(queueName + "_" + serviceInfoProvider.getServiceId());
-                        asset.setType("TbServiceQueue");
+                        asset.setType(TB_SERVICE_QUEUE);
                         asset = assetService.saveAsset(asset);
                     }
                     assetId = asset.getId();

application/src/test/java/org/thingsboard/server/controller/BaseAssetControllerTest.java

@@ -32,6 +32,7 @@ import org.thingsboard.server.common.data.page.TextPageData;
 import org.thingsboard.server.common.data.page.TextPageLink;
 import org.thingsboard.server.common.data.security.Authority;
 import org.thingsboard.server.dao.model.ModelConstants;
+import org.thingsboard.server.service.stats.DefaultRuleEngineStatisticsService;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -130,7 +131,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
             assets.add(doPost("/api/asset", asset, Asset.class));
         }
         List<EntitySubtype> assetTypes = doGetTyped("/api/asset/types",
-                new TypeReference<List<EntitySubtype>>(){});
+                new TypeReference<List<EntitySubtype>>() {
+                });
 
         Assert.assertNotNull(assetTypes);
         Assert.assertEquals(3, assetTypes.size());
@@ -262,13 +264,16 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
         TextPageData<Asset> pageData = null;
         do {
             pageData = doGetTypedWithPageLink("/api/tenant/assets?",
-                    new TypeReference<TextPageData<Asset>>(){}, pageLink);
+                    new TypeReference<TextPageData<Asset>>() {
+                    }, pageLink);
             loadedAssets.addAll(pageData.getData());
             if (pageData.hasNext()) {
                 pageLink = pageData.getNextPageLink();
             }
         } while (pageData.hasNext());
 
+        loadedAssets.removeIf(asset -> asset.getType().equals(DefaultRuleEngineStatisticsService.TB_SERVICE_QUEUE));
+
         Collections.sort(assets, idComparator);
         Collections.sort(loadedAssets, idComparator);
 
@@ -305,7 +310,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
         TextPageData<Asset> pageData = null;
         do {
             pageData = doGetTypedWithPageLink("/api/tenant/assets?",
-                    new TypeReference<TextPageData<Asset>>(){}, pageLink);
+                    new TypeReference<TextPageData<Asset>>() {
+                    }, pageLink);
             loadedAssetsTitle1.addAll(pageData.getData());
             if (pageData.hasNext()) {
                 pageLink = pageData.getNextPageLink();
@@ -321,7 +327,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
         pageLink = new TextPageLink(4, title2);
         do {
             pageData = doGetTypedWithPageLink("/api/tenant/assets?",
-                    new TypeReference<TextPageData<Asset>>(){}, pageLink);
+                    new TypeReference<TextPageData<Asset>>() {
+                    }, pageLink);
             loadedAssetsTitle2.addAll(pageData.getData());
             if (pageData.hasNext()) {
                 pageLink = pageData.getNextPageLink();
@@ -340,7 +347,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
 
         pageLink = new TextPageLink(4, title1);
         pageData = doGetTypedWithPageLink("/api/tenant/assets?",
-                new TypeReference<TextPageData<Asset>>(){}, pageLink);
+                new TypeReference<TextPageData<Asset>>() {
+                }, pageLink);
         Assert.assertFalse(pageData.hasNext());
         Assert.assertEquals(0, pageData.getData().size());
 
@@ -351,7 +359,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
 
         pageLink = new TextPageLink(4, title2);
         pageData = doGetTypedWithPageLink("/api/tenant/assets?",
-                new TypeReference<TextPageData<Asset>>(){}, pageLink);
+                new TypeReference<TextPageData<Asset>>() {
+                }, pageLink);
         Assert.assertFalse(pageData.hasNext());
         Assert.assertEquals(0, pageData.getData().size());
     }
@@ -388,7 +397,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
         TextPageData<Asset> pageData = null;
         do {
             pageData = doGetTypedWithPageLink("/api/tenant/assets?type={type}&",
-                    new TypeReference<TextPageData<Asset>>(){}, pageLink, type1);
+                    new TypeReference<TextPageData<Asset>>() {
+                    }, pageLink, type1);
             loadedAssetsType1.addAll(pageData.getData());
             if (pageData.hasNext()) {
                 pageLink = pageData.getNextPageLink();
@@ -404,7 +414,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
         pageLink = new TextPageLink(4);
         do {
             pageData = doGetTypedWithPageLink("/api/tenant/assets?type={type}&",
-                    new TypeReference<TextPageData<Asset>>(){}, pageLink, type2);
+                    new TypeReference<TextPageData<Asset>>() {
+                    }, pageLink, type2);
             loadedAssetsType2.addAll(pageData.getData());
             if (pageData.hasNext()) {
                 pageLink = pageData.getNextPageLink();
@@ -423,7 +434,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
 
         pageLink = new TextPageLink(4);
         pageData = doGetTypedWithPageLink("/api/tenant/assets?type={type}&",
-                new TypeReference<TextPageData<Asset>>(){}, pageLink, type1);
+                new TypeReference<TextPageData<Asset>>() {
+                }, pageLink, type1);
         Assert.assertFalse(pageData.hasNext());
         Assert.assertEquals(0, pageData.getData().size());
 
@@ -434,7 +446,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
 
         pageLink = new TextPageLink(4);
         pageData = doGetTypedWithPageLink("/api/tenant/assets?type={type}&",
-                new TypeReference<TextPageData<Asset>>(){}, pageLink, type2);
+                new TypeReference<TextPageData<Asset>>() {
+                }, pageLink, type2);
         Assert.assertFalse(pageData.hasNext());
         Assert.assertEquals(0, pageData.getData().size());
     }
@@ -461,7 +474,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
         TextPageData<Asset> pageData = null;
         do {
             pageData = doGetTypedWithPageLink("/api/customer/" + customerId.getId().toString() + "/assets?",
-                    new TypeReference<TextPageData<Asset>>(){}, pageLink);
+                    new TypeReference<TextPageData<Asset>>() {
+                    }, pageLink);
             loadedAssets.addAll(pageData.getData());
             if (pageData.hasNext()) {
                 pageLink = pageData.getNextPageLink();
@@ -513,7 +527,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
         TextPageData<Asset> pageData = null;
         do {
             pageData = doGetTypedWithPageLink("/api/customer/" + customerId.getId().toString() + "/assets?",
-                    new TypeReference<TextPageData<Asset>>(){}, pageLink);
+                    new TypeReference<TextPageData<Asset>>() {
+                    }, pageLink);
             loadedAssetsTitle1.addAll(pageData.getData());
             if (pageData.hasNext()) {
                 pageLink = pageData.getNextPageLink();
@@ -529,7 +544,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
         pageLink = new TextPageLink(4, title2);
         do {
             pageData = doGetTypedWithPageLink("/api/customer/" + customerId.getId().toString() + "/assets?",
-                    new TypeReference<TextPageData<Asset>>(){}, pageLink);
+                    new TypeReference<TextPageData<Asset>>() {
+                    }, pageLink);
             loadedAssetsTitle2.addAll(pageData.getData());
             if (pageData.hasNext()) {
                 pageLink = pageData.getNextPageLink();
@@ -548,7 +564,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
 
         pageLink = new TextPageLink(4, title1);
         pageData = doGetTypedWithPageLink("/api/customer/" + customerId.getId().toString() + "/assets?",
-                new TypeReference<TextPageData<Asset>>(){}, pageLink);
+                new TypeReference<TextPageData<Asset>>() {
+                }, pageLink);
         Assert.assertFalse(pageData.hasNext());
         Assert.assertEquals(0, pageData.getData().size());
 
@@ -559,7 +576,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
 
         pageLink = new TextPageLink(4, title2);
         pageData = doGetTypedWithPageLink("/api/customer/" + customerId.getId().toString() + "/assets?",
-                new TypeReference<TextPageData<Asset>>(){}, pageLink);
+                new TypeReference<TextPageData<Asset>>() {
+                }, pageLink);
         Assert.assertFalse(pageData.hasNext());
         Assert.assertEquals(0, pageData.getData().size());
     }
@@ -605,7 +623,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
         TextPageData<Asset> pageData = null;
         do {
             pageData = doGetTypedWithPageLink("/api/customer/" + customerId.getId().toString() + "/assets?type={type}&",
-                    new TypeReference<TextPageData<Asset>>(){}, pageLink, type1);
+                    new TypeReference<TextPageData<Asset>>() {
+                    }, pageLink, type1);
             loadedAssetsType1.addAll(pageData.getData());
             if (pageData.hasNext()) {
                 pageLink = pageData.getNextPageLink();
@@ -621,7 +640,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
         pageLink = new TextPageLink(4);
         do {
             pageData = doGetTypedWithPageLink("/api/customer/" + customerId.getId().toString() + "/assets?type={type}&",
-                    new TypeReference<TextPageData<Asset>>(){}, pageLink, type2);
+                    new TypeReference<TextPageData<Asset>>() {
+                    }, pageLink, type2);
             loadedAssetsType2.addAll(pageData.getData());
             if (pageData.hasNext()) {
                 pageLink = pageData.getNextPageLink();
@@ -640,7 +660,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
 
         pageLink = new TextPageLink(4);
         pageData = doGetTypedWithPageLink("/api/customer/" + customerId.getId().toString() + "/assets?type={type}&",
-                new TypeReference<TextPageData<Asset>>(){}, pageLink, type1);
+                new TypeReference<TextPageData<Asset>>() {
+                }, pageLink, type1);
         Assert.assertFalse(pageData.hasNext());
         Assert.assertEquals(0, pageData.getData().size());
 
@@ -651,7 +672,8 @@ public abstract class BaseAssetControllerTest extends AbstractControllerTest {
 
         pageLink = new TextPageLink(4);
         pageData = doGetTypedWithPageLink("/api/customer/" + customerId.getId().toString() + "/assets?type={type}&",
-                new TypeReference<TextPageData<Asset>>(){}, pageLink, type2);
+                new TypeReference<TextPageData<Asset>>() {
+                }, pageLink, type2);
         Assert.assertFalse(pageData.hasNext());
         Assert.assertEquals(0, pageData.getData().size());
     }

application/src/test/java/org/thingsboard/server/controller/BaseEntityViewControllerTest.java

@@ -426,7 +426,7 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
         message.setPayload(strKvs.getBytes());
         client.publish("v1/devices/me/telemetry", message);
         Thread.sleep(1000);
-//        client.disconnect();
+        client.disconnect();
     }
 
     private void awaitConnected(MqttAsyncClient client, long ms) throws InterruptedException {
@@ -463,13 +463,13 @@ 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((stringKV).getBytes());
         client.publish("v1/devices/me/attributes", message);
         Thread.sleep(1000);
-
+        client.disconnect();
         return new HashSet<>(doGetAsync("/api/plugins/telemetry/DEVICE/" + viewDeviceId + "/keys/attributes", List.class));
     }
 

application/src/test/java/org/thingsboard/server/controller/ControllerNoSqlTestSuite.java

@@ -16,10 +16,12 @@
 package org.thingsboard.server.controller;
 
 import org.cassandraunit.dataset.cql.ClassPathCQLDataSet;
+import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.extensions.cpsuite.ClasspathSuite;
 import org.junit.runner.RunWith;
 import org.thingsboard.server.dao.CustomCassandraCQLUnit;
+import org.thingsboard.server.queue.memory.InMemoryStorage;
 
 import java.util.Arrays;
 
@@ -37,4 +39,9 @@ public class ControllerNoSqlTestSuite {
                             new ClassPathCQLDataSet("cassandra/system-data.cql", false, false),
                             new ClassPathCQLDataSet("cassandra/system-test.cql", false, false)),
                     "cassandra-test.yaml", 30000l);
+
+    @BeforeClass
+    public static void cleanupInMemStorage(){
+        InMemoryStorage.getInstance().cleanup();
+    }
 }

application/src/test/java/org/thingsboard/server/controller/ControllerSqlTestSuite.java

@@ -15,10 +15,12 @@
  */
 package org.thingsboard.server.controller;
 
+import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.extensions.cpsuite.ClasspathSuite;
 import org.junit.runner.RunWith;
 import org.thingsboard.server.dao.CustomSqlUnit;
+import org.thingsboard.server.queue.memory.InMemoryStorage;
 
 import java.util.Arrays;
 
@@ -33,4 +35,9 @@ public class ControllerSqlTestSuite {
             Arrays.asList("sql/schema-ts-hsql.sql", "sql/schema-entities-hsql.sql", "sql/schema-entities-idx.sql", "sql/system-data.sql"),
             "sql/drop-all-tables.sql",
             "sql-test.properties");
+
+    @BeforeClass
+    public static void cleanupInMemStorage(){
+        InMemoryStorage.getInstance().cleanup();
+    }
 }

application/src/test/java/org/thingsboard/server/mqtt/MqttNoSqlTestSuite.java

@@ -16,10 +16,12 @@
 package org.thingsboard.server.mqtt;
 
 import org.cassandraunit.dataset.cql.ClassPathCQLDataSet;
+import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.extensions.cpsuite.ClasspathSuite;
 import org.junit.runner.RunWith;
 import org.thingsboard.server.dao.CustomCassandraCQLUnit;
+import org.thingsboard.server.queue.memory.InMemoryStorage;
 
 import java.util.Arrays;
 
@@ -36,4 +38,9 @@ public class MqttNoSqlTestSuite {
                             new ClassPathCQLDataSet("cassandra/schema-entities.cql", false, false),
                             new ClassPathCQLDataSet("cassandra/system-data.cql", false, false)),
                     "cassandra-test.yaml", 30000l);
+
+    @BeforeClass
+    public static void cleanupInMemStorage(){
+        InMemoryStorage.getInstance().cleanup();
+    }
 }

application/src/test/java/org/thingsboard/server/mqtt/MqttSqlTestSuite.java

@@ -15,10 +15,12 @@
  */
 package org.thingsboard.server.mqtt;
 
+import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.extensions.cpsuite.ClasspathSuite;
 import org.junit.runner.RunWith;
 import org.thingsboard.server.dao.CustomSqlUnit;
+import org.thingsboard.server.queue.memory.InMemoryStorage;
 
 import java.util.Arrays;
 
@@ -32,4 +34,9 @@ public class MqttSqlTestSuite {
             Arrays.asList("sql/schema-ts-hsql.sql", "sql/schema-entities-hsql.sql", "sql/system-data.sql"),
             "sql/drop-all-tables.sql",
             "sql-test.properties");
+
+    @BeforeClass
+    public static void cleanupInMemStorage(){
+        InMemoryStorage.getInstance().cleanup();
+    }
 }

application/src/test/java/org/thingsboard/server/rules/RuleEngineNoSqlTestSuite.java

@@ -16,11 +16,13 @@
 package org.thingsboard.server.rules;
 
 import org.cassandraunit.dataset.cql.ClassPathCQLDataSet;
+import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.extensions.cpsuite.ClasspathSuite;
 import org.junit.runner.RunWith;
 import org.thingsboard.server.dao.CustomCassandraCQLUnit;
 import org.thingsboard.server.dao.CustomSqlUnit;
+import org.thingsboard.server.queue.memory.InMemoryStorage;
 
 import java.util.Arrays;
 
@@ -40,4 +42,9 @@ public class RuleEngineNoSqlTestSuite {
                             new ClassPathCQLDataSet("cassandra/system-data.cql", false, false)),
                     "cassandra-test.yaml", 30000l);
 
+    @BeforeClass
+    public static void cleanupInMemStorage(){
+        InMemoryStorage.getInstance().cleanup();
+    }
+
 }

application/src/test/java/org/thingsboard/server/rules/RuleEngineSqlTestSuite.java

@@ -15,10 +15,12 @@
  */
 package org.thingsboard.server.rules;
 
+import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.extensions.cpsuite.ClasspathSuite;
 import org.junit.runner.RunWith;
 import org.thingsboard.server.dao.CustomSqlUnit;
+import org.thingsboard.server.queue.memory.InMemoryStorage;
 
 import java.util.Arrays;
 
@@ -33,4 +35,9 @@ public class RuleEngineSqlTestSuite {
             Arrays.asList("sql/schema-ts-hsql.sql", "sql/schema-entities-hsql.sql", "sql/system-data.sql"),
             "sql/drop-all-tables.sql",
             "sql-test.properties");
+
+    @BeforeClass
+    public static void cleanupInMemStorage(){
+        InMemoryStorage.getInstance().cleanup();
+    }
 }

application/src/test/java/org/thingsboard/server/system/SystemNoSqlTestSuite.java

@@ -16,10 +16,12 @@
 package org.thingsboard.server.system;
 
 import org.cassandraunit.dataset.cql.ClassPathCQLDataSet;
+import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.extensions.cpsuite.ClasspathSuite;
 import org.junit.runner.RunWith;
 import org.thingsboard.server.dao.CustomCassandraCQLUnit;
+import org.thingsboard.server.queue.memory.InMemoryStorage;
 
 import java.util.Arrays;
 
@@ -38,4 +40,9 @@ public class SystemNoSqlTestSuite {
                             new ClassPathCQLDataSet("cassandra/schema-entities.cql", false, false),
                             new ClassPathCQLDataSet("cassandra/system-data.cql", false, false)),
                     "cassandra-test.yaml", 30000l);
+
+    @BeforeClass
+    public static void cleanupInMemStorage(){
+        InMemoryStorage.getInstance().cleanup();
+    }
 }

application/src/test/java/org/thingsboard/server/system/SystemSqlTestSuite.java

@@ -15,10 +15,12 @@
  */
 package org.thingsboard.server.system;
 
+import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.extensions.cpsuite.ClasspathSuite;
 import org.junit.runner.RunWith;
 import org.thingsboard.server.dao.CustomSqlUnit;
+import org.thingsboard.server.queue.memory.InMemoryStorage;
 
 import java.util.Arrays;
 
@@ -35,4 +37,9 @@ public class SystemSqlTestSuite {
             "sql/drop-all-tables.sql",
             "sql-test.properties");
 
+    @BeforeClass
+    public static void cleanupInMemStorage(){
+        InMemoryStorage.getInstance().cleanup();
+    }
+
 }

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

@@ -30,11 +30,9 @@ import java.util.concurrent.TimeUnit;
 public final class InMemoryStorage {
     private static InMemoryStorage instance;
     private final ConcurrentHashMap<String, BlockingQueue<TbQueueMsg>> storage;
-    private volatile boolean stopped;
 
     private InMemoryStorage() {
         storage = new ConcurrentHashMap<>();
-        stopped = false;
     }
 
     public static InMemoryStorage getInstance() {
@@ -52,9 +50,8 @@ public final class InMemoryStorage {
         return storage.computeIfAbsent(topic, (t) -> new LinkedBlockingQueue<>()).add(msg);
     }
 
-    public <T extends TbQueueMsg> List<T> get(String topic, long durationInMillis) {
+    public <T extends TbQueueMsg> List<T> get(String topic, long durationInMillis) throws InterruptedException {
         if (storage.containsKey(topic)) {
-            try {
             List<T> entities;
             T first = (T) storage.get(topic).poll(durationInMillis, TimeUnit.MILLISECONDS);
             if (first != null) {
@@ -69,16 +66,15 @@ public final class InMemoryStorage {
                 entities = Collections.emptyList();
             }
             return entities;
-            } catch (InterruptedException e) {
-                if (!stopped) {
-                    log.warn("Queue was interrupted", e);
-                }
-            }
         }
         return Collections.emptyList();
     }
 
-    public void stop() {
+    /**
-        stopped = true;
+     * Used primarily for testing.
+     */
+    public void cleanup() {
+        storage.clear();
     }
+
 }

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

@@ -66,7 +66,16 @@ public class InMemoryTbQueueConsumer<T extends TbQueueMsg> implements TbQueueCon
         if (subscribed) {
             List<T> messages = partitions
                     .stream()
-                    .map(tpi -> storage.get(tpi.getFullTopicName(), durationInMillis))
+                    .map(tpi -> {
+                        try {
+                            return storage.get(tpi.getFullTopicName(), durationInMillis);
+                        } catch (InterruptedException e) {
+                            if (!stopped) {
+                                log.error("Queue was interrupted.", e);
+                            }
+                            return Collections.emptyList();
+                        }
+                    })
                     .flatMap(List::stream)
                     .map(msg -> (T) msg).collect(Collectors.toList());
             if (messages.size() > 0) {

tools/pom.xml

@@ -54,7 +54,7 @@
         <dependency>
             <groupId>org.apache.cassandra</groupId>
             <artifactId>cassandra-all</artifactId>
-            <version>3.11.4</version>
+            <version>3.11.6</version>
         </dependency>
         <dependency>
             <groupId>com.datastax.cassandra</groupId>