moved queueName from rule-node config to rule-node

application/src/main/data/upgrade/3.6.2/schema_update.sql

@@ -0,0 +1,19 @@
+--
+-- Copyright © 2016-2023 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.
+--
+
+ALTER TABLE rule_node ADD COLUMN IF NOT EXISTS queue_name varchar(255);
+
+ALTER TABLE component_descriptor ADD COLUMN IF NOT EXISTS has_queue_name boolean DEFAULT false;

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

@@ -185,7 +185,7 @@ class DefaultTbContext implements TbContext {
 
     @Override
     public void enqueue(TbMsg tbMsg, Runnable onSuccess, Consumer<Throwable> onFailure) {
-        TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, getTenantId(), tbMsg.getOriginator());
+        TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, nodeCtx.getSelf().getQueueName(), getTenantId(), tbMsg.getOriginator());
         enqueue(tpi, tbMsg, onFailure, onSuccess);
     }
 
@@ -310,7 +310,7 @@ class DefaultTbContext implements TbContext {
 
     @Override
     public boolean isLocalEntity(EntityId entityId) {
-        return mainCtx.resolve(ServiceType.TB_RULE_ENGINE, getTenantId(), entityId).isMyPartition();
+        return mainCtx.resolve(ServiceType.TB_RULE_ENGINE, nodeCtx.getSelf().getQueueName(), getTenantId(), entityId).isMyPartition();
     }
 
     private void scheduleMsgWithDelay(TbActorMsg msg, long delayInMs, TbActorRef target) {

application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java

@@ -277,6 +277,11 @@ public class ThingsboardInstallService {
                             } else {
                                 log.info("Skipping images migration. Run the upgrade with fromVersion as '3.6.2-images' to migrate");
                             }
+                            break;
+                        case "3.6.2":
+                            log.info("Upgrading ThingsBoard from version 3.6.2 to 3.6.3 ...");
+                            databaseEntitiesUpgradeService.upgradeDatabase("3.6.2");
+                            dataUpdateService.updateData("3.6.2");
                             //TODO DON'T FORGET to update switch statement in the CacheCleanupService if you need to clear the cache
                             break;
                         default:

application/src/main/java/org/thingsboard/server/service/component/AnnotationComponentDiscoveryService.java

@@ -189,6 +189,7 @@ public class AnnotationComponentDiscoveryService implements ComponentDiscoverySe
             scannedComponent.setName(ruleNodeAnnotation.name());
             scannedComponent.setScope(ruleNodeAnnotation.scope());
             scannedComponent.setClusteringMode(ruleNodeAnnotation.clusteringMode());
+            scannedComponent.setHasQueueName(ruleNodeAnnotation.hasQueueName());
             NodeDefinition nodeDefinition = prepareNodeDefinition(clazz, ruleNodeAnnotation);
             ObjectNode configurationDescriptor = JacksonUtil.newObjectNode();
             JsonNode node = JacksonUtil.valueToTree(nodeDefinition);

application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java

@@ -772,6 +772,9 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService
                     }
                 });
                 break;
+            case "3.6.2":
+                updateSchema("3.6.2", 3006002, "3.6.3", 3006003, null);
+                break;
             default:
                 throw new RuntimeException("Unable to upgrade SQL database, unsupported fromVersion: " + fromVersion);
         }

application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java

@@ -227,11 +227,25 @@ public class DefaultDataUpdateService implements DataUpdateService {
                 log.info("Updating data from version 3.6.0 to 3.6.1 ...");
                 migrateDeviceConnectivity();
                 break;
+            case "3.6.2":
+                updateRuleNodesQueueName();
+                break;
             default:
                 throw new RuntimeException("Unable to update data, unsupported fromVersion: " + fromVersion);
         }
     }
 
+    private void updateRuleNodesQueueName() {
+        String[] ruleNodeTypes = {
+                "org.thingsboard.rule.engine.debug.TbMsgGeneratorNode",
+                "org.thingsboard.rule.engine.flow.TbCheckpointNode",
+                "org.thingsboard.rule.engine.deduplication.TbMsgDeduplicationNode"
+        };
+        for (String ruleNodeType : ruleNodeTypes) {
+            ruleNodeQueueNameUpdater.updateEntities(ruleNodeType);
+        }
+    }
+
     private void migrateEdgeEvents(String logPrefix) {
         boolean skipEdgeEventsMigration = getEnv("TB_SKIP_EDGE_EVENTS_MIGRATION", false);
         if (!skipEdgeEventsMigration) {
@@ -792,4 +806,39 @@ public class DefaultDataUpdateService implements DataUpdateService {
         }
     }
 
+    private final PaginatedUpdater<String, RuleNode> ruleNodeQueueNameUpdater =
+            new PaginatedUpdater<>() {
+
+                @Override
+                protected String getName() {
+                    return "RuleNode queue name updater";
+                }
+
+                @Override
+                protected boolean forceReportTotal() {
+                    return true;
+                }
+
+                @Override
+                protected PageData<RuleNode> findEntities(String type, PageLink pageLink) {
+                    return ruleChainService.findAllRuleNodesByType(type, pageLink);
+                }
+
+                @Override
+                protected void updateEntity(RuleNode ruleNode) {
+                    try {
+                        ObjectNode configuration = (ObjectNode) ruleNode.getConfiguration();
+                        JsonNode queueName = configuration.remove("queueName");
+                        if (queueName != null) {
+                            if (!queueName.isNull()) {
+                                ruleNode.setQueueName(queueName.asText());
+                            }
+                            ruleChainService.saveRuleNode(null, ruleNode);
+                        }
+                    } catch (Exception e) {
+                        log.error("Unable to update RuleNode", e);
+                    }
+                }
+            };
+
 }

application/src/main/java/org/thingsboard/server/service/sync/ie/importing/impl/RuleChainImportService.java

@@ -15,13 +15,14 @@
  */
 package org.thingsboard.server.service.sync.ie.importing.impl;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.User;
 import org.thingsboard.server.common.data.audit.ActionType;
-import org.thingsboard.server.common.data.exception.ThingsboardException;
 import org.thingsboard.server.common.data.id.RuleChainId;
 import org.thingsboard.server.common.data.id.RuleNodeId;
 import org.thingsboard.server.common.data.id.TenantId;
@@ -84,12 +85,14 @@ public class RuleChainImportService extends BaseEntityImportService<RuleChainId,
                 node.setRuleChainId(old.getId());
                 node.setExternalId(node.getId());
                 node.setId((RuleNodeId) ctx.getInternalId(node.getId()));
+                setQueueName(node);
             });
         } else {
             ruleNodes.forEach(node -> {
                 node.setRuleChainId(null);
                 node.setExternalId(node.getId());
                 node.setId(null);
+                setQueueName(node);
             });
         }
 
@@ -105,6 +108,22 @@ public class RuleChainImportService extends BaseEntityImportService<RuleChainId,
         return ruleChain;
     }
 
+    private void setQueueName(RuleNode ruleNode) {
+        try {
+            Class<?> clazz = Class.forName(ruleNode.getType());
+            org.thingsboard.rule.engine.api.RuleNode ruleNodeAnnotation = clazz.getAnnotation(org.thingsboard.rule.engine.api.RuleNode.class);
+            if (ruleNodeAnnotation.hasQueueName()) {
+                ObjectNode configuration = (ObjectNode) ruleNode.getConfiguration();
+                JsonNode queueName = configuration.remove("queueName");
+                if (queueName != null && !queueName.isNull()) {
+                    ruleNode.setQueueName(queueName.asText());
+                }
+            }
+        } catch (ClassNotFoundException e) {
+            log.warn("[{}] RuleNode class not found [{}]", ruleNode.getName(), ruleNode.getType());
+        }
+    }
+
     @Override
     protected RuleChain saveOrUpdate(EntitiesImportCtx ctx, RuleChain ruleChain, RuleChainExportData exportData, IdProvider idProvider) {
         ruleChain = ruleChainService.saveRuleChain(ruleChain);

common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentDescriptor.java

@@ -55,6 +55,8 @@ public class ComponentDescriptor extends BaseData<ComponentDescriptorId> {
     @Length(fieldName = "actions")
     @ApiModelProperty(position = 10, value = "Rule Node Actions. Deprecated. Always null.", accessMode = ApiModelProperty.AccessMode.READ_ONLY)
     @Getter @Setter private String actions;
+    @ApiModelProperty(position = 11, value = "Indicates that the RuleNode is support queue name.", accessMode = ApiModelProperty.AccessMode.READ_ONLY, example = "true")
+    @Getter @Setter private boolean hasQueueName;
 
     public ComponentDescriptor() {
         super();
@@ -74,6 +76,7 @@ public class ComponentDescriptor extends BaseData<ComponentDescriptorId> {
         this.configurationDescriptor = plugin.getConfigurationDescriptor();
         this.configurationVersion = plugin.getConfigurationVersion();
         this.actions = plugin.getActions();
+        this.hasQueueName = plugin.isHasQueueName();
     }
 
     @ApiModelProperty(position = 1, value = "JSON object with the descriptor Id. " +
@@ -104,6 +107,8 @@ public class ComponentDescriptor extends BaseData<ComponentDescriptorId> {
         if (!Objects.equals(actions, that.actions)) return false;
         if (!Objects.equals(configurationDescriptor, that.configurationDescriptor)) return false;
         if (configurationVersion != that.configurationVersion) return false;
+        if (clusteringMode != that.clusteringMode) return false;
+        if (hasQueueName != that.isHasQueueName()) return false;
         return Objects.equals(clazz, that.clazz);
     }
 
@@ -115,6 +120,8 @@ public class ComponentDescriptor extends BaseData<ComponentDescriptorId> {
         result = 31 * result + (name != null ? name.hashCode() : 0);
         result = 31 * result + (clazz != null ? clazz.hashCode() : 0);
         result = 31 * result + (actions != null ? actions.hashCode() : 0);
+        result = 31 * result + (clusteringMode != null ? clusteringMode.hashCode() : 0);
+        result = 31 * result + (hasQueueName ? 1 : 0);
         return result;
     }
 

common/data/src/main/java/org/thingsboard/server/common/data/rule/RuleNode.java

@@ -50,9 +50,11 @@ public class RuleNode extends BaseDataWithAdditionalInfo<RuleNodeId> implements
     private boolean debugMode;
     @ApiModelProperty(position = 7, value = "Enable/disable singleton mode. ", example = "false")
     private boolean singletonMode;
-    @ApiModelProperty(position = 8, value = "Version of rule node configuration. ", example = "0")
+    @ApiModelProperty(position = 8, value = "Queue name. ", example = "Main")
+    private String queueName;
+    @ApiModelProperty(position = 9, value = "Version of rule node configuration. ", example = "0")
     private int configurationVersion;
-    @ApiModelProperty(position = 9, value = "JSON with the rule node configuration. Structure depends on the rule node implementation.", dataType = "com.fasterxml.jackson.databind.JsonNode")
+    @ApiModelProperty(position = 10, value = "JSON with the rule node configuration. Structure depends on the rule node implementation.", dataType = "com.fasterxml.jackson.databind.JsonNode")
     private transient JsonNode configuration;
     @JsonIgnore
     private byte[] configurationBytes;

dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java

@@ -354,6 +354,7 @@ public class ModelConstants {
     public static final String COMPONENT_DESCRIPTOR_CONFIGURATION_DESCRIPTOR_PROPERTY = "configuration_descriptor";
     public static final String COMPONENT_DESCRIPTOR_CONFIGURATION_VERSION_PROPERTY = "configuration_version";
     public static final String COMPONENT_DESCRIPTOR_ACTIONS_PROPERTY = "actions";
+    public static final String COMPONENT_DESCRIPTOR_HAS_QUEUE_NAME_PROPERTY = "has_queue_name";
 
     /**
      * Event constants.
@@ -389,6 +390,7 @@ public class ModelConstants {
 
     public static final String DEBUG_MODE = "debug_mode";
     public static final String SINGLETON_MODE = "singleton_mode";
+    public static final String QUEUE_NAME = "queue_name";
 
     /**
      * Rule chain constants.

dao/src/main/java/org/thingsboard/server/dao/model/sql/ComponentDescriptorEntity.java

@@ -70,6 +70,9 @@ public class ComponentDescriptorEntity extends BaseSqlEntity<ComponentDescriptor
     @Column(name = ModelConstants.COMPONENT_DESCRIPTOR_ACTIONS_PROPERTY)
     private String actions;
 
+    @Column(name = ModelConstants.COMPONENT_DESCRIPTOR_HAS_QUEUE_NAME_PROPERTY)
+    private boolean hasQueueName;
+
     public ComponentDescriptorEntity() {
     }
 
@@ -86,6 +89,7 @@ public class ComponentDescriptorEntity extends BaseSqlEntity<ComponentDescriptor
         this.clazz = component.getClazz();
         this.configurationDescriptor = component.getConfigurationDescriptor();
         this.configurationVersion = component.getConfigurationVersion();
+        this.hasQueueName = component.isHasQueueName();
     }
 
     @Override
@@ -100,6 +104,7 @@ public class ComponentDescriptorEntity extends BaseSqlEntity<ComponentDescriptor
         data.setActions(this.getActions());
         data.setConfigurationDescriptor(configurationDescriptor);
         data.setConfigurationVersion(configurationVersion);
+        data.setHasQueueName(hasQueueName);
         return data;
     }
 }

dao/src/main/java/org/thingsboard/server/dao/model/sql/RuleNodeEntity.java

@@ -66,6 +66,9 @@ public class RuleNodeEntity extends BaseSqlEntity<RuleNode> {
     @Column(name = ModelConstants.SINGLETON_MODE)
     private boolean singletonMode;
 
+    @Column(name = ModelConstants.QUEUE_NAME)
+    private String queueName;
+
     @Column(name = ModelConstants.EXTERNAL_ID_PROPERTY)
     private UUID externalId;
 
@@ -84,6 +87,7 @@ public class RuleNodeEntity extends BaseSqlEntity<RuleNode> {
         this.name = ruleNode.getName();
         this.debugMode = ruleNode.isDebugMode();
         this.singletonMode = ruleNode.isSingletonMode();
+        this.queueName = ruleNode.getQueueName();
         this.configurationVersion = ruleNode.getConfigurationVersion();
         this.configuration = ruleNode.getConfiguration();
         this.additionalInfo = ruleNode.getAdditionalInfo();
@@ -103,6 +107,7 @@ public class RuleNodeEntity extends BaseSqlEntity<RuleNode> {
         ruleNode.setName(name);
         ruleNode.setDebugMode(debugMode);
         ruleNode.setSingletonMode(singletonMode);
+        ruleNode.setQueueName(queueName);
         ruleNode.setConfigurationVersion(configurationVersion);
         ruleNode.setConfiguration(configuration);
         ruleNode.setAdditionalInfo(additionalInfo);

dao/src/main/java/org/thingsboard/server/dao/sql/component/AbstractComponentDescriptorInsertRepository.java

@@ -77,7 +77,8 @@ public abstract class AbstractComponentDescriptorInsertRepository implements Com
                 .setParameter("name", entity.getName())
                 .setParameter("scope", entity.getScope().name())
                 .setParameter("type", entity.getType().name())
-                .setParameter("clustering_mode", entity.getClusteringMode().name());
+                .setParameter("clustering_mode", entity.getClusteringMode().name())
+                .setParameter("has_queue_name", entity.isHasQueueName());
     }
 
     private ComponentDescriptorEntity processSaveOrUpdate(ComponentDescriptorEntity entity, String query) {

dao/src/main/java/org/thingsboard/server/dao/sql/component/SqlComponentDescriptorInsertRepository.java

@@ -44,10 +44,10 @@ public class SqlComponentDescriptorInsertRepository extends AbstractComponentDes
     }
 
     private static String getInsertOrUpdateStatement(String conflictKeyStatement, String updateKeyStatement) {
-        return "INSERT INTO component_descriptor (id, created_time, actions, clazz, configuration_descriptor, configuration_version, name, scope, type, clustering_mode) VALUES (:id, :created_time, :actions, :clazz, :configuration_descriptor, :configuration_version, :name, :scope, :type, :clustering_mode) ON CONFLICT " + conflictKeyStatement + " DO UPDATE SET " + updateKeyStatement + " returning *";
+        return "INSERT INTO component_descriptor (id, created_time, actions, clazz, configuration_descriptor, configuration_version, name, scope, type, clustering_mode, has_queue_name) VALUES (:id, :created_time, :actions, :clazz, :configuration_descriptor, :configuration_version, :name, :scope, :type, :clustering_mode, :has_queue_name) ON CONFLICT " + conflictKeyStatement + " DO UPDATE SET " + updateKeyStatement + " returning *";
     }
 
     private static String getUpdateStatement(String id) {
-        return "actions = :actions, " + id + ",created_time = :created_time, configuration_descriptor = :configuration_descriptor, configuration_version = :configuration_version, name = :name, scope = :scope, type = :type, clustering_mode = :clustering_mode";
+        return "actions = :actions, " + id + ",created_time = :created_time, configuration_descriptor = :configuration_descriptor, configuration_version = :configuration_version, name = :name, scope = :scope, type = :type, clustering_mode = :clustering_mode, has_queue_name = :has_queue_name";
     }
 }

dao/src/main/resources/sql/schema-entities.sql

@@ -126,7 +126,8 @@ CREATE TABLE IF NOT EXISTS component_descriptor (
     name varchar(255),
     scope varchar(255),
     type varchar(255),
-    clustering_mode varchar(255)
+    clustering_mode varchar(255),
+    has_queue_name boolean DEFAULT false
 );
 
 CREATE TABLE IF NOT EXISTS customer (
@@ -187,6 +188,7 @@ CREATE TABLE IF NOT EXISTS rule_node (
     name varchar(255),
     debug_mode boolean,
     singleton_mode boolean,
+    queue_name varchar(255),
     external_id uuid
 );
 

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

@@ -42,6 +42,8 @@ public @interface RuleNode {
 
     ComponentClusteringMode clusteringMode() default ComponentClusteringMode.ENABLED;
 
+    boolean hasQueueName() default false;
+
     boolean inEnabled() default true;
 
     boolean outEnabled() default true;

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java

@@ -50,6 +50,7 @@ import static org.thingsboard.common.util.DonAsynchron.withCallback;
         type = ComponentType.ACTION,
         name = "generator",
         configClazz = TbMsgGeneratorNodeConfiguration.class,
+        hasQueueName = true,
         nodeDescription = "Periodically generates messages",
         nodeDetails = "Generates messages with configurable period. Javascript function used for message generation.",
         inEnabled = false,
@@ -69,6 +70,7 @@ public class TbMsgGeneratorNode implements TbNode {
     private UUID nextTickId;
     private TbMsg prevMsg;
     private final AtomicBoolean initialized = new AtomicBoolean(false);
+    private String queueName;
 
     @Override
     public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
@@ -76,6 +78,7 @@ public class TbMsgGeneratorNode implements TbNode {
         this.config = TbNodeUtils.convert(configuration, TbMsgGeneratorNodeConfiguration.class);
         this.delay = TimeUnit.SECONDS.toMillis(config.getPeriodInSeconds());
         this.currentMsgCount = 0;
+        this.queueName = ctx.getSelf().getQueueName();
         if (!StringUtils.isEmpty(config.getOriginatorId())) {
             originatorId = EntityIdFactory.getByTypeAndUuid(config.getOriginatorType(), config.getOriginatorId());
             ctx.checkTenantEntity(originatorId);
@@ -137,7 +140,7 @@ public class TbMsgGeneratorNode implements TbNode {
         }
         lastScheduledTs = lastScheduledTs + delay;
         long curDelay = Math.max(0L, (lastScheduledTs - curTs));
-        TbMsg tickMsg = ctx.newMsg(config.getQueueName(), TbMsgType.GENERATOR_NODE_SELF_MSG, ctx.getSelfId(),
+        TbMsg tickMsg = ctx.newMsg(queueName, TbMsgType.GENERATOR_NODE_SELF_MSG, ctx.getSelfId(),
                 getCustomerIdFromMsg(msg), TbMsgMetaData.EMPTY, TbMsg.EMPTY_STRING);
         nextTickId = tickMsg.getId();
         ctx.tellSelf(tickMsg, curDelay);
@@ -146,14 +149,14 @@ public class TbMsgGeneratorNode implements TbNode {
     private ListenableFuture<TbMsg> generate(TbContext ctx, TbMsg msg) {
         log.trace("generate, config {}", config);
         if (prevMsg == null) {
-            prevMsg = ctx.newMsg(config.getQueueName(), TbMsg.EMPTY_STRING, originatorId, msg.getCustomerId(), TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT);
+            prevMsg = ctx.newMsg(queueName, TbMsg.EMPTY_STRING, originatorId, msg.getCustomerId(), TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT);
         }
         if (initialized.get()) {
             ctx.logJsEvalRequest();
             return Futures.transformAsync(scriptEngine.executeGenerateAsync(prevMsg), generated -> {
                 log.trace("generate process response, generated {}, config {}", generated, config);
                 ctx.logJsEvalResponse();
-                prevMsg = ctx.newMsg(config.getQueueName(), generated.getType(), originatorId, msg.getCustomerId(), generated.getMetaData(), generated.getData());
+                prevMsg = ctx.newMsg(queueName, generated.getType(), originatorId, msg.getCustomerId(), generated.getMetaData(), generated.getData());
                 return Futures.immediateFuture(prevMsg);
             }, MoreExecutors.directExecutor()); //usually it runs on js-executor-remote-callback thread pool
         }

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNodeConfiguration.java

@@ -36,7 +36,6 @@ public class TbMsgGeneratorNodeConfiguration implements NodeConfiguration<TbMsgG
     private ScriptLanguage scriptLang;
     private String jsScript;
     private String tbelScript;
-    private String queueName;
 
     @Override
     public TbMsgGeneratorNodeConfiguration defaultConfiguration() {

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java

@@ -47,6 +47,7 @@ import java.util.concurrent.TimeUnit;
         type = ComponentType.TRANSFORMATION,
         name = "deduplication",
         configClazz = TbMsgDeduplicationNodeConfiguration.class,
+        hasQueueName = true,
         nodeDescription = "Deduplicate messages within the same originator entity for a configurable period " +
                 "based on a specified deduplication strategy.",
         nodeDetails = "Deduplication strategies: <ul><li><strong>FIRST</strong> - return first message that arrived during deduplication period.</li>" +
@@ -66,6 +67,7 @@ public class TbMsgDeduplicationNode implements TbNode {
 
     private final Map<EntityId, DeduplicationData> deduplicationMap;
     private long deduplicationInterval;
+    private String queueName;
 
     public TbMsgDeduplicationNode() {
         this.deduplicationMap = new HashMap<>();
@@ -75,6 +77,7 @@ public class TbMsgDeduplicationNode implements TbNode {
     public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbMsgDeduplicationNodeConfiguration.class);
         this.deduplicationInterval = TimeUnit.SECONDS.toMillis(config.getInterval());
+        this.queueName = ctx.getSelf().getQueueName();
     }
 
     @Override
@@ -132,7 +135,7 @@ public class TbMsgDeduplicationNode implements TbNode {
                         }
                     }
                     deduplicationResults.add(TbMsg.newMsg(
-                            config.getQueueName(),
+                            queueName,
                             config.getOutMsgType(),
                             deduplicationId,
                             getMetadata(),

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNodeConfiguration.java

@@ -26,7 +26,6 @@ public class TbMsgDeduplicationNodeConfiguration implements NodeConfiguration<Tb
 
     // only for DeduplicationStrategy.ALL:
     private String outMsgType;
-    private String queueName;
 
     // Advanced settings:
     private int maxPendingMsgs;

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNode.java

@@ -16,6 +16,7 @@
 package org.thingsboard.rule.engine.flow;
 
 import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.rule.engine.api.EmptyNodeConfiguration;
 import org.thingsboard.rule.engine.api.RuleNode;
 import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.rule.engine.api.TbNode;
@@ -30,11 +31,12 @@ import org.thingsboard.server.common.msg.TbMsg;
 @RuleNode(
         type = ComponentType.FLOW,
         name = "checkpoint",
-        configClazz = TbCheckpointNodeConfiguration.class,
+        configClazz = EmptyNodeConfiguration.class,
+        hasQueueName = true,
         nodeDescription = "transfers the message to another queue",
         nodeDetails = "After successful transfer incoming message is automatically acknowledged. Queue name is configurable.",
         uiResources = {"static/rulenode/rulenode-core-config.js"},
-        configDirective = "tbActionNodeCheckPointConfig"
+        configDirective = "tbNodeEmptyConfig"
 )
 public class TbCheckpointNode implements TbNode {
 
@@ -42,8 +44,7 @@ public class TbCheckpointNode implements TbNode {
 
     @Override
     public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
-        TbCheckpointNodeConfiguration config = TbNodeUtils.convert(configuration, TbCheckpointNodeConfiguration.class);
-        this.queueName = config.getQueueName();
+        this.queueName = ctx.getSelf().getQueueName();
     }
 
     @Override

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNodeConfiguration.java

@@ -1,30 +0,0 @@
-/**
- * Copyright © 2016-2023 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.rule.engine.flow;
-
-import lombok.Data;
-import org.thingsboard.rule.engine.api.NodeConfiguration;
-
-@Data
-public class TbCheckpointNodeConfiguration implements NodeConfiguration<TbCheckpointNodeConfiguration> {
-
-    private String queueName;
-
-    @Override
-    public TbCheckpointNodeConfiguration defaultConfiguration() {
-        return new TbCheckpointNodeConfiguration();
-    }
-}

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbMsgDeduplicationNodeTest.java

@@ -40,6 +40,7 @@ import org.thingsboard.server.common.data.id.RuleNodeId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.msg.TbMsgType;
 import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
+import org.thingsboard.server.common.data.rule.RuleNode;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
 
@@ -241,7 +242,7 @@ public class TbMsgDeduplicationNodeTest {
         config.setInterval(deduplicationInterval);
         config.setStrategy(DeduplicationStrategy.ALL);
         config.setOutMsgType(TbMsgType.POST_ATTRIBUTES_REQUEST.name());
-        config.setQueueName(DataConstants.HP_QUEUE_NAME);
+        setQueueName(DataConstants.HP_QUEUE_NAME);
         nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
         node.init(ctx, nodeConfiguration);
 
@@ -268,7 +269,7 @@ public class TbMsgDeduplicationNodeTest {
         Assertions.assertEquals(getMergedData(inputMsgs), outMessage.getData());
         Assertions.assertEquals(deviceId, outMessage.getOriginator());
         Assertions.assertEquals(config.getOutMsgType(), outMessage.getType());
-        Assertions.assertEquals(config.getQueueName(), outMessage.getQueueName());
+        Assertions.assertEquals(DataConstants.HP_QUEUE_NAME, outMessage.getQueueName());
     }
 
     @Test
@@ -281,7 +282,7 @@ public class TbMsgDeduplicationNodeTest {
         config.setInterval(deduplicationInterval);
         config.setStrategy(DeduplicationStrategy.ALL);
         config.setOutMsgType(TbMsgType.POST_ATTRIBUTES_REQUEST.name());
-        config.setQueueName(DataConstants.HP_QUEUE_NAME);
+        setQueueName(DataConstants.HP_QUEUE_NAME);
         nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
         node.init(ctx, nodeConfiguration);
 
@@ -316,13 +317,13 @@ public class TbMsgDeduplicationNodeTest {
         Assertions.assertEquals(getMergedData(firstMsgPack), firstMsg.getData());
         Assertions.assertEquals(deviceId, firstMsg.getOriginator());
         Assertions.assertEquals(config.getOutMsgType(), firstMsg.getType());
-        Assertions.assertEquals(config.getQueueName(), firstMsg.getQueueName());
+        Assertions.assertEquals(DataConstants.HP_QUEUE_NAME, firstMsg.getQueueName());
 
         TbMsg secondMsg = resultMsgs.get(1);
         Assertions.assertEquals(getMergedData(secondMsgPack), secondMsg.getData());
         Assertions.assertEquals(deviceId, secondMsg.getOriginator());
         Assertions.assertEquals(config.getOutMsgType(), secondMsg.getType());
-        Assertions.assertEquals(config.getQueueName(), secondMsg.getQueueName());
+        Assertions.assertEquals(DataConstants.HP_QUEUE_NAME, secondMsg.getQueueName());
     }
 
     @Test
@@ -429,4 +430,10 @@ public class TbMsgDeduplicationNodeTest {
         return JacksonUtil.toString(mergedData);
     }
 
+    private void setQueueName(String queueName) {
+        RuleNode ruleNode = new RuleNode();
+        ruleNode.setQueueName(queueName);
+        when(ctx.getSelf()).thenReturn(ruleNode);
+    }
+
 }