From 33c3d1daf431e3ced775b7ab198305ef053e1a60 Mon Sep 17 00:00:00 2001
From: jacklicn <jacklicn@outlook.com>
Date: Wed, 16 May 2018 11:56:52 +0800
Subject: [PATCH 1/9] Add missed import statements!

---
 ui/src/app/widget/widget-editor.controller.js | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/ui/src/app/widget/widget-editor.controller.js b/ui/src/app/widget/widget-editor.controller.js
index 301fc5ae28..6f0ad8761e 100644
--- a/ui/src/app/widget/widget-editor.controller.js
+++ b/ui/src/app/widget/widget-editor.controller.js
@@ -20,6 +20,7 @@ import 'brace/mode/javascript';
 import 'brace/mode/html';
 import 'brace/mode/css';
 import 'brace/mode/json';
+import 'ace-builds/src-min-noconflict/ace';
 import 'ace-builds/src-min-noconflict/snippets/javascript';
 import 'ace-builds/src-min-noconflict/snippets/text';
 import 'ace-builds/src-min-noconflict/snippets/html';
@@ -662,4 +663,4 @@ export default function WidgetEditorController(widgetService, userService, types
 
 }
 
-/* eslint-enable angular/angularelement */
\ No newline at end of file
+/* eslint-enable angular/angularelement */

From b9a988f33a2927860f742a967195778d9b22cc23 Mon Sep 17 00:00:00 2001
From: Volodymyr Babak <volodymyr.babak@gmail.com>
Date: Wed, 16 May 2018 18:12:31 +0300
Subject: [PATCH 2/9] Fix for correct install if env var changed

---
 docker/tb/run-application.sh | 11 +++++------
 1 file changed, 5 insertions(+), 6 deletions(-)

diff --git a/docker/tb/run-application.sh b/docker/tb/run-application.sh
index e297335fe2..a2a1e2beba 100755
--- a/docker/tb/run-application.sh
+++ b/docker/tb/run-application.sh
@@ -18,6 +18,11 @@
 
 dpkg -i /thingsboard.deb
 
+# Copying env variables into conf files
+printenv | awk -F "=" '{print "export " $1 "='\''" $2 "'\''"}' >> /usr/share/thingsboard/conf/thingsboard.conf
+
+cat /usr/share/thingsboard/conf/thingsboard.conf
+
 if [ "$DATABASE_TYPE" == "cassandra" ]; then
     until nmap $CASSANDRA_HOST -p $CASSANDRA_PORT | grep "$CASSANDRA_PORT/tcp open\|filtered"
     do
@@ -46,12 +51,6 @@ if [ "$ADD_SCHEMA_AND_SYSTEM_DATA" == "true" ]; then
     fi
 fi
 
-
-# Copying env variables into conf files
-printenv | awk -F "=" '{print "export " $1 "='\''" $2 "'\''"}' >> /usr/share/thingsboard/conf/thingsboard.conf
-
-cat /usr/share/thingsboard/conf/thingsboard.conf
-
 echo "Starting 'Thingsboard' service..."
 service thingsboard start
 

From 96a77fc2ead25ac3502e415fd6788f37897a7f56 Mon Sep 17 00:00:00 2001
From: liuyuan <405653510@qq.com>
Date: Thu, 17 May 2018 18:27:26 +0800
Subject: [PATCH 3/9] MqttTransportService  NioEventLoopGroup shutdown order
 error

---
 .../thingsboard/server/transport/mqtt/MqttTransportService.java | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java b/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java
index f0129e1b6c..cbe3ba6a13 100644
--- a/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java
+++ b/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java
@@ -119,8 +119,8 @@ public class MqttTransportService {
         try {
             serverChannel.close().sync();
         } finally {
-            bossGroup.shutdownGracefully();
             workerGroup.shutdownGracefully();
+            bossGroup.shutdownGracefully();
         }
         log.info("MQTT transport stopped!");
     }

From 4591d168445a61e1e86023beabb41a07516a7280 Mon Sep 17 00:00:00 2001
From: jacklicn <jacklicn@outlook.com>
Date: Mon, 21 May 2018 12:42:36 +0800
Subject: [PATCH 4/9] Add Chinese translation for audit-log.

---
 ui/src/app/locale/locale.constant-zh.js | 32 +++++++++++++++++++++++++
 1 file changed, 32 insertions(+)

diff --git a/ui/src/app/locale/locale.constant-zh.js b/ui/src/app/locale/locale.constant-zh.js
index d94141cb04..c25291adc0 100644
--- a/ui/src/app/locale/locale.constant-zh.js
+++ b/ui/src/app/locale/locale.constant-zh.js
@@ -280,6 +280,38 @@ export default function addLocaleChinese(locales) {
             "selected-attributes": "{ count, select, 1 {1 属性} other {# 属性} } 被选中",
             "selected-telemetry": "{ count, select, 1 {1 遥测} other {# 遥测} } 被选中"
         },
+        "audit-log": {
+            "audit": "审计",
+            "audit-logs": "审计日志",
+            "timestamp": "时间戳",
+            "entity-type": "实体类型",
+            "entity-name": "实体名称",
+            "user": "用户",
+            "type": "类型",
+            "status": "状态",
+            "details": "详情",
+            "type-added": "添加",
+            "type-deleted": "删除",
+            "type-updated": "更新",
+            "type-attributes-updated": "更新属性",
+            "type-attributes-deleted": "删除属性",
+            "type-rpc-call": "RPC调用",
+            "type-credentials-updated": "更新凭证",
+            "type-assigned-to-customer": "分配给客户",
+            "type-unassigned-from-customer": "未分配给客户",
+            "type-activated": "激活",
+            "type-suspended": "暂停",
+            "type-credentials-read": "读取凭证",
+            "type-attributes-read": "读取属性",
+            "status-success": "成功",
+            "status-failure": "失败",
+            "audit-log-details": "审计日志详情",
+            "no-audit-logs-prompt": "找不到日志",
+            "action-data": "活动数据",
+            "failure-details": "失败详情",
+            "search": "查找审计日志",
+            "clear-search": "清空查找"
+        },
         "confirm-on-exit": {
             "message": "您有未保存的更改。确定要离开此页吗？",
             "html-message": "您有未保存的更改。<br/> 确定要离开此页面吗？",

From 55127ae61c1ba7040ede0fafc447327449fde961 Mon Sep 17 00:00:00 2001
From: Andrew Shvayka <ashvayka@thingsboard.io>
Date: Wed, 23 May 2018 14:30:32 +0300
Subject: [PATCH 5/9] Fixed integration tests

---
 .../RuleChainActorMessageProcessor.java       | 34 +++++++++++-----
 .../AbstractRuleEngineControllerTest.java     |  6 +++
 ...AbstractRuleEngineFlowIntegrationTest.java | 40 +++++++++++--------
 ...actRuleEngineLifecycleIntegrationTest.java | 20 ++++++----
 .../metadata/TbAbstractGetAttributesNode.java | 25 +++++++++---
 .../engine/telemetry/TbMsgTimeseriesNode.java |  6 +--
 6 files changed, 85 insertions(+), 46 deletions(-)

diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
index ac902a7147..cc73eaedac 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
@@ -1,12 +1,12 @@
 /**
  * Copyright © 2016-2018 The Thingsboard Authors
- *
+ * <p>
  * 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
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
  * 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.
@@ -183,23 +183,35 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
 
     void onServiceToRuleEngineMsg(ServiceToRuleEngineMsg envelope) {
         checkActive();
-        putToQueue(enrichWithRuleChainId(envelope.getTbMsg()), msg -> pushMsgToNode(firstNode, msg, ""));
+        if (firstNode != null) {
+            putToQueue(enrichWithRuleChainId(envelope.getTbMsg()), msg -> pushMsgToNode(firstNode, msg, ""));
+        }
     }
 
     void onDeviceActorToRuleEngineMsg(DeviceActorToRuleEngineMsg envelope) {
         checkActive();
-        putToQueue(enrichWithRuleChainId(envelope.getTbMsg()), msg -> {
-            pushMsgToNode(firstNode, msg, "");
-            envelope.getCallbackRef().tell(new RuleEngineQueuePutAckMsg(msg.getId()), self);
-        });
+        if (firstNode != null) {
+            putToQueue(enrichWithRuleChainId(envelope.getTbMsg()), msg -> {
+                pushMsgToNode(firstNode, msg, "");
+                envelope.getCallbackRef().tell(new RuleEngineQueuePutAckMsg(msg.getId()), self);
+            });
+        }
     }
 
     void onRuleChainToRuleChainMsg(RuleChainToRuleChainMsg envelope) {
         checkActive();
         if (envelope.isEnqueue()) {
-            putToQueue(enrichWithRuleChainId(envelope.getMsg()), msg -> pushMsgToNode(firstNode, msg, envelope.getFromRelationType()));
+            if (firstNode != null) {
+                putToQueue(enrichWithRuleChainId(envelope.getMsg()), msg -> pushMsgToNode(firstNode, msg, envelope.getFromRelationType()));
+            }
         } else {
-            pushMsgToNode(firstNode, envelope.getMsg(), envelope.getFromRelationType());
+            if (firstNode != null) {
+                pushMsgToNode(firstNode, envelope.getMsg(), envelope.getFromRelationType());
+            } else {
+                TbMsg msg = envelope.getMsg();
+                EntityId ackId = msg.getRuleNodeId() != null ? msg.getRuleNodeId() : msg.getRuleChainId();
+                queue.ack(tenantId, envelope.getMsg(), ackId.getId(), msg.getClusterPartition());
+            }
         }
     }
 
diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java
index 6c2d3dbf7d..1b042a821c 100644
--- a/application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java
@@ -32,6 +32,7 @@ import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.service.queue.MsgQueueService;
 
 import java.io.IOException;
+import java.util.function.Predicate;
 
 /**
  * Created by ashvayka on 20.03.18.
@@ -75,4 +76,9 @@ public class AbstractRuleEngineControllerTest extends AbstractControllerTest {
             throw new RuntimeException(e);
         }
     }
+
+    protected Predicate<Event> filterByCustomEvent() {
+        return event -> event.getBody().get("msgType").textValue().equals("CUSTOM");
+    }
+
 }
diff --git a/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java
index 356dfeea41..c86d49609e 100644
--- a/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java
@@ -47,6 +47,8 @@ import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
 
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
@@ -157,15 +159,15 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
 
         Thread.sleep(3000);
 
-        TimePageData<Event> events = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
+        TimePageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
+        List<Event> events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
+        Assert.assertEquals(2, events.size());
 
-        Assert.assertEquals(2, events.getData().size());
-
-        Event inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
+        Event inEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
         Assert.assertEquals(ruleChain.getFirstRuleNodeId(), inEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText());
 
-        Event outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
+        Event outEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
         Assert.assertEquals(ruleChain.getFirstRuleNodeId(), outEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
 
@@ -174,15 +176,16 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
         RuleChain finalRuleChain = ruleChain;
         RuleNode lastRuleNode = metaData.getNodes().stream().filter(node -> !node.getId().equals(finalRuleChain.getFirstRuleNodeId())).findFirst().get();
 
-        events = getDebugEvents(savedTenant.getId(), lastRuleNode.getId(), 1000);
+        eventsPage = getDebugEvents(savedTenant.getId(), lastRuleNode.getId(), 1000);
+        events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
 
-        Assert.assertEquals(2, events.getData().size());
+        Assert.assertEquals(2, events.size());
 
-        inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
+        inEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
         Assert.assertEquals(lastRuleNode.getId(), inEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText());
 
-        outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
+        outEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
         Assert.assertEquals(lastRuleNode.getId(), outEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
 
@@ -274,15 +277,16 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
 
         Thread.sleep(3000);
 
-        TimePageData<Event> events = getDebugEvents(savedTenant.getId(), rootRuleChain.getFirstRuleNodeId(), 1000);
+        TimePageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), rootRuleChain.getFirstRuleNodeId(), 1000);
+        List<Event> events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
 
-        Assert.assertEquals(2, events.getData().size());
+        Assert.assertEquals(2, events.size());
 
-        Event inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
+        Event inEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
         Assert.assertEquals(rootRuleChain.getFirstRuleNodeId(), inEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText());
 
-        Event outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
+        Event outEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
         Assert.assertEquals(rootRuleChain.getFirstRuleNodeId(), outEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
 
@@ -291,15 +295,17 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
         RuleChain finalRuleChain = rootRuleChain;
         RuleNode lastRuleNode = secondaryMetaData.getNodes().stream().filter(node -> !node.getId().equals(finalRuleChain.getFirstRuleNodeId())).findFirst().get();
 
-        events = getDebugEvents(savedTenant.getId(), lastRuleNode.getId(), 1000);
+        eventsPage = getDebugEvents(savedTenant.getId(), lastRuleNode.getId(), 1000);
+        events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
+
 
-        Assert.assertEquals(2, events.getData().size());
+        Assert.assertEquals(2, events.size());
 
-        inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
+        inEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
         Assert.assertEquals(lastRuleNode.getId(), inEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText());
 
-        outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
+        outEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
         Assert.assertEquals(lastRuleNode.getId(), outEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
 
diff --git a/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java
index 2f25b97eee..7ac0789383 100644
--- a/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java
@@ -45,6 +45,8 @@ import org.thingsboard.server.dao.attributes.AttributesService;
 
 import java.io.IOException;
 import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
 
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
@@ -144,15 +146,16 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac
 
         Thread.sleep(3000);
 
-        TimePageData<Event> events = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
+        TimePageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
+        List<Event> events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
 
-        Assert.assertEquals(2, events.getData().size());
+        Assert.assertEquals(2, events.size());
 
-        Event inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
+        Event inEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
         Assert.assertEquals(ruleChain.getFirstRuleNodeId(), inEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText());
 
-        Event outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
+        Event outEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
         Assert.assertEquals(ruleChain.getFirstRuleNodeId(), outEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
 
@@ -212,15 +215,16 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac
 
         Thread.sleep(3000);
 
-        TimePageData<Event> events = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
+        TimePageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
+        List<Event> events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
 
-        Assert.assertEquals(2, events.getData().size());
+        Assert.assertEquals(2, events.size());
 
-        Event inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
+        Event inEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
         Assert.assertEquals(ruleChain.getFirstRuleNodeId(), inEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText());
 
-        Event outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
+        Event outEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
         Assert.assertEquals(ruleChain.getFirstRuleNodeId(), outEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
 
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java
index c6212cc045..65b2f1136e 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java
@@ -15,7 +15,6 @@
  */
 package org.thingsboard.rule.engine.metadata;
 
-import com.google.common.base.Function;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import org.apache.commons.collections.CollectionUtils;
@@ -33,9 +32,11 @@ import java.util.List;
 import static org.thingsboard.rule.engine.DonAsynchron.withCallback;
 import static org.thingsboard.rule.engine.api.TbRelationTypes.FAILURE;
 import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
-import static org.thingsboard.server.common.data.DataConstants.*;
+import static org.thingsboard.server.common.data.DataConstants.CLIENT_SCOPE;
+import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE;
+import static org.thingsboard.server.common.data.DataConstants.SHARED_SCOPE;
 
-public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeConfiguration, T extends EntityId>  implements TbNode {
+public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeConfiguration, T extends EntityId> implements TbNode {
 
     protected C config;
 
@@ -59,7 +60,7 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
     }
 
     private void safePutAttributes(TbContext ctx, TbMsg msg, T entityId) {
-        if(entityId == null || entityId.isNullUid()) {
+        if (entityId == null || entityId.isNullUid()) {
             ctx.tellNext(msg, FAILURE);
             return;
         }
@@ -78,7 +79,13 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
         }
         ListenableFuture<List<AttributeKvEntry>> latest = ctx.getAttributesService().find(entityId, scope, keys);
         return Futures.transform(latest, l -> {
-            l.forEach(r -> msg.getMetaData().putValue(prefix + r.getKey(), r.getValueAsString()));
+            l.forEach(r -> {
+                if (r.getValue() != null) {
+                    msg.getMetaData().putValue(prefix + r.getKey(), r.getValueAsString());
+                } else {
+                    throw new RuntimeException("[" + scope + "][" + r.getKey() + "] attribute value is not present in the DB!");
+                }
+            });
             return null;
         });
     }
@@ -89,7 +96,13 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
         }
         ListenableFuture<List<TsKvEntry>> latest = ctx.getTimeseriesService().findLatest(entityId, keys);
         return Futures.transform(latest, l -> {
-            l.forEach(r -> msg.getMetaData().putValue(r.getKey(), r.getValueAsString()));
+            l.forEach(r -> {
+                if (r.getValue() != null) {
+                    msg.getMetaData().putValue(r.getKey(), r.getValueAsString());
+                } else {
+                    throw new RuntimeException("[" + r.getKey() + "] telemetry value is not present in the DB!");
+                }
+            });
             return null;
         });
     }
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java
index ea4446dbe5..dc0fcdd08d 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java
@@ -69,10 +69,8 @@ public class TbMsgTimeseriesNode implements TbNode {
             try {
                 ts = Long.parseLong(tsStr);
             } catch (NumberFormatException e) {}
-        }
-        if (ts == -1) {
-            ctx.tellFailure(msg, new IllegalArgumentException("Msg metadata doesn't contain valid ts value: " + msg.getMetaData()));
-            return;
+        } else {
+            ts = System.currentTimeMillis();
         }
         String src = msg.getData();
         TelemetryUploadRequest telemetryUploadRequest = JsonConverter.convertToTelemetry(new JsonParser().parse(src), ts);

From 94db399367c62d3a3885c00c0564e45b4055f424 Mon Sep 17 00:00:00 2001
From: Igor Kulikov <ikulikov@thingsboard.io>
Date: Wed, 23 May 2018 17:06:51 +0300
Subject: [PATCH 6/9] New system parameters: default cassandra ts key/val ttl;
 allow system mail service for rules.

---
 .../server/actors/ActorSystemContext.java     |  4 ++++
 .../actors/ruleChain/DefaultTbContext.java    |  6 +++++-
 .../service/queue/DefaultMsgQueueService.java |  4 ++--
 .../src/main/resources/thingsboard.yml        | 20 +++++++++----------
 .../dao/queue/db/nosql/CassandraMsgQueue.java |  2 +-
 .../dao/queue/memory/InMemoryMsgQueue.java    |  2 +-
 .../CassandraBaseTimeseriesDao.java           | 16 +++++++++++++++
 .../test/resources/cassandra-test.properties  |  2 ++
 8 files changed, 41 insertions(+), 15 deletions(-)

diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
index f7c7f1a5a8..4840f2a0e8 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
@@ -235,6 +235,10 @@ public class ActorSystemContext {
     @Getter
     private boolean tenantComponentsInitEnabled;
 
+    @Value("${actors.rule.allow_system_mail_service}")
+    @Getter
+    private boolean allowSystemMailService;
+
     @Getter
     @Setter
     private ActorSystem actorSystem;
diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
index b888bc33e4..70509fb875 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
@@ -209,7 +209,11 @@ class DefaultTbContext implements TbContext {
 
     @Override
     public MailService getMailService() {
-        return mainCtx.getMailService();
+        if (mainCtx.isAllowSystemMailService()) {
+            return mainCtx.getMailService();
+        } else {
+            throw new RuntimeException("Access to System Mail Service is forbidden!");
+        }
     }
 
     @Override
diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultMsgQueueService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultMsgQueueService.java
index a4558eb0a4..927584789d 100644
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultMsgQueueService.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultMsgQueueService.java
@@ -40,10 +40,10 @@ import java.util.concurrent.atomic.AtomicLong;
 @Slf4j
 public class DefaultMsgQueueService implements MsgQueueService {
 
-    @Value("${rule.queue.max_size}")
+    @Value("${actors.rule.queue.max_size}")
     private long queueMaxSize;
 
-    @Value("${rule.queue.cleanup_period}")
+    @Value("${actors.rule.queue.cleanup_period}")
     private long queueCleanUpPeriod;
 
     @Autowired
diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml
index 9a1089549e..a10ef7d7c0 100644
--- a/application/src/main/resources/thingsboard.yml
+++ b/application/src/main/resources/thingsboard.yml
@@ -203,6 +203,7 @@ cassandra:
     default_fetch_size: "${CASSANDRA_DEFAULT_FETCH_SIZE:2000}"
     # Specify partitioning size for timestamp key-value storage. Example MINUTES, HOURS, DAYS, MONTHS
     ts_key_value_partitioning: "${TS_KV_PARTITIONING:MONTHS}"
+    ts_key_value_ttl: "${TS_KV_TTL:0}"
     buffer_size: "${CASSANDRA_QUERY_BUFFER_SIZE:200000}"
     concurrent_limit: "${CASSANDRA_QUERY_CONCURRENT_LIMIT:1000}"
     permit_max_wait_time: "${PERMIT_MAX_WAIT_TIME:120000}"
@@ -236,6 +237,8 @@ actors:
     js_thread_pool_size: "${ACTORS_RULE_JS_THREAD_POOL_SIZE:10}"
     # Specify thread pool size for mail sender executor service
     mail_thread_pool_size: "${ACTORS_RULE_MAIL_THREAD_POOL_SIZE:10}"
+    # Whether to allow usage of system mail service for rules
+    allow_system_mail_service: "${ACTORS_RULE_ALLOW_SYSTEM_MAIL_SERVICE:true}"
     # Specify thread pool size for external call service
     external_call_thread_pool_size: "${ACTORS_RULE_EXTERNAL_CALL_THREAD_POOL_SIZE:10}"
     js_sandbox:
@@ -253,6 +256,13 @@ actors:
     node:
       # Errors for particular actor are persisted once per specified amount of milliseconds
       error_persist_frequency: "${ACTORS_RULE_NODE_ERROR_FREQUENCY:3000}"
+    queue:
+      # Message queue type (memory or db)
+      type: "${ACTORS_RULE_QUEUE_TYPE:memory}"
+      # Message queue maximum size (per tenant)
+      max_size: "${ACTORS_RULE_QUEUE_MAX_SIZE:100}"
+      # Message queue cleanup period in seconds
+      cleanup_period: "${ACTORS_RULE_QUEUE_CLEANUP_PERIOD:3600}"
   statistics:
     # Enable/disable actor statistics
     enabled: "${ACTORS_STATISTICS_ENABLED:true}"
@@ -333,16 +343,6 @@ spring:
     username: "${SPRING_DATASOURCE_USERNAME:sa}"
     password: "${SPRING_DATASOURCE_PASSWORD:}"
 
-rule:
-  queue:
-    #Message queue type (memory or db)
-    type: "${RULE_QUEUE_TYPE:memory}"
-    #Message queue maximum size (per tenant)
-    max_size: "${RULE_QUEUE_MAX_SIZE:100}"
-    #Message queue cleanup period in seconds
-    cleanup_period: "${RULE_QUEUE_CLEANUP_PERIOD:3600}"
-
-
 # PostgreSQL DAO Configuration
 #spring:
 #  data:
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java
index ce481b78b8..9cc87b47b6 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java
@@ -36,7 +36,7 @@ import java.util.List;
 import java.util.UUID;
 
 @Component
-@ConditionalOnProperty(prefix = "rule.queue", value = "type", havingValue = "db")
+@ConditionalOnProperty(prefix = "actors.rule.queue", value = "type", havingValue = "db")
 @Slf4j
 @NoSqlDao
 public class CassandraMsgQueue implements MsgQueue {
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/memory/InMemoryMsgQueue.java b/dao/src/main/java/org/thingsboard/server/dao/queue/memory/InMemoryMsgQueue.java
index 4532e023ce..93057784f6 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/memory/InMemoryMsgQueue.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/queue/memory/InMemoryMsgQueue.java
@@ -40,7 +40,7 @@ import java.util.concurrent.Executors;
  * Created by ashvayka on 27.04.18.
  */
 @Component
-@ConditionalOnProperty(prefix = "rule.queue", value = "type", havingValue = "memory", matchIfMissing = true)
+@ConditionalOnProperty(prefix = "actors.rule.queue", value = "type", havingValue = "memory", matchIfMissing = true)
 @Slf4j
 public class InMemoryMsgQueue implements MsgQueue {
 
diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java
index 0fa9653ad2..7aa317cbf9 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java
@@ -82,6 +82,9 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
     @Value("${cassandra.query.ts_key_value_partitioning}")
     private String partitioning;
 
+    @Value("${cassandra.query.ts_key_value_ttl}")
+    private long systemTtl;
+
     private TsPartitionDate tsFormat;
 
     private PreparedStatement partitionInsertStmt;
@@ -287,6 +290,7 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
 
     @Override
     public ListenableFuture<Void> save(EntityId entityId, TsKvEntry tsKvEntry, long ttl) {
+        ttl = computeTtl(ttl);
         long partition = toPartitionTs(tsKvEntry.getTs());
         DataType type = tsKvEntry.getDataType();
         BoundStatement stmt = (ttl == 0 ? getSaveStmt(type) : getSaveTtlStmt(type)).bind();
@@ -304,6 +308,7 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
 
     @Override
     public ListenableFuture<Void> savePartition(EntityId entityId, long tsKvEntryTs, String key, long ttl) {
+        ttl = computeTtl(ttl);
         long partition = toPartitionTs(tsKvEntryTs);
         log.debug("Saving partition {} for the entity [{}-{}] and key {}", partition, entityId.getEntityType(), entityId.getId(), key);
         BoundStatement stmt = (ttl == 0 ? getPartitionInsertStmt() : getPartitionInsertTtlStmt()).bind();
@@ -317,6 +322,17 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
         return getFuture(executeAsyncWrite(stmt), rs -> null);
     }
 
+    private long computeTtl(long ttl) {
+        if (systemTtl > 0) {
+            if (ttl == 0) {
+                ttl = systemTtl;
+            } else {
+                ttl = Math.min(systemTtl, ttl);
+            }
+        }
+        return ttl;
+    }
+
     @Override
     public ListenableFuture<Void> saveLatest(EntityId entityId, TsKvEntry tsKvEntry) {
         BoundStatement stmt = getLatestStmt().bind()
diff --git a/dao/src/test/resources/cassandra-test.properties b/dao/src/test/resources/cassandra-test.properties
index 737687f053..cf07b22e42 100644
--- a/dao/src/test/resources/cassandra-test.properties
+++ b/dao/src/test/resources/cassandra-test.properties
@@ -46,6 +46,8 @@ cassandra.query.default_fetch_size=2000
 
 cassandra.query.ts_key_value_partitioning=HOURS
 
+cassandra.query.ts_key_value_ttl=0
+
 cassandra.query.max_limit_per_request=1000
 cassandra.query.buffer_size=100000
 cassandra.query.concurrent_limit=1000

From d94ae25b4e3b2bec5263872937e726394f6693bb Mon Sep 17 00:00:00 2001
From: Igor Kulikov <ikulikov@thingsboard.io>
Date: Wed, 23 May 2018 17:22:34 +0300
Subject: [PATCH 7/9] Fix license header

---
 .../actors/ruleChain/RuleChainActorMessageProcessor.java  | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
index cc73eaedac..7d560db061 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
@@ -1,12 +1,12 @@
 /**
  * Copyright © 2016-2018 The Thingsboard Authors
- * <p>
+ *
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *     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.

From d1552a13b469694a1d57a37c8b91ab51ad545536 Mon Sep 17 00:00:00 2001
From: Igor Kulikov <ikulikov@thingsboard.io>
Date: Thu, 24 May 2018 10:04:05 +0300
Subject: [PATCH 8/9] Move update scripts to ver.2.0.0

---
 .../src/main/data/upgrade/{1.5.0 => 2.0.0}/schema_update.cql    | 0
 .../src/main/data/upgrade/{1.5.0 => 2.0.0}/schema_update.sql    | 0
 .../thingsboard/server/install/ThingsboardInstallService.java   | 2 +-
 .../server/service/install/CassandraDatabaseUpgradeService.java | 2 +-
 .../server/service/install/DefaultDataUpdateService.java        | 2 +-
 .../server/service/install/SqlDatabaseUpgradeService.java       | 2 +-
 6 files changed, 4 insertions(+), 4 deletions(-)
 rename application/src/main/data/upgrade/{1.5.0 => 2.0.0}/schema_update.cql (100%)
 rename application/src/main/data/upgrade/{1.5.0 => 2.0.0}/schema_update.sql (100%)

diff --git a/application/src/main/data/upgrade/1.5.0/schema_update.cql b/application/src/main/data/upgrade/2.0.0/schema_update.cql
similarity index 100%
rename from application/src/main/data/upgrade/1.5.0/schema_update.cql
rename to application/src/main/data/upgrade/2.0.0/schema_update.cql
diff --git a/application/src/main/data/upgrade/1.5.0/schema_update.sql b/application/src/main/data/upgrade/2.0.0/schema_update.sql
similarity index 100%
rename from application/src/main/data/upgrade/1.5.0/schema_update.sql
rename to application/src/main/data/upgrade/2.0.0/schema_update.sql
diff --git a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
index d2ae77cc3d..f863d0bb2a 100644
--- a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
+++ b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
@@ -82,7 +82,7 @@ public class ThingsboardInstallService {
                         databaseUpgradeService.upgradeDatabase("1.3.1");
 
                     case "1.4.0":
-                        log.info("Upgrading ThingsBoard from version 1.4.0 to 1.5.0 ...");
+                        log.info("Upgrading ThingsBoard from version 1.4.0 to 2.0.0 ...");
 
                         databaseUpgradeService.upgradeDatabase("1.4.0");
 
diff --git a/application/src/main/java/org/thingsboard/server/service/install/CassandraDatabaseUpgradeService.java b/application/src/main/java/org/thingsboard/server/service/install/CassandraDatabaseUpgradeService.java
index d38cddd4a1..4d2adeaad1 100644
--- a/application/src/main/java/org/thingsboard/server/service/install/CassandraDatabaseUpgradeService.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/CassandraDatabaseUpgradeService.java
@@ -198,7 +198,7 @@ public class CassandraDatabaseUpgradeService implements DatabaseUpgradeService {
             case "1.4.0":
 
                 log.info("Updating schema ...");
-                schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "1.5.0", SCHEMA_UPDATE_CQL);
+                schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "2.0.0", SCHEMA_UPDATE_CQL);
                 loadCql(schemaUpdateFile);
                 log.info("Schema updated.");
 
diff --git a/application/src/main/java/org/thingsboard/server/service/install/DefaultDataUpdateService.java b/application/src/main/java/org/thingsboard/server/service/install/DefaultDataUpdateService.java
index 37e5b30d11..b3723682e2 100644
--- a/application/src/main/java/org/thingsboard/server/service/install/DefaultDataUpdateService.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/DefaultDataUpdateService.java
@@ -47,7 +47,7 @@ public class DefaultDataUpdateService implements DataUpdateService {
     public void updateData(String fromVersion) throws Exception {
         switch (fromVersion) {
             case "1.4.0":
-                log.info("Updating data from version 1.4.0 to 1.5.0 ...");
+                log.info("Updating data from version 1.4.0 to 2.0.0 ...");
                 tenantsDefaultRuleChainUpdater.updateEntities(null);
                 break;
             default:
diff --git a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
index ecb607084c..29d5c65be9 100644
--- a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
@@ -104,7 +104,7 @@ public class SqlDatabaseUpgradeService implements DatabaseUpgradeService {
             case "1.4.0":
                 try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
                     log.info("Updating schema ...");
-                    schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "1.5.0", SCHEMA_UPDATE_SQL);
+                    schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "2.0.0", SCHEMA_UPDATE_SQL);
                     String sql = new String(Files.readAllBytes(schemaUpdateFile), Charset.forName("UTF-8"));
                     conn.createStatement().execute(sql); //NOSONAR, ignoring because method used to execute thingsboard database upgrade script
                     log.info("Schema updated.");

From 02e2c14fc959156cd2a8b406f33c6af3e875c7f1 Mon Sep 17 00:00:00 2001
From: Igor Kulikov <ikulikov@thingsboard.io>
Date: Thu, 24 May 2018 10:18:06 +0300
Subject: [PATCH 9/9] Cleanup

---
 .../src/main/resources/thingsboard.yml        |  2 +-
 .../server/dao/queue/db/MsgAck.java           | 32 -------
 .../dao/queue/db/UnprocessedMsgFilter.java    | 35 --------
 .../dao/queue/db/nosql/CassandraMsgQueue.java | 87 -------------------
 .../dao/queue/db/nosql/QueuePartitioner.java  | 86 ------------------
 .../repository/CassandraAckRepository.java    | 68 ---------------
 .../repository/CassandraMsgRepository.java    | 67 --------------
 ...CassandraProcessedPartitionRepository.java | 64 --------------
 .../queue/db/repository/AckRepository.java    | 29 -------
 .../queue/db/repository/MsgRepository.java    | 30 -------
 .../ProcessedPartitionRepository.java         | 29 -------
 .../server/dao/queue/db/sql/SqlMsgQueue.java  | 20 -----
 .../queue/db/nosql/QueuePartitionerTest.java  | 81 -----------------
 .../db/nosql/UnprocessedMsgFilterTest.java    | 47 ----------
 .../CassandraAckRepositoryTest.java           | 82 -----------------
 .../CassandraMsgRepositoryTest.java           | 87 -------------------
 ...andraProcessedPartitionRepositoryTest.java | 83 ------------------
 17 files changed, 1 insertion(+), 928 deletions(-)
 delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/queue/db/MsgAck.java
 delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/queue/db/UnprocessedMsgFilter.java
 delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java
 delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/QueuePartitioner.java
 delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepository.java
 delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraMsgRepository.java
 delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraProcessedPartitionRepository.java
 delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/AckRepository.java
 delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/MsgRepository.java
 delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/ProcessedPartitionRepository.java
 delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/queue/db/sql/SqlMsgQueue.java
 delete mode 100644 dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/QueuePartitionerTest.java
 delete mode 100644 dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/UnprocessedMsgFilterTest.java
 delete mode 100644 dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepositoryTest.java
 delete mode 100644 dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraMsgRepositoryTest.java
 delete mode 100644 dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraProcessedPartitionRepositoryTest.java

diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml
index a10ef7d7c0..188291a4e2 100644
--- a/application/src/main/resources/thingsboard.yml
+++ b/application/src/main/resources/thingsboard.yml
@@ -257,7 +257,7 @@ actors:
       # Errors for particular actor are persisted once per specified amount of milliseconds
       error_persist_frequency: "${ACTORS_RULE_NODE_ERROR_FREQUENCY:3000}"
     queue:
-      # Message queue type (memory or db)
+      # Message queue type
       type: "${ACTORS_RULE_QUEUE_TYPE:memory}"
       # Message queue maximum size (per tenant)
       max_size: "${ACTORS_RULE_QUEUE_MAX_SIZE:100}"
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/MsgAck.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/MsgAck.java
deleted file mode 100644
index a1b039a1a7..0000000000
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/MsgAck.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db;
-
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-
-import java.util.UUID;
-
-@Data
-@EqualsAndHashCode
-public class MsgAck {
-
-    private final UUID msgId;
-    private final UUID nodeId;
-    private final long clusteredPartition;
-    private final long tsPartition;
-
-}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/UnprocessedMsgFilter.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/UnprocessedMsgFilter.java
deleted file mode 100644
index 66eaa6d46b..0000000000
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/UnprocessedMsgFilter.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db;
-
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.common.msg.TbMsg;
-import org.thingsboard.server.dao.queue.db.MsgAck;
-
-import java.util.Collection;
-import java.util.List;
-import java.util.Set;
-import java.util.UUID;
-import java.util.stream.Collectors;
-
-@Component
-public class UnprocessedMsgFilter {
-
-    public Collection<TbMsg> filter(List<TbMsg> msgs, List<MsgAck> acks) {
-        Set<UUID> processedIds = acks.stream().map(MsgAck::getMsgId).collect(Collectors.toSet());
-        return msgs.stream().filter(i -> !processedIds.contains(i.getId())).collect(Collectors.toList());
-    }
-}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java
deleted file mode 100644
index 9cc87b47b6..0000000000
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.nosql;
-
-import com.datastax.driver.core.utils.UUIDs;
-import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.common.msg.TbMsg;
-import org.thingsboard.server.dao.queue.MsgQueue;
-import org.thingsboard.server.dao.queue.db.MsgAck;
-import org.thingsboard.server.dao.queue.db.UnprocessedMsgFilter;
-import org.thingsboard.server.dao.queue.db.repository.AckRepository;
-import org.thingsboard.server.dao.queue.db.repository.MsgRepository;
-import org.thingsboard.server.dao.util.NoSqlDao;
-
-import java.util.List;
-import java.util.UUID;
-
-@Component
-@ConditionalOnProperty(prefix = "actors.rule.queue", value = "type", havingValue = "db")
-@Slf4j
-@NoSqlDao
-public class CassandraMsgQueue implements MsgQueue {
-
-    @Autowired
-    private MsgRepository msgRepository;
-    @Autowired
-    private AckRepository ackRepository;
-    @Autowired
-    private UnprocessedMsgFilter unprocessedMsgFilter;
-    @Autowired
-    private QueuePartitioner queuePartitioner;
-
-    @Override
-    public ListenableFuture<Void> put(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition) {
-        long msgTime = getMsgTime(msg);
-        long tsPartition = queuePartitioner.getPartition(msgTime);
-        return msgRepository.save(msg, nodeId, clusterPartition, tsPartition, msgTime);
-    }
-
-    @Override
-    public ListenableFuture<Void> ack(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition) {
-        long tsPartition = queuePartitioner.getPartition(getMsgTime(msg));
-        MsgAck ack = new MsgAck(msg.getId(), nodeId, clusterPartition, tsPartition);
-        return ackRepository.ack(ack);
-    }
-
-    @Override
-    public Iterable<TbMsg> findUnprocessed(TenantId tenantId, UUID nodeId, long clusterPartition) {
-        List<TbMsg> unprocessedMsgs = Lists.newArrayList();
-        for (Long tsPartition : queuePartitioner.findUnprocessedPartitions(nodeId, clusterPartition)) {
-            List<TbMsg> msgs = msgRepository.findMsgs(nodeId, clusterPartition, tsPartition);
-            List<MsgAck> acks = ackRepository.findAcks(nodeId, clusterPartition, tsPartition);
-            unprocessedMsgs.addAll(unprocessedMsgFilter.filter(msgs, acks));
-        }
-        return unprocessedMsgs;
-    }
-
-    @Override
-    public ListenableFuture<Void> cleanUp(TenantId tenantId) {
-        return Futures.immediateFuture(null);
-    }
-
-    private long getMsgTime(TbMsg msg) {
-        return UUIDs.unixTimestamp(msg.getId());
-    }
-
-}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/QueuePartitioner.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/QueuePartitioner.java
deleted file mode 100644
index 6076d93e9f..0000000000
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/QueuePartitioner.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.nosql;
-
-import com.google.common.collect.Lists;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.dao.queue.db.repository.ProcessedPartitionRepository;
-import org.thingsboard.server.dao.timeseries.TsPartitionDate;
-import org.thingsboard.server.dao.util.NoSqlDao;
-
-import java.time.Clock;
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZoneOffset;
-import java.util.List;
-import java.util.Optional;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-
-@Component
-@Slf4j
-@NoSqlDao
-public class QueuePartitioner {
-
-    private final TsPartitionDate tsFormat;
-    private ProcessedPartitionRepository processedPartitionRepository;
-    private Clock clock = Clock.systemUTC();
-
-    public QueuePartitioner(@Value("${cassandra.queue.partitioning}") String partitioning,
-                            ProcessedPartitionRepository processedPartitionRepository) {
-        this.processedPartitionRepository = processedPartitionRepository;
-        Optional<TsPartitionDate> partition = TsPartitionDate.parse(partitioning);
-        if (partition.isPresent()) {
-            tsFormat = partition.get();
-        } else {
-            log.warn("Incorrect configuration of partitioning {}", partitioning);
-            throw new RuntimeException("Failed to parse partitioning property: " + partitioning + "!");
-        }
-    }
-
-    public long getPartition(long ts) {
-        //TODO: use TsPartitionDate.truncateTo?
-        LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(ts), ZoneOffset.UTC);
-        return tsFormat.truncatedTo(time).toInstant(ZoneOffset.UTC).toEpochMilli();
-    }
-
-    public List<Long> findUnprocessedPartitions(UUID nodeId, long clusteredHash) {
-        Optional<Long> lastPartitionOption = processedPartitionRepository.findLastProcessedPartition(nodeId, clusteredHash);
-        long lastPartition = lastPartitionOption.orElse(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(7));
-        List<Long> unprocessedPartitions = Lists.newArrayList();
-
-        LocalDateTime current = LocalDateTime.ofInstant(Instant.ofEpochMilli(lastPartition), ZoneOffset.UTC);
-        LocalDateTime end = LocalDateTime.ofInstant(Instant.now(clock), ZoneOffset.UTC)
-                .plus(1L, tsFormat.getTruncateUnit());
-
-        while (current.isBefore(end)) {
-            current = current.plus(1L, tsFormat.getTruncateUnit());
-            unprocessedPartitions.add(tsFormat.truncatedTo(current).toInstant(ZoneOffset.UTC).toEpochMilli());
-        }
-
-        return unprocessedPartitions;
-    }
-
-    public void setClock(Clock clock) {
-        this.clock = clock;
-    }
-
-    public void checkProcessedPartitions() {
-        //todo-vp: we need to implement this
-    }
-}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepository.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepository.java
deleted file mode 100644
index 6c59c5997c..0000000000
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepository.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.nosql.repository;
-
-import com.datastax.driver.core.BoundStatement;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.ResultSetFuture;
-import com.datastax.driver.core.Row;
-import com.google.common.base.Function;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.dao.nosql.CassandraAbstractDao;
-import org.thingsboard.server.dao.queue.db.MsgAck;
-import org.thingsboard.server.dao.queue.db.repository.AckRepository;
-import org.thingsboard.server.dao.util.NoSqlDao;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
-
-@Component
-@NoSqlDao
-public class CassandraAckRepository extends CassandraAbstractDao implements AckRepository {
-
-    @Value("${cassandra.queue.ack.ttl}")
-    private int ackQueueTtl;
-
-    @Override
-    public ListenableFuture<Void> ack(MsgAck msgAck) {
-        String insert = "INSERT INTO msg_ack_queue (node_id, cluster_partition, ts_partition, msg_id) VALUES (?, ?, ?, ?) USING TTL ?";
-        PreparedStatement statement = prepare(insert);
-        BoundStatement boundStatement = statement.bind(msgAck.getNodeId(), msgAck.getClusteredPartition(),
-                msgAck.getTsPartition(), msgAck.getMsgId(), ackQueueTtl);
-        ResultSetFuture resultSetFuture = executeAsyncWrite(boundStatement);
-        return Futures.transform(resultSetFuture, (Function<ResultSet, Void>) input -> null);
-    }
-
-    @Override
-    public List<MsgAck> findAcks(UUID nodeId, long clusterPartition, long tsPartition) {
-        String select = "SELECT msg_id FROM msg_ack_queue WHERE " +
-                "node_id = ? AND cluster_partition = ? AND ts_partition = ?";
-        PreparedStatement statement = prepare(select);
-        BoundStatement boundStatement = statement.bind(nodeId, clusterPartition, tsPartition);
-        ResultSet rows = executeRead(boundStatement);
-        List<MsgAck> msgs = new ArrayList<>();
-        for (Row row : rows) {
-            msgs.add(new MsgAck(row.getUUID("msg_id"), nodeId, clusterPartition, tsPartition));
-        }
-        return msgs;
-    }
-
-}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraMsgRepository.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraMsgRepository.java
deleted file mode 100644
index a90b71d99c..0000000000
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraMsgRepository.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.nosql.repository;
-
-import com.datastax.driver.core.BoundStatement;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.ResultSetFuture;
-import com.datastax.driver.core.Row;
-import com.google.common.base.Function;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.common.msg.TbMsg;
-import org.thingsboard.server.dao.nosql.CassandraAbstractDao;
-import org.thingsboard.server.dao.queue.db.repository.MsgRepository;
-import org.thingsboard.server.dao.util.NoSqlDao;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
-
-@Component
-@NoSqlDao
-public class CassandraMsgRepository extends CassandraAbstractDao implements MsgRepository {
-
-    @Value("${cassandra.queue.msg.ttl}")
-    private int msqQueueTtl;
-
-    @Override
-    public ListenableFuture<Void> save(TbMsg msg, UUID nodeId, long clusterPartition, long tsPartition, long msgTs) {
-        String insert = "INSERT INTO msg_queue (node_id, cluster_partition, ts_partition, ts, msg) VALUES (?, ?, ?, ?, ?) USING TTL ?";
-        PreparedStatement statement = prepare(insert);
-        BoundStatement boundStatement = statement.bind(nodeId, clusterPartition, tsPartition, msgTs, TbMsg.toBytes(msg), msqQueueTtl);
-        ResultSetFuture resultSetFuture = executeAsyncWrite(boundStatement);
-        return Futures.transform(resultSetFuture, (Function<ResultSet, Void>) input -> null);
-    }
-
-    @Override
-    public List<TbMsg> findMsgs(UUID nodeId, long clusterPartition, long tsPartition) {
-        String select = "SELECT node_id, cluster_partition, ts_partition, ts, msg FROM msg_queue WHERE " +
-                "node_id = ? AND cluster_partition = ? AND ts_partition = ?";
-        PreparedStatement statement = prepare(select);
-        BoundStatement boundStatement = statement.bind(nodeId, clusterPartition, tsPartition);
-        ResultSet rows = executeRead(boundStatement);
-        List<TbMsg> msgs = new ArrayList<>();
-        for (Row row : rows) {
-            msgs.add(TbMsg.fromBytes(row.getBytes("msg")));
-        }
-        return msgs;
-    }
-
-}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraProcessedPartitionRepository.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraProcessedPartitionRepository.java
deleted file mode 100644
index 831c6fdf4b..0000000000
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraProcessedPartitionRepository.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.nosql.repository;
-
-import com.datastax.driver.core.BoundStatement;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.ResultSetFuture;
-import com.datastax.driver.core.Row;
-import com.google.common.base.Function;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.dao.nosql.CassandraAbstractDao;
-import org.thingsboard.server.dao.queue.db.repository.ProcessedPartitionRepository;
-import org.thingsboard.server.dao.util.NoSqlDao;
-
-import java.util.Optional;
-import java.util.UUID;
-
-@Component
-@NoSqlDao
-public class CassandraProcessedPartitionRepository extends CassandraAbstractDao implements ProcessedPartitionRepository {
-
-    @Value("${cassandra.queue.partitions.ttl}")
-    private int partitionsTtl;
-
-    @Override
-    public ListenableFuture<Void> partitionProcessed(UUID nodeId, long clusterPartition, long tsPartition) {
-        String insert = "INSERT INTO processed_msg_partitions (node_id, cluster_partition, ts_partition) VALUES (?, ?, ?) USING TTL ?";
-        PreparedStatement prepared = prepare(insert);
-        BoundStatement boundStatement = prepared.bind(nodeId, clusterPartition, tsPartition, partitionsTtl);
-        ResultSetFuture resultSetFuture = executeAsyncWrite(boundStatement);
-        return Futures.transform(resultSetFuture, (Function<ResultSet, Void>) input -> null);
-    }
-
-    @Override
-    public Optional<Long> findLastProcessedPartition(UUID nodeId, long clusteredHash) {
-        String select = "SELECT ts_partition FROM processed_msg_partitions WHERE " +
-                "node_id = ? AND cluster_partition = ?";
-        PreparedStatement prepared = prepare(select);
-        BoundStatement boundStatement = prepared.bind(nodeId, clusteredHash);
-        Row row = executeRead(boundStatement).one();
-        if (row == null) {
-            return Optional.empty();
-        }
-
-        return Optional.of(row.getLong("ts_partition"));
-    }
-}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/AckRepository.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/AckRepository.java
deleted file mode 100644
index 6fbd2da57e..0000000000
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/AckRepository.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.repository;
-
-import com.google.common.util.concurrent.ListenableFuture;
-import org.thingsboard.server.dao.queue.db.MsgAck;
-
-import java.util.List;
-import java.util.UUID;
-
-public interface AckRepository {
-
-    ListenableFuture<Void> ack(MsgAck msgAck);
-
-    List<MsgAck> findAcks(UUID nodeId, long clusterPartition, long tsPartition);
-}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/MsgRepository.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/MsgRepository.java
deleted file mode 100644
index 0ca6900fc9..0000000000
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/MsgRepository.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.repository;
-
-import com.google.common.util.concurrent.ListenableFuture;
-import org.thingsboard.server.common.msg.TbMsg;
-
-import java.util.List;
-import java.util.UUID;
-
-public interface MsgRepository {
-
-    ListenableFuture<Void> save(TbMsg msg, UUID nodeId, long clusterPartition, long tsPartition, long msgTs);
-
-    List<TbMsg> findMsgs(UUID nodeId, long clusterPartition, long tsPartition);
-
-}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/ProcessedPartitionRepository.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/ProcessedPartitionRepository.java
deleted file mode 100644
index b11fc6cc1c..0000000000
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/ProcessedPartitionRepository.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.repository;
-
-import com.google.common.util.concurrent.ListenableFuture;
-
-import java.util.Optional;
-import java.util.UUID;
-
-public interface ProcessedPartitionRepository {
-
-    ListenableFuture<Void> partitionProcessed(UUID nodeId, long clusteredHash, long partition);
-
-    Optional<Long> findLastProcessedPartition(UUID nodeId, long clusteredHash);
-
-}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/sql/SqlMsgQueue.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/sql/SqlMsgQueue.java
deleted file mode 100644
index f9dda43a36..0000000000
--- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/sql/SqlMsgQueue.java
+++ /dev/null
@@ -1,20 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.sql;
-
-//@todo-vp: implement
-public class SqlMsgQueue {
-}
diff --git a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/QueuePartitionerTest.java b/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/QueuePartitionerTest.java
deleted file mode 100644
index e76ca2f946..0000000000
--- a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/QueuePartitionerTest.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.nosql;
-
-
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.Mock;
-import org.mockito.runners.MockitoJUnitRunner;
-import org.thingsboard.server.dao.queue.db.repository.ProcessedPartitionRepository;
-
-import java.time.Clock;
-import java.time.Instant;
-import java.time.ZoneOffset;
-import java.time.temporal.ChronoUnit;
-import java.util.List;
-import java.util.Optional;
-import java.util.UUID;
-
-import static org.junit.Assert.assertEquals;
-import static org.mockito.Mockito.when;
-
-@RunWith(MockitoJUnitRunner.class)
-public class QueuePartitionerTest {
-
-    private QueuePartitioner queuePartitioner;
-
-    @Mock
-    private ProcessedPartitionRepository partitionRepo;
-
-    private Instant startInstant;
-    private Instant endInstant;
-
-    @Before
-    public void init() {
-        queuePartitioner = new QueuePartitioner("MINUTES", partitionRepo);
-        startInstant = Instant.now();
-        endInstant = startInstant.plus(2, ChronoUnit.MINUTES);
-        queuePartitioner.setClock(Clock.fixed(endInstant, ZoneOffset.UTC));
-    }
-
-    @Test
-    public void partitionCalculated() {
-        long time = 1519390191425L;
-        long partition = queuePartitioner.getPartition(time);
-        assertEquals(1519390140000L, partition);
-    }
-
-    @Test
-    public void unprocessedPartitionsReturned() {
-        UUID nodeId = UUID.randomUUID();
-        long clusteredHash = 101L;
-        when(partitionRepo.findLastProcessedPartition(nodeId, clusteredHash)).thenReturn(Optional.of(startInstant.toEpochMilli()));
-        List<Long> actual = queuePartitioner.findUnprocessedPartitions(nodeId, clusteredHash);
-        assertEquals(3, actual.size());
-    }
-
-    @Test
-    public void defaultShiftUsedIfNoPartitionWasProcessed() {
-        UUID nodeId = UUID.randomUUID();
-        long clusteredHash = 101L;
-        when(partitionRepo.findLastProcessedPartition(nodeId, clusteredHash)).thenReturn(Optional.empty());
-        List<Long> actual = queuePartitioner.findUnprocessedPartitions(nodeId, clusteredHash);
-        assertEquals(10083, actual.size());
-    }
-
-}
\ No newline at end of file
diff --git a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/UnprocessedMsgFilterTest.java b/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/UnprocessedMsgFilterTest.java
deleted file mode 100644
index fd9bf21164..0000000000
--- a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/UnprocessedMsgFilterTest.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.nosql;
-
-import com.google.common.collect.Lists;
-import org.junit.Test;
-import org.thingsboard.server.common.msg.TbMsg;
-import org.thingsboard.server.dao.queue.db.MsgAck;
-import org.thingsboard.server.dao.queue.db.UnprocessedMsgFilter;
-
-import java.util.Collection;
-import java.util.List;
-import java.util.UUID;
-
-import static org.junit.Assert.assertEquals;
-
-public class UnprocessedMsgFilterTest {
-
-    private UnprocessedMsgFilter msgFilter = new UnprocessedMsgFilter();
-
-    @Test
-    public void acknowledgedMsgsAreFilteredOut() {
-        UUID id1 = UUID.randomUUID();
-        UUID id2 = UUID.randomUUID();
-        TbMsg msg1 = new TbMsg(id1, "T", null, null, null, null, null, null, 0L);
-        TbMsg msg2 = new TbMsg(id2, "T", null, null, null, null, null, null, 0L);
-        List<TbMsg> msgs = Lists.newArrayList(msg1, msg2);
-        List<MsgAck> acks = Lists.newArrayList(new MsgAck(id2, UUID.randomUUID(), 1L, 1L));
-        Collection<TbMsg> actual = msgFilter.filter(msgs, acks);
-        assertEquals(1, actual.size());
-        assertEquals(msg1, actual.iterator().next());
-    }
-
-}
\ No newline at end of file
diff --git a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepositoryTest.java b/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepositoryTest.java
deleted file mode 100644
index b2f38dc539..0000000000
--- a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepositoryTest.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.nosql.repository;
-
-import com.datastax.driver.core.utils.UUIDs;
-import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.ListenableFuture;
-import org.junit.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.test.util.ReflectionTestUtils;
-import org.thingsboard.server.dao.service.AbstractServiceTest;
-import org.thingsboard.server.dao.service.DaoNoSqlTest;
-import org.thingsboard.server.dao.queue.db.MsgAck;
-
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-@DaoNoSqlTest
-public class CassandraAckRepositoryTest extends AbstractServiceTest {
-
-    @Autowired
-    private CassandraAckRepository ackRepository;
-
-    @Test
-    public void acksInPartitionCouldBeFound() {
-        UUID nodeId = UUID.fromString("055eee50-1883-11e8-b380-65b5d5335ba9");
-
-        List<MsgAck> extectedAcks = Lists.newArrayList(
-                new MsgAck(UUID.fromString("bebaeb60-1888-11e8-bf21-65b5d5335ba9"), nodeId, 101L, 300L),
-                new MsgAck(UUID.fromString("12baeb60-1888-11e8-bf21-65b5d5335ba9"), nodeId, 101L, 300L)
-        );
-
-        List<MsgAck> actualAcks = ackRepository.findAcks(nodeId, 101L, 300L);
-        assertEquals(extectedAcks, actualAcks);
-    }
-
-    @Test
-    public void ackCanBeSavedAndRead() throws ExecutionException, InterruptedException {
-        UUID msgId = UUIDs.timeBased();
-        UUID nodeId = UUIDs.timeBased();
-        MsgAck ack = new MsgAck(msgId, nodeId, 10L, 20L);
-        ListenableFuture<Void> future = ackRepository.ack(ack);
-        future.get();
-        List<MsgAck> actualAcks = ackRepository.findAcks(nodeId, 10L, 20L);
-        assertEquals(1, actualAcks.size());
-        assertEquals(ack, actualAcks.get(0));
-    }
-
-    @Test
-    public void expiredAcksAreNotReturned() throws ExecutionException, InterruptedException {
-        ReflectionTestUtils.setField(ackRepository, "ackQueueTtl", 1);
-        UUID msgId = UUIDs.timeBased();
-        UUID nodeId = UUIDs.timeBased();
-        MsgAck ack = new MsgAck(msgId, nodeId, 30L, 40L);
-        ListenableFuture<Void> future = ackRepository.ack(ack);
-        future.get();
-        List<MsgAck> actualAcks = ackRepository.findAcks(nodeId, 30L, 40L);
-        assertEquals(1, actualAcks.size());
-        TimeUnit.SECONDS.sleep(2);
-        assertTrue(ackRepository.findAcks(nodeId, 30L, 40L).isEmpty());
-    }
-
-
-}
\ No newline at end of file
diff --git a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraMsgRepositoryTest.java b/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraMsgRepositoryTest.java
deleted file mode 100644
index f31db877ef..0000000000
--- a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraMsgRepositoryTest.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.nosql.repository;
-
-//import static org.junit.jupiter.api.Assertions.*;
-
-import com.datastax.driver.core.utils.UUIDs;
-import com.google.common.util.concurrent.ListenableFuture;
-import org.junit.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.test.util.ReflectionTestUtils;
-import org.thingsboard.server.common.data.id.DeviceId;
-import org.thingsboard.server.common.data.id.RuleChainId;
-import org.thingsboard.server.common.data.id.RuleNodeId;
-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.dao.service.AbstractServiceTest;
-import org.thingsboard.server.dao.service.DaoNoSqlTest;
-
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-@DaoNoSqlTest
-public class CassandraMsgRepositoryTest extends AbstractServiceTest {
-
-    @Autowired
-    private CassandraMsgRepository msgRepository;
-
-    @Test
-    public void msgCanBeSavedAndRead() throws ExecutionException, InterruptedException {
-        TbMsg msg = new TbMsg(UUIDs.timeBased(), "type", new DeviceId(UUIDs.timeBased()), null, TbMsgDataType.JSON, "0000",
-                new RuleChainId(UUIDs.timeBased()), new RuleNodeId(UUIDs.timeBased()), 0L);
-        UUID nodeId = UUIDs.timeBased();
-        ListenableFuture<Void> future = msgRepository.save(msg, nodeId, 1L, 1L, 1L);
-        future.get();
-        List<TbMsg> msgs = msgRepository.findMsgs(nodeId, 1L, 1L);
-        assertEquals(1, msgs.size());
-    }
-
-    @Test
-    public void expiredMsgsAreNotReturned() throws ExecutionException, InterruptedException {
-        ReflectionTestUtils.setField(msgRepository, "msqQueueTtl", 1);
-        TbMsg msg = new TbMsg(UUIDs.timeBased(), "type", new DeviceId(UUIDs.timeBased()), null, TbMsgDataType.JSON, "0000",
-                new RuleChainId(UUIDs.timeBased()), new RuleNodeId(UUIDs.timeBased()), 0L);
-        UUID nodeId = UUIDs.timeBased();
-        ListenableFuture<Void> future = msgRepository.save(msg, nodeId, 2L, 2L, 2L);
-        future.get();
-        TimeUnit.SECONDS.sleep(2);
-        assertTrue(msgRepository.findMsgs(nodeId, 2L, 2L).isEmpty());
-    }
-
-    @Test
-    public void protoBufConverterWorkAsExpected() throws ExecutionException, InterruptedException {
-        TbMsgMetaData metaData = new TbMsgMetaData();
-        metaData.putValue("key", "value");
-        String dataStr = "someContent";
-        TbMsg msg = new TbMsg(UUIDs.timeBased(), "type", new DeviceId(UUIDs.timeBased()), metaData, TbMsgDataType.JSON, dataStr,
-                new RuleChainId(UUIDs.timeBased()), new RuleNodeId(UUIDs.timeBased()), 0L);
-        UUID nodeId = UUIDs.timeBased();
-        ListenableFuture<Void> future = msgRepository.save(msg, nodeId, 1L, 1L, 1L);
-        future.get();
-        List<TbMsg> msgs = msgRepository.findMsgs(nodeId, 1L, 1L);
-        assertEquals(1, msgs.size());
-        assertEquals(msg, msgs.get(0));
-    }
-
-
-}
\ No newline at end of file
diff --git a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraProcessedPartitionRepositoryTest.java b/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraProcessedPartitionRepositoryTest.java
deleted file mode 100644
index a76fd965ca..0000000000
--- a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraProcessedPartitionRepositoryTest.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/**
- * Copyright © 2016-2018 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.dao.queue.db.nosql.repository;
-
-import com.datastax.driver.core.utils.UUIDs;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import org.junit.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.test.util.ReflectionTestUtils;
-import org.thingsboard.server.dao.service.AbstractServiceTest;
-import org.thingsboard.server.dao.service.DaoNoSqlTest;
-
-import java.util.List;
-import java.util.Optional;
-import java.util.UUID;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-@DaoNoSqlTest
-public class CassandraProcessedPartitionRepositoryTest extends AbstractServiceTest {
-
-    @Autowired
-    private CassandraProcessedPartitionRepository partitionRepository;
-
-    @Test
-    public void lastProcessedPartitionCouldBeFound() {
-        UUID nodeId = UUID.fromString("055eee50-1883-11e8-b380-65b5d5335ba9");
-        Optional<Long> lastProcessedPartition = partitionRepository.findLastProcessedPartition(nodeId, 101L);
-        assertTrue(lastProcessedPartition.isPresent());
-        assertEquals((Long) 777L, lastProcessedPartition.get());
-    }
-
-    @Test
-    public void highestProcessedPartitionReturned() throws ExecutionException, InterruptedException {
-        UUID nodeId = UUIDs.timeBased();
-        ListenableFuture<Void> future1 = partitionRepository.partitionProcessed(nodeId, 303L, 100L);
-        ListenableFuture<Void> future2 = partitionRepository.partitionProcessed(nodeId, 303L, 200L);
-        ListenableFuture<Void> future3 = partitionRepository.partitionProcessed(nodeId, 303L, 10L);
-        ListenableFuture<List<Void>> allFutures = Futures.allAsList(future1, future2, future3);
-        allFutures.get();
-        Optional<Long> actual = partitionRepository.findLastProcessedPartition(nodeId, 303L);
-        assertTrue(actual.isPresent());
-        assertEquals((Long) 200L, actual.get());
-    }
-
-    @Test
-    public void expiredPartitionsAreNotReturned() throws ExecutionException, InterruptedException {
-        ReflectionTestUtils.setField(partitionRepository, "partitionsTtl", 1);
-        UUID nodeId = UUIDs.timeBased();
-        ListenableFuture<Void> future = partitionRepository.partitionProcessed(nodeId, 404L, 10L);
-        future.get();
-        Optional<Long> actual = partitionRepository.findLastProcessedPartition(nodeId, 404L);
-        assertEquals((Long) 10L, actual.get());
-        TimeUnit.SECONDS.sleep(2);
-        assertFalse(partitionRepository.findLastProcessedPartition(nodeId, 404L).isPresent());
-    }
-
-    @Test
-    public void ifNoPartitionsWereProcessedEmptyResultReturned() {
-        UUID nodeId = UUIDs.timeBased();
-        Optional<Long> actual = partitionRepository.findLastProcessedPartition(nodeId, 505L);
-        assertFalse(actual.isPresent());
-    }
-
-}
\ No newline at end of file