Merge with 2.5

application/src/main/data/json/demo/rule_chains/root_rule_chain.json

@@ -43,7 +43,8 @@
         "name": "Save Client Attributes",
         "debugMode": false,
         "configuration": {
-          "scope": "CLIENT_SCOPE"
+          "scope": "CLIENT_SCOPE",
+          "notifyDevice": "false"
         }
       },
       {

application/src/main/data/json/tenant/rule_chains/root_rule_chain.json

@@ -31,7 +31,8 @@
         "name": "Save Client Attributes",
         "debugMode": false,
         "configuration": {
-          "scope": "CLIENT_SCOPE"
+          "scope": "CLIENT_SCOPE",
+          "notifyDevice": "false"
         }
       },
       {

application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java

@@ -226,6 +226,11 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
 
     @Override
     public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, TbCallback callback) {
+        onAttributesUpdate(tenantId, entityId, scope, attributes, true, callback);
+    }
+
+    @Override
+    public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, TbCallback callback) {
         onLocalTelemetrySubUpdate(entityId,
                 s -> {
                     if (TbSubscriptionType.ATTRIBUTES.equals(s.getType())) {
@@ -254,7 +259,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
                         deviceStateService.onDeviceInactivityTimeoutUpdate(new DeviceId(entityId.getId()), attribute.getLongValue().orElse(0L));
                     }
                 }
-            } else if (TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(scope)) {
+            } else if (TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(scope) && notifyDevice) {
                 clusterService.pushMsgToCore(DeviceAttributesEventNotificationMsg.onUpdate(tenantId,
                         new DeviceId(entityId.getId()), DataConstants.SHARED_SCOPE, new ArrayList<>(attributes))
                         , null);

application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java

@@ -17,13 +17,12 @@ package org.thingsboard.server.service.subscription;
 
 import org.springframework.context.ApplicationListener;
 import org.thingsboard.server.common.data.alarm.Alarm;
-import org.thingsboard.server.common.data.id.AlarmId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
-import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 
 import java.util.List;
 
@@ -37,9 +36,13 @@ public interface SubscriptionManagerService extends ApplicationListener<Partitio
 
     void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, TbCallback callback);
 
+    void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, TbCallback callback);
+
     void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List<String> keys, TbCallback empty);
 
     void onAlarmUpdate(TenantId tenantId, EntityId entityId, Alarm alarm, TbCallback callback);
 
     void onAlarmDeleted(TenantId tenantId, EntityId entityId, Alarm alarm, TbCallback callback);
+
+
 }

application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java

@@ -171,9 +171,14 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
 
     @Override
     public void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback) {
+        saveAndNotify(tenantId, entityId, scope, attributes, true, callback);
+    }
+
+    @Override
+    public void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback) {
         ListenableFuture<List<Void>> saveFuture = attrService.save(tenantId, entityId, scope, attributes);
         addMainCallback(saveFuture, callback);
-        addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope, attributes));
+        addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice));
     }
 
     @Override
@@ -236,11 +241,11 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
                 , System.currentTimeMillis())), callback);
     }
 
-    private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes) {
+    private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice) {
         TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
         if (currentPartitions.contains(tpi)) {
             if (subscriptionManagerService.isPresent()) {
-                subscriptionManagerService.get().onAttributesUpdate(tenantId, entityId, scope, attributes, TbCallback.EMPTY);
+                subscriptionManagerService.get().onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice, TbCallback.EMPTY);
             } else {
                 log.warn("Possible misconfiguration because subscriptionManagerService is null!");
             }

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

@@ -298,6 +298,7 @@ public class DefaultTransportService implements TransportService {
             TbMsgMetaData metaData = new TbMsgMetaData();
             metaData.putValue("deviceName", sessionInfo.getDeviceName());
             metaData.putValue("deviceType", sessionInfo.getDeviceType());
+            metaData.putValue("notifyDevice", "false");
             TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), deviceId, metaData, gson.toJson(json));
             sendToRuleEngine(tenantId, tbMsg, new TransportTbQueueCallback(callback));
         }

dao/src/test/resources/sql-test.properties

@@ -13,7 +13,7 @@ spring.jpa.database-platform=org.hibernate.dialect.HSQLDialect
 
 spring.datasource.username=sa
 spring.datasource.password=
-spring.datasource.url=jdbc:hsqldb:file:/tmp/testDb;sql.enforce_size=false
+spring.datasource.url=jdbc:hsqldb:file:target/tmp/testDb;sql.enforce_size=false
 spring.datasource.driverClassName=org.hsqldb.jdbc.JDBCDriver
 spring.datasource.hikari.maximumPoolSize = 50
 

msa/js-executor/package.json

@@ -19,7 +19,7 @@
     "azure-sb": "^0.11.1",
     "config": "^3.3.1",
     "js-yaml": "^3.14.0",
-    "kafkajs": "^1.12.0",
+    "kafkajs": "^1.14.0",
     "long": "^4.0.0",
     "uuid-parse": "^1.1.0",
     "uuid-random": "^1.3.2",

msa/js-executor/queue/kafkaTemplate.js

@@ -27,20 +27,10 @@ let kafkaAdmin;
 let consumer;
 let producer;
 
-const topics = [];
 const configEntries = [];
 
 function KafkaProducer() {
     this.send = async (responseTopic, scriptId, rawResponse, headers) => {
-
-        if (!topics.includes(responseTopic)) {
-            let createResponseTopicResult = await createTopic(responseTopic, 1);
-            topics.push(responseTopic);
-            if (createResponseTopicResult) {
-                logger.info('Created new topic: %s', requestTopic);
-            }
-        }
-
         return producer.send(
             {
                 topic: responseTopic,
@@ -99,10 +89,13 @@ function KafkaProducer() {
             }
         }
 
-        let createRequestTopicResult = await createTopic(requestTopic, partitions);
+        let topics = await kafkaAdmin.listTopics();
 
-        if (createRequestTopicResult) {
+        if (!topics.includes(requestTopic)) {
-            logger.info('Created new topic: %s', requestTopic);
+            let createRequestTopicResult = await createTopic(requestTopic, partitions);
+            if (createRequestTopicResult) {
+                logger.info('Created new topic: %s', requestTopic);
+            }
         }
 
         consumer = kafkaClient.consumer({groupId: 'js-executor-group'});

msa/js-executor/yarn.lock

@@ -1665,12 +1665,10 @@ jws@^4.0.0:
     jwa "^2.0.0"
     safe-buffer "^5.0.1"
 
-kafkajs@^1.12.0:
+kafkajs@^1.14.0:
-  version "1.12.0"
+  version "1.14.0"
-  resolved "https://registry.yarnpkg.com/kafkajs/-/kafkajs-1.12.0.tgz#50ad336baee95f3324af8ae8df6fadc96e07c613"
+  resolved "https://registry.yarnpkg.com/kafkajs/-/kafkajs-1.14.0.tgz#3d998a77bfde54dc502e8e88690eedf0b21a1ed6"
-  integrity sha512-Izkd9iFRgeeKaHEgVpGQH08ygzCbHSxTbnu8W3G3uiNaVjGibUTmTwjv1Qf2M8NORXcPfzwVyg6bBlVj4SKr9g==
+  integrity sha512-W+WCekiooY5rJP3Me5N3gWcQ8O6uG6lw0vv9t+sI+WqXKjKwj2+CWIXJy241x+ITE+1M1D19ABSiL2J8lKja5A==
-  dependencies:
-    long "^4.0.0"
 
 keyv@^3.0.0:
   version "3.1.0"

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java

@@ -16,7 +16,6 @@
 package org.thingsboard.rule.engine.api;
 
 import com.google.common.util.concurrent.FutureCallback;
-import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
@@ -24,7 +23,6 @@ import org.thingsboard.server.common.data.kv.TsKvEntry;
 
 import java.util.Collection;
 import java.util.List;
-import java.util.Set;
 
 /**
  * Created by ashvayka on 02.04.18.
@@ -37,6 +35,8 @@ public interface RuleEngineTelemetryService {
 
     void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback);
 
+    void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback);
+
     void saveLatestAndNotify(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback);
 
     void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, long value, FutureCallback<Void> callback);

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java

@@ -17,15 +17,13 @@ package org.thingsboard.rule.engine.telemetry;
 
 import com.google.gson.JsonParser;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.thingsboard.rule.engine.api.RuleNode;
 import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.rule.engine.api.TbNode;
 import org.thingsboard.rule.engine.api.TbNodeConfiguration;
 import org.thingsboard.rule.engine.api.TbNodeException;
 import org.thingsboard.rule.engine.api.util.TbNodeUtils;
-import org.thingsboard.server.common.data.DataConstants;
-import org.thingsboard.server.common.data.EntityType;
-import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.common.msg.TbMsg;
@@ -53,6 +51,9 @@ public class TbMsgAttributesNode implements TbNode {
     @Override
     public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbMsgAttributesNodeConfiguration.class);
+        if (config.getNotifyDevice() == null) {
+            config.setNotifyDevice(true);
+        }
     }
 
     @Override
@@ -63,7 +64,15 @@ public class TbMsgAttributesNode implements TbNode {
         }
         String src = msg.getData();
         Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(src));
-        ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), msg.getOriginator(), config.getScope(), new ArrayList<>(attributes), new TelemetryNodeCallback(ctx, msg));
+        String notifyDeviceStr = msg.getMetaData().getValue("notifyDevice");
+        ctx.getTelemetryService().saveAndNotify(
+                ctx.getTenantId(),
+                msg.getOriginator(),
+                config.getScope(),
+                new ArrayList<>(attributes),
+                config.getNotifyDevice() || StringUtils.isEmpty(notifyDeviceStr) || Boolean.parseBoolean(notifyDeviceStr),
+                new TelemetryNodeCallback(ctx, msg)
+        );
     }
 
     @Override

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfiguration.java

@@ -24,10 +24,13 @@ public class TbMsgAttributesNodeConfiguration implements NodeConfiguration<TbMsg
 
     private String scope;
 
+    private Boolean notifyDevice;
+
     @Override
     public TbMsgAttributesNodeConfiguration defaultConfiguration() {
         TbMsgAttributesNodeConfiguration configuration = new TbMsgAttributesNodeConfiguration();
         configuration.setScope(DataConstants.SERVER_SCOPE);
+        configuration.setNotifyDevice(false);
         return configuration;
     }
 }