Merge remote-tracking branch 'origin/fixes/notification-system' into develop/3.5.2

application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@@ -90,7 +90,7 @@ import org.thingsboard.server.dao.widget.WidgetsBundleService;
 import org.thingsboard.server.queue.discovery.DiscoveryService;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
-import org.thingsboard.server.queue.notification.NotificationRuleProcessor;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
 import org.thingsboard.server.queue.util.DataDecodingEncodingService;
 import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
 import org.thingsboard.server.service.component.ComponentDiscoveryService;

application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java

@@ -49,7 +49,6 @@ import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.aware.DeviceAwareMsg;
 import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg;
 import org.thingsboard.server.common.msg.edge.EdgeSessionMsg;
-import org.thingsboard.server.common.msg.notification.trigger.RuleEngineMsgTrigger;
 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
 import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
@@ -211,10 +210,6 @@ public class TenantActor extends RuleChainManagerActor {
             log.trace("[{}] Ack message because Rule Engine is disabled", tenantId);
             tbMsg.getCallback().onSuccess();
         }
-        systemContext.getNotificationRuleProcessor().process(RuleEngineMsgTrigger.builder()
-                .tenantId(tenantId)
-                .msg(tbMsg)
-                .build());
     }
 
     private void onRuleChainMsg(RuleChainAwareMsg msg) {

application/src/main/java/org/thingsboard/server/controller/AlarmCommentController.java

@@ -77,7 +77,7 @@ public class AlarmCommentController extends BaseController {
                                          @PathVariable(ALARM_ID) String strAlarmId, @ApiParam(value = "A JSON value representing the comment.") @RequestBody AlarmComment alarmComment) throws ThingsboardException {
         checkParameter(ALARM_ID, strAlarmId);
         AlarmId alarmId = new AlarmId(toUUID(strAlarmId));
-        Alarm alarm = checkAlarmId(alarmId, Operation.WRITE);
+        Alarm alarm = checkAlarmInfoId(alarmId, Operation.WRITE);
         alarmComment.setAlarmId(alarmId);
         return tbAlarmCommentService.saveAlarmComment(alarm, alarmComment, getCurrentUser());
     }

application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java

@@ -28,7 +28,9 @@ import org.thingsboard.server.common.data.HasName;
 import org.thingsboard.server.common.data.HasTenantId;
 import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.common.data.User;
+import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.alarm.AlarmComment;
+import org.thingsboard.server.common.data.alarm.AlarmInfo;
 import org.thingsboard.server.common.data.audit.ActionType;
 import org.thingsboard.server.common.data.edge.EdgeEventActionType;
 import org.thingsboard.server.common.data.id.CustomerId;
@@ -40,10 +42,12 @@ import org.thingsboard.server.common.data.relation.EntityRelation;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgDataType;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
+import org.thingsboard.server.common.msg.notification.trigger.AlarmAssignmentTrigger;
+import org.thingsboard.server.common.msg.notification.trigger.AlarmCommentTrigger;
 import org.thingsboard.server.common.msg.notification.trigger.EntitiesLimitTrigger;
 import org.thingsboard.server.common.msg.notification.trigger.EntityActionTrigger;
 import org.thingsboard.server.dao.audit.AuditLogService;
-import org.thingsboard.server.queue.notification.NotificationRuleProcessor;
 
 import java.util.List;
 import java.util.Map;
@@ -241,19 +245,7 @@ public class EntityActionService {
                     }
                 }
                 if (tenantId != null && !tenantId.isSysTenantId()) {
-                    if (actionType == ActionType.ADDED) {
+                    processNotificationRules(tenantId, entityId, entity, actionType, user, additionalInfo);
-                        notificationRuleProcessor.process(EntitiesLimitTrigger.builder()
-                                .tenantId(tenantId)
-                                .entityType(entityId.getEntityType())
-                                .build());
-                    }
-                    notificationRuleProcessor.process(EntityActionTrigger.builder()
-                            .tenantId(tenantId)
-                            .entityId(entityId)
-                            .entity(entity)
-                            .actionType(actionType)
-                            .user(user)
-                            .build());
                 }
                 TbMsg tbMsg = TbMsg.newMsg(msgType, entityId, customerId, metaData, TbMsgDataType.JSON, JacksonUtil.toString(entityNode));
                 tbClusterService.pushMsgToRuleEngine(tenantId, entityId, tbMsg, null);
@@ -263,6 +255,53 @@ public class EntityActionService {
         }
     }
 
+    private void processNotificationRules(TenantId tenantId, EntityId entityId, HasName entity, ActionType actionType, User user, Object... additionalInfo) {
+        switch (actionType) {
+            case ADDED:
+                notificationRuleProcessor.process(EntitiesLimitTrigger.builder()
+                        .tenantId(tenantId)
+                        .entityType(entityId.getEntityType())
+                        .build());
+            case UPDATED:
+            case DELETED:
+                notificationRuleProcessor.process(EntityActionTrigger.builder()
+                        .tenantId(tenantId)
+                        .entityId(entityId)
+                        .entity(entity)
+                        .actionType(actionType)
+                        .user(user)
+                        .build());
+                break;
+            case ALARM_ASSIGNED:
+            case ALARM_UNASSIGNED:
+                if (!(entity instanceof AlarmInfo)) { // should not normally happen
+                    log.warn("Invalid alarm assignment event: entity is not instance of AlarmInfo");
+                    break;
+                }
+                notificationRuleProcessor.process(AlarmAssignmentTrigger.builder()
+                        .tenantId(tenantId)
+                        .alarmInfo((AlarmInfo) entity)
+                        .actionType(actionType)
+                        .user(user)
+                        .build());
+                break;
+            case ADDED_COMMENT:
+            case UPDATED_COMMENT:
+                if (!(entity instanceof Alarm)) { // should not normally happen
+                    log.warn("Invalid alarm comment event: entity is not instance of Alarm");
+                    break;
+                }
+                notificationRuleProcessor.process(AlarmCommentTrigger.builder()
+                        .tenantId(tenantId)
+                        .comment(extractParameter(AlarmComment.class, 0, additionalInfo))
+                        .alarm((Alarm) entity)
+                        .actionType(actionType)
+                        .user(user)
+                        .build());
+                break;
+        }
+    }
+
     public <E extends HasName, I extends EntityId> void logEntityAction(User user, I entityId, E entity, CustomerId customerId,
                                                                         ActionType actionType, Exception e, Object... additionalInfo) {
         if (customerId == null || customerId.isNullUid()) {

application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java

@@ -61,7 +61,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceM
 import org.thingsboard.server.gen.transport.TransportProtos.UsageStatsKVProto;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.PartitionService;
-import org.thingsboard.server.queue.notification.NotificationRuleProcessor;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
 import org.thingsboard.server.service.apiusage.BaseApiUsageState.StatsCalculationResult;
 import org.thingsboard.server.service.executors.DbCallbackExecutorService;
 import org.thingsboard.server.service.mail.MailExecutorService;

application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationCenter.java

@@ -15,13 +15,10 @@
  */
 package org.thingsboard.server.service.notification;
 
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
-import org.thingsboard.common.util.DonAsynchron;
 import org.thingsboard.rule.engine.api.NotificationCenter;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.User;
@@ -59,14 +56,12 @@ import org.thingsboard.server.dao.notification.NotificationService;
 import org.thingsboard.server.dao.notification.NotificationSettingsService;
 import org.thingsboard.server.dao.notification.NotificationTargetService;
 import org.thingsboard.server.dao.notification.NotificationTemplateService;
-import org.thingsboard.server.dao.user.UserService;
+import org.thingsboard.server.dao.util.limits.LimitedApi;
+import org.thingsboard.server.dao.util.limits.RateLimitService;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.NotificationsTopicService;
 import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
-import org.thingsboard.server.dao.util.limits.LimitedApi;
-import org.thingsboard.server.dao.util.limits.RateLimitService;
-import org.thingsboard.server.service.executors.DbCallbackExecutorService;
 import org.thingsboard.server.service.executors.NotificationExecutorService;
 import org.thingsboard.server.service.notification.channels.NotificationChannel;
 import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
@@ -74,7 +69,6 @@ import org.thingsboard.server.service.telemetry.AbstractSubscriptionService;
 import org.thingsboard.server.service.ws.notification.sub.NotificationRequestUpdate;
 import org.thingsboard.server.service.ws.notification.sub.NotificationUpdate;
 
-import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
@@ -87,7 +81,7 @@ import java.util.stream.Collectors;
 @Service
 @Slf4j
 @RequiredArgsConstructor
-@SuppressWarnings({"UnstableApiUsage", "rawtypes"})
+@SuppressWarnings({"rawtypes"})
 public class DefaultNotificationCenter extends AbstractSubscriptionService implements NotificationCenter, NotificationChannel<User, WebDeliveryMethodNotificationTemplate> {
 
     private final NotificationTargetService notificationTargetService;
@@ -95,9 +89,7 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
     private final NotificationService notificationService;
     private final NotificationTemplateService notificationTemplateService;
     private final NotificationSettingsService notificationSettingsService;
-    private final UserService userService;
     private final NotificationExecutorService notificationExecutor;
-    private final DbCallbackExecutorService dbCallbackExecutorService;
     private final NotificationsTopicService notificationsTopicService;
     private final TbQueueProducerProvider producerProvider;
     private final RateLimitService rateLimitService;
@@ -172,37 +164,32 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
                 .build();
 
         notificationExecutor.submit(() -> {
-            List<ListenableFuture<Void>> results = new ArrayList<>();
-
             for (NotificationTarget target : targets) {
-                List<ListenableFuture<Void>> result = processForTarget(target, ctx);
+                processForTarget(target, ctx);
-                results.addAll(result);
+            }
+
+            NotificationRequestId requestId = ctx.getRequest().getId();
+            log.debug("[{}] Notification request processing is finished", requestId);
+            NotificationRequestStats stats = ctx.getStats();
+            try {
+                notificationRequestService.updateNotificationRequest(tenantId, requestId, NotificationRequestStatus.SENT, stats);
+            } catch (Exception e) {
+                log.error("[{}] Failed to update stats for notification request", requestId, e);
             }
 
-            Futures.whenAllComplete(results).run(() -> {
+            if (callback != null) {
-                NotificationRequestId requestId = ctx.getRequest().getId();
-                log.debug("[{}] Notification request processing is finished", requestId);
-                NotificationRequestStats stats = ctx.getStats();
                 try {
-                    notificationRequestService.updateNotificationRequest(tenantId, requestId, NotificationRequestStatus.SENT, stats);
+                    callback.accept(stats);
                 } catch (Exception e) {
-                    log.error("[{}] Failed to update stats for notification request", requestId, e);
+                    log.error("Failed to process callback for notification request {}", requestId, e);
-                }
-
-                if (callback != null) {
-                    try {
-                        callback.accept(stats);
-                    } catch (Exception e) {
-                        log.error("Failed to process callback for notification request {}", requestId, e);
-                    }
                 }
-            }, dbCallbackExecutorService);
+            }
         });
 
         return request;
     }
 
-    private List<ListenableFuture<Void>> processForTarget(NotificationTarget target, NotificationProcessingContext ctx) {
+    private void processForTarget(NotificationTarget target, NotificationProcessingContext ctx) {
         Iterable<? extends NotificationRecipient> recipients;
         switch (target.getConfiguration().getType()) {
             case PLATFORM_USERS: {
@@ -231,43 +218,35 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
         Set<NotificationDeliveryMethod> deliveryMethods = new HashSet<>(ctx.getDeliveryMethods());
         deliveryMethods.removeIf(deliveryMethod -> !target.getConfiguration().getType().getSupportedDeliveryMethods().contains(deliveryMethod));
         log.debug("[{}] Processing notification request for {} target ({}) for delivery methods {}", ctx.getRequest().getId(), target.getConfiguration().getType(), target.getId(), deliveryMethods);
+        if (deliveryMethods.isEmpty()) {
+            return;
+        }
 
-        List<ListenableFuture<Void>> results = new ArrayList<>();
+        for (NotificationRecipient recipient : recipients) {
-        if (!deliveryMethods.isEmpty()) {
+            for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) {
-            for (NotificationRecipient recipient : recipients) {
+                try {
-                for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) {
+                    processForRecipient(deliveryMethod, recipient, ctx);
-                    ListenableFuture<Void> resultFuture = processForRecipient(deliveryMethod, recipient, ctx);
+                    ctx.getStats().reportSent(deliveryMethod, recipient);
-                    DonAsynchron.withCallback(resultFuture, result -> {
+                } catch (Exception error) {
-                        ctx.getStats().reportSent(deliveryMethod, recipient);
+                    ctx.getStats().reportError(deliveryMethod, error, recipient);
-                    }, error -> {
-                        ctx.getStats().reportError(deliveryMethod, error, recipient);
-                    });
-                    results.add(resultFuture);
                 }
             }
         }
-        return results;
     }
 
-    private ListenableFuture<Void> processForRecipient(NotificationDeliveryMethod deliveryMethod, NotificationRecipient recipient, NotificationProcessingContext ctx) {
+    private void processForRecipient(NotificationDeliveryMethod deliveryMethod, NotificationRecipient recipient, NotificationProcessingContext ctx) throws Exception {
         if (ctx.getStats().contains(deliveryMethod, recipient.getId())) {
-            return Futures.immediateFailedFuture(new AlreadySentException());
+            throw new AlreadySentException();
         }
-
-        DeliveryMethodNotificationTemplate processedTemplate;
-        try {
-            processedTemplate = ctx.getProcessedTemplate(deliveryMethod, recipient);
-        } catch (Exception e) {
-            return Futures.immediateFailedFuture(e);
-        }
-
         NotificationChannel notificationChannel = channels.get(deliveryMethod);
+        DeliveryMethodNotificationTemplate processedTemplate = ctx.getProcessedTemplate(deliveryMethod, recipient);
+
         log.trace("[{}] Sending {} notification for recipient {}", ctx.getRequest().getId(), deliveryMethod, recipient);
-        return notificationChannel.sendNotification(recipient, processedTemplate, ctx);
+        notificationChannel.sendNotification(recipient, processedTemplate, ctx);
     }
 
     @Override
-    public ListenableFuture<Void> sendNotification(User recipient, WebDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) {
+    public void sendNotification(User recipient, WebDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) throws Exception {
         NotificationRequest request = ctx.getRequest();
         Notification notification = Notification.builder()
                 .requestId(request.getId())
@@ -283,14 +262,14 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
             notification = notificationService.saveNotification(recipient.getTenantId(), notification);
         } catch (Exception e) {
             log.error("Failed to create notification for recipient {}", recipient.getId(), e);
-            return Futures.immediateFailedFuture(e);
+            throw e;
         }
 
         NotificationUpdate update = NotificationUpdate.builder()
                 .created(true)
                 .notification(notification)
                 .build();
-        return onNotificationUpdate(recipient.getTenantId(), recipient.getId(), update);
+        onNotificationUpdate(recipient.getTenantId(), recipient.getId(), update);
     }
 
     @Override
@@ -384,13 +363,11 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
         clusterService.pushMsgToCore(tenantId, notificationRequestId, toCoreMsg, null);
     }
 
-    private ListenableFuture<Void> onNotificationUpdate(TenantId tenantId, UserId recipientId, NotificationUpdate update) {
+    private void onNotificationUpdate(TenantId tenantId, UserId recipientId, NotificationUpdate update) {
         log.trace("Submitting notification update for recipient {}: {}", recipientId, update);
-        return Futures.submit(() -> {
+        forwardToSubscriptionManagerService(tenantId, recipientId, subscriptionManagerService -> {
-            forwardToSubscriptionManagerService(tenantId, recipientId, subscriptionManagerService -> {
+            subscriptionManagerService.onNotificationUpdate(tenantId, recipientId, update, TbCallback.EMPTY);
-                subscriptionManagerService.onNotificationUpdate(tenantId, recipientId, update, TbCallback.EMPTY);
+        }, () -> TbSubscriptionUtils.notificationUpdateToProto(tenantId, recipientId, update));
-            }, () -> TbSubscriptionUtils.notificationUpdateToProto(tenantId, recipientId, update));
-        }, wsCallBackExecutor);
     }
 
     private void onNotificationRequestUpdate(TenantId tenantId, NotificationRequestUpdate update) {

application/src/main/java/org/thingsboard/server/service/notification/channels/EmailNotificationChannel.java

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.server.service.notification.channels;
 
-import com.google.common.util.concurrent.ListenableFuture;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;
 import org.thingsboard.rule.engine.api.MailService;
@@ -24,7 +23,6 @@ import org.thingsboard.server.common.data.User;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod;
 import org.thingsboard.server.common.data.notification.template.EmailDeliveryMethodNotificationTemplate;
-import org.thingsboard.server.service.mail.MailExecutorService;
 import org.thingsboard.server.service.notification.NotificationProcessingContext;
 
 @Component
@@ -32,19 +30,15 @@ import org.thingsboard.server.service.notification.NotificationProcessingContext
 public class EmailNotificationChannel implements NotificationChannel<User, EmailDeliveryMethodNotificationTemplate> {
 
     private final MailService mailService;
-    private final MailExecutorService executor;
 
     @Override
-    public ListenableFuture<Void> sendNotification(User recipient, EmailDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) {
+    public void sendNotification(User recipient, EmailDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) throws Exception {
-        return executor.submit(() -> {
+        mailService.send(recipient.getTenantId(), null, TbEmail.builder()
-            mailService.send(recipient.getTenantId(), null, TbEmail.builder()
+                .to(recipient.getEmail())
-                    .to(recipient.getEmail())
+                .subject(processedTemplate.getSubject())
-                    .subject(processedTemplate.getSubject())
+                .body(processedTemplate.getBody())
-                    .body(processedTemplate.getBody())
+                .html(true)
-                    .html(true)
+                .build());
-                    .build());
-            return null;
-        });
     }
 
     @Override

application/src/main/java/org/thingsboard/server/service/notification/channels/NotificationChannel.java

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.server.service.notification.channels;
 
-import com.google.common.util.concurrent.ListenableFuture;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod;
 import org.thingsboard.server.common.data.notification.targets.NotificationRecipient;
@@ -24,7 +23,7 @@ import org.thingsboard.server.service.notification.NotificationProcessingContext
 
 public interface NotificationChannel<R extends NotificationRecipient, T extends DeliveryMethodNotificationTemplate> {
 
-    ListenableFuture<Void> sendNotification(R recipient, T processedTemplate, NotificationProcessingContext ctx);
+    void sendNotification(R recipient, T processedTemplate, NotificationProcessingContext ctx) throws Exception;
 
     void check(TenantId tenantId) throws Exception;
 

application/src/main/java/org/thingsboard/server/service/notification/channels/SlackNotificationChannel.java

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.server.service.notification.channels;
 
-import com.google.common.util.concurrent.ListenableFuture;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;
 import org.thingsboard.rule.engine.api.slack.SlackService;
@@ -26,7 +25,6 @@ import org.thingsboard.server.common.data.notification.settings.SlackNotificatio
 import org.thingsboard.server.common.data.notification.targets.slack.SlackConversation;
 import org.thingsboard.server.common.data.notification.template.SlackDeliveryMethodNotificationTemplate;
 import org.thingsboard.server.dao.notification.NotificationSettingsService;
-import org.thingsboard.server.service.executors.ExternalCallExecutorService;
 import org.thingsboard.server.service.notification.NotificationProcessingContext;
 
 @Component
@@ -35,15 +33,11 @@ public class SlackNotificationChannel implements NotificationChannel<SlackConver
 
     private final SlackService slackService;
     private final NotificationSettingsService notificationSettingsService;
-    private final ExternalCallExecutorService executor;
 
     @Override
-    public ListenableFuture<Void> sendNotification(SlackConversation conversation, SlackDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) {
+    public void sendNotification(SlackConversation conversation, SlackDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) throws Exception {
         SlackNotificationDeliveryMethodConfig config = ctx.getDeliveryMethodConfig(NotificationDeliveryMethod.SLACK);
-        return executor.submit(() -> {
+        slackService.sendMessage(ctx.getTenantId(), config.getBotToken(), conversation.getId(), processedTemplate.getBody());
-            slackService.sendMessage(ctx.getTenantId(), config.getBotToken(), conversation.getId(), processedTemplate.getBody());
-            return null;
-        });
     }
 
     @Override

application/src/main/java/org/thingsboard/server/service/notification/channels/SmsNotificationChannel.java

@@ -15,8 +15,6 @@
  */
 package org.thingsboard.server.service.notification.channels;
 
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Component;
@@ -26,26 +24,21 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod;
 import org.thingsboard.server.common.data.notification.template.SmsDeliveryMethodNotificationTemplate;
 import org.thingsboard.server.service.notification.NotificationProcessingContext;
-import org.thingsboard.server.service.sms.SmsExecutorService;
 
 @Component
 @RequiredArgsConstructor
 public class SmsNotificationChannel implements NotificationChannel<User, SmsDeliveryMethodNotificationTemplate> {
 
     private final SmsService smsService;
-    private final SmsExecutorService executor;
 
     @Override
-    public ListenableFuture<Void> sendNotification(User recipient, SmsDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) {
+    public void sendNotification(User recipient, SmsDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) throws Exception {
         String phone = recipient.getPhone();
         if (StringUtils.isBlank(phone)) {
-            return Futures.immediateFailedFuture(new RuntimeException("User does not have phone number"));
+            throw new RuntimeException("User does not have phone number");
         }
 
-        return executor.submit(() -> {
+        smsService.sendSms(recipient.getTenantId(), recipient.getCustomerId(), new String[]{phone}, processedTemplate.getBody());
-            smsService.sendSms(recipient.getTenantId(), recipient.getCustomerId(), new String[]{phone}, processedTemplate.getBody());
-            return null;
-        });
     }
 
     @Override

application/src/main/java/org/thingsboard/server/service/notification/rule/DefaultNotificationRuleProcessor.java

@@ -15,10 +15,11 @@
  */
 package org.thingsboard.server.service.notification.rule;
 
-import lombok.Data;
 import lombok.RequiredArgsConstructor;
+import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.cache.Cache;
 import org.springframework.cache.CacheManager;
 import org.springframework.context.annotation.Lazy;
@@ -38,34 +39,33 @@ import org.thingsboard.server.common.data.notification.info.NotificationInfo;
 import org.thingsboard.server.common.data.notification.rule.NotificationRule;
 import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerConfig;
 import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
+import org.thingsboard.server.common.data.notification.settings.TriggerTypeConfig;
 import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
 import org.thingsboard.server.common.msg.notification.trigger.NotificationRuleTrigger;
-import org.thingsboard.server.common.msg.notification.trigger.RuleEngineMsgTrigger;
 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.dao.notification.NotificationRequestService;
 import org.thingsboard.server.dao.util.limits.LimitedApi;
 import org.thingsboard.server.dao.util.limits.RateLimitService;
 import org.thingsboard.server.queue.discovery.PartitionService;
-import org.thingsboard.server.queue.notification.NotificationRuleProcessor;
 import org.thingsboard.server.service.executors.NotificationExecutorService;
 import org.thingsboard.server.service.notification.rule.cache.NotificationRulesCache;
 import org.thingsboard.server.service.notification.rule.trigger.NotificationRuleTriggerProcessor;
-import org.thingsboard.server.service.notification.rule.trigger.RuleEngineMsgNotificationRuleTriggerProcessor;
 
 import javax.annotation.PostConstruct;
-import java.io.Serializable;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.EnumMap;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.stream.Collectors;
 
 @Service
 @RequiredArgsConstructor
+@ConfigurationProperties(prefix = "notification-system.rules")
 @Slf4j
 @SuppressWarnings({"rawtypes", "unchecked"})
 public class DefaultNotificationRuleProcessor implements NotificationRuleProcessor {
@@ -79,6 +79,8 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
     private final NotificationExecutorService notificationExecutor;
     private final CacheManager cacheManager;
     private Cache sentNotifications;
+    @Setter
+    private Map<NotificationRuleTriggerType, TriggerTypeConfig> triggerTypesConfigs;
 
     private final Map<NotificationRuleTriggerType, NotificationRuleTriggerProcessor> triggerProcessors = new EnumMap<>(NotificationRuleTriggerType.class);
 
@@ -93,20 +95,27 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
     @Override
     public void process(NotificationRuleTrigger trigger) {
         NotificationRuleTriggerType triggerType = trigger.getType();
-        if (triggerType == null) return;
         TenantId tenantId = triggerType.isTenantLevel() ? trigger.getTenantId() : TenantId.SYS_TENANT_ID;
 
         try {
-            List<NotificationRule> rules = notificationRulesCache.getEnabled(tenantId, triggerType);
+            List<NotificationRule> enabledRules = notificationRulesCache.getEnabled(tenantId, triggerType);
-            for (NotificationRule rule : rules) {
+            if (enabledRules.isEmpty()) {
-                notificationExecutor.submit(() -> {
+                return;
+            }
+            if (trigger.deduplicate()) {
+                enabledRules = new ArrayList<>(enabledRules);
+                enabledRules.removeIf(rule -> alreadySent(rule, trigger));
+            }
+            final List<NotificationRule> rules = enabledRules;
+            notificationExecutor.submit(() -> {
+                for (NotificationRule rule : rules) {
                     try {
                         processNotificationRule(rule, trigger);
                     } catch (Throwable e) {
                         log.error("Failed to process notification rule {} for trigger type {} with trigger object {}", rule.getId(), rule.getTriggerType(), trigger, e);
                     }
-                });
+                }
-            }
+            });
         } catch (Throwable e) {
             log.error("Failed to process notification rules for trigger: {}", trigger, e);
         }
@@ -142,9 +151,6 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
                 log.debug("[{}] Rate limit for notification requests per rule was exceeded (rule '{}')", rule.getTenantId(), rule.getName());
                 return;
             }
-            if (trigger.getType().isDeduplicate() && alreadySent(rule.getId(), trigger)) {
-                return;
-            }
 
             NotificationInfo notificationInfo = constructNotificationInfo(trigger, triggerConfig);
             rule.getRecipientsConfig().getTargetsTable().forEach((delay, targets) -> {
@@ -172,14 +178,13 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
                 .ruleId(rule.getId())
                 .originatorEntityId(originatorEntityId)
                 .build();
-        notificationExecutor.submit(() -> {
+
-            try {
+        try {
-                log.debug("Submitting notification request for rule '{}' with delay of {} sec to targets {}", rule.getName(), delayInSec, targets);
+            log.debug("Submitting notification request for rule '{}' with delay of {} sec to targets {}", rule.getName(), delayInSec, targets);
-                notificationCenter.processNotificationRequest(rule.getTenantId(), notificationRequest, null);
+            notificationCenter.processNotificationRequest(rule.getTenantId(), notificationRequest, null);
-            } catch (Exception e) {
+        } catch (Exception e) {
-                log.error("Failed to process notification request for tenant {} for rule {}", rule.getTenantId(), rule.getId(), e);
+            log.error("Failed to process notification request for tenant {} for rule {}", rule.getTenantId(), rule.getId(), e);
-            }
+        }
-        });
     }
 
     private boolean matchesFilter(NotificationRuleTrigger trigger, NotificationRuleTriggerConfig triggerConfig) {
@@ -194,23 +199,34 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
         return triggerProcessors.get(triggerConfig.getTriggerType()).constructNotificationInfo(trigger);
     }
 
-    private boolean alreadySent(NotificationRuleId ruleId, NotificationRuleTrigger trigger) {
+    private boolean alreadySent(NotificationRule rule, NotificationRuleTrigger trigger) {
-        String key = ruleId + "_" + trigger.getOriginatorEntityId();
+        String deduplicationKey = getDeduplicationKey(trigger, rule);
-        SentNotification sent = sentNotifications.get(key, SentNotification.class);
+
-        boolean alreadySent;
+        boolean alreadySent = false;
-        if (sent != null && sent.getTrigger().equals(trigger)) {
+        Long lastSentTs = sentNotifications.get(deduplicationKey, Long.class);
-            alreadySent = true;
+        if (lastSentTs != null) {
-            log.debug("Notification for {} trigger was already sent, ignoring", trigger.getType());
+            long deduplicationDuration = Optional.ofNullable(triggerTypesConfigs)
-            // updating cache anyway so that the value is not removed by ttl
+                    .map(triggerTypes -> triggerTypes.get(trigger.getType()))
-        } else {
+                    .map(TriggerTypeConfig::getDeduplicationDuration)
-            alreadySent = false;
+                    .orElseGet(trigger::getDefaultDeduplicationDuration);
-            sent = new SentNotification(trigger);
+            long passed = System.currentTimeMillis() - lastSentTs;
+            log.trace("Deduplicating trigger {} for rule '{}' by key '{}'. Deduplication duration: {} ms, passed: {} ms",
+                    trigger.getType(), rule.getName(), deduplicationKey, deduplicationDuration, passed);
+            if (deduplicationDuration == 0 || passed <= deduplicationDuration) {
+                alreadySent = true;
+            }
+        }
+        if (!alreadySent) {
+            lastSentTs = System.currentTimeMillis();
         }
-        log.trace("[{}] Putting to sentNotifications cache: {}", ruleId, trigger);
+        sentNotifications.put(deduplicationKey, lastSentTs);
-        sentNotifications.put(key, sent);
         return alreadySent;
     }
 
+    public static String getDeduplicationKey(NotificationRuleTrigger trigger, NotificationRule rule) {
+        return String.join("_", trigger.getDeduplicationKey(), rule.getDeduplicationKey());
+    }
+
     @EventListener(ComponentLifecycleMsg.class)
     public void onNotificationRuleDeleted(ComponentLifecycleMsg componentLifecycleMsg) {
         if (componentLifecycleMsg.getEvent() != ComponentLifecycleEvent.DELETED ||
@@ -232,24 +248,9 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
 
     @Autowired
     public void setTriggerProcessors(Collection<NotificationRuleTriggerProcessor> processors) {
-        Map<String, NotificationRuleTriggerType> ruleEngineMsgTypeToTriggerType = new HashMap<>();
         processors.forEach(processor -> {
             triggerProcessors.put(processor.getTriggerType(), processor);
-            if (processor instanceof RuleEngineMsgNotificationRuleTriggerProcessor) {
-                Set<String> supportedMsgTypes = ((RuleEngineMsgNotificationRuleTriggerProcessor<?>) processor).getSupportedMsgTypes();
-                supportedMsgTypes.forEach(supportedMsgType -> {
-                    ruleEngineMsgTypeToTriggerType.put(supportedMsgType, processor.getTriggerType());
-                });
-            }
         });
-        RuleEngineMsgTrigger.msgTypeToTriggerType = ruleEngineMsgTypeToTriggerType;
-    }
-
-    @Data
-    private static class SentNotification implements Serializable {
-        private static final long serialVersionUID = 38973480405095422L;
-
-        private final NotificationRuleTrigger trigger;
     }
 
 }

application/src/main/java/org/thingsboard/server/service/notification/rule/cache/DefaultNotificationRulesCache.java

@@ -17,7 +17,6 @@ package org.thingsboard.server.service.notification.rule.cache;
 
 import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
-import lombok.Data;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
@@ -50,7 +49,7 @@ public class DefaultNotificationRulesCache implements NotificationRulesCache {
     private int cacheMaxSize;
     @Value("${cache.notificationRules.timeToLiveInMinutes:30}")
     private int cacheValueTtl;
-    private Cache<CacheKey, List<NotificationRule>> cache;
+    private Cache<String, List<NotificationRule>> cache;
 
     private final ReadWriteLock lock = new ReentrantReadWriteLock();
 
@@ -96,21 +95,15 @@ public class DefaultNotificationRulesCache implements NotificationRulesCache {
         }
     }
 
-    private void evict(TenantId tenantId) {
+    public void evict(TenantId tenantId) {
         cache.invalidateAll(Arrays.stream(NotificationRuleTriggerType.values())
                 .map(triggerType -> key(tenantId, triggerType))
                 .collect(Collectors.toList()));
         log.trace("Evicted all notification rules for tenant {} from cache", tenantId);
     }
 
-    private static CacheKey key(TenantId tenantId, NotificationRuleTriggerType triggerType) {
+    private static String key(TenantId tenantId, NotificationRuleTriggerType triggerType) {
-        return new CacheKey(tenantId, triggerType);
+        return tenantId + "_" + triggerType;
-    }
-
-    @Data
-    private static class CacheKey {
-        private final TenantId tenantId;
-        private final NotificationRuleTriggerType triggerType;
     }
 
 }

application/src/main/java/org/thingsboard/server/service/notification/rule/trigger/AlarmAssignmentTriggerProcessor.java

@@ -16,52 +16,48 @@
 package org.thingsboard.server.service.notification.rule.trigger;
 
 import org.springframework.stereotype.Service;
-import org.thingsboard.common.util.JacksonUtil;
-import org.thingsboard.server.common.data.DataConstants;
-import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.alarm.AlarmAssignee;
 import org.thingsboard.server.common.data.alarm.AlarmInfo;
 import org.thingsboard.server.common.data.alarm.AlarmStatusFilter;
+import org.thingsboard.server.common.data.audit.ActionType;
 import org.thingsboard.server.common.data.notification.info.AlarmAssignmentNotificationInfo;
 import org.thingsboard.server.common.data.notification.info.RuleOriginatedNotificationInfo;
 import org.thingsboard.server.common.data.notification.rule.trigger.AlarmAssignmentNotificationRuleTriggerConfig;
 import org.thingsboard.server.common.data.notification.rule.trigger.AlarmAssignmentNotificationRuleTriggerConfig.Action;
 import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
-import org.thingsboard.server.common.msg.notification.trigger.RuleEngineMsgTrigger;
+import org.thingsboard.server.common.msg.notification.trigger.AlarmAssignmentTrigger;
-
-import java.util.Set;
 
 import static org.apache.commons.collections.CollectionUtils.isEmpty;
 import static org.thingsboard.server.common.data.util.CollectionsUtil.emptyOrContains;
 
 @Service
-public class AlarmAssignmentTriggerProcessor implements RuleEngineMsgNotificationRuleTriggerProcessor<AlarmAssignmentNotificationRuleTriggerConfig> {
+public class AlarmAssignmentTriggerProcessor implements NotificationRuleTriggerProcessor<AlarmAssignmentTrigger, AlarmAssignmentNotificationRuleTriggerConfig> {
 
     @Override
-    public boolean matchesFilter(RuleEngineMsgTrigger trigger, AlarmAssignmentNotificationRuleTriggerConfig triggerConfig) {
+    public boolean matchesFilter(AlarmAssignmentTrigger trigger, AlarmAssignmentNotificationRuleTriggerConfig triggerConfig) {
-        Action action = trigger.getMsg().getType().equals(DataConstants.ALARM_ASSIGNED) ? Action.ASSIGNED : Action.UNASSIGNED;
+        Action action = trigger.getActionType() == ActionType.ALARM_ASSIGNED ? Action.ASSIGNED : Action.UNASSIGNED;
         if (!triggerConfig.getNotifyOn().contains(action)) {
             return false;
         }
-        Alarm alarm = JacksonUtil.fromString(trigger.getMsg().getData(), Alarm.class);
+        AlarmInfo alarmInfo = trigger.getAlarmInfo();
-        return emptyOrContains(triggerConfig.getAlarmTypes(), alarm.getType()) &&
+        return emptyOrContains(triggerConfig.getAlarmTypes(), alarmInfo.getType()) &&
-                emptyOrContains(triggerConfig.getAlarmSeverities(), alarm.getSeverity()) &&
+                emptyOrContains(triggerConfig.getAlarmSeverities(), alarmInfo.getSeverity()) &&
-                (isEmpty(triggerConfig.getAlarmStatuses()) || AlarmStatusFilter.from(triggerConfig.getAlarmStatuses()).matches(alarm));
+                (isEmpty(triggerConfig.getAlarmStatuses()) || AlarmStatusFilter.from(triggerConfig.getAlarmStatuses()).matches(alarmInfo));
     }
 
     @Override
-    public RuleOriginatedNotificationInfo constructNotificationInfo(RuleEngineMsgTrigger trigger) {
+    public RuleOriginatedNotificationInfo constructNotificationInfo(AlarmAssignmentTrigger trigger) {
-        AlarmInfo alarmInfo = JacksonUtil.fromString(trigger.getMsg().getData(), AlarmInfo.class);
+        AlarmInfo alarmInfo = trigger.getAlarmInfo();
         AlarmAssignee assignee = alarmInfo.getAssignee();
         return AlarmAssignmentNotificationInfo.builder()
-                .action(trigger.getMsg().getType().equals(DataConstants.ALARM_ASSIGNED) ? "assigned" : "unassigned")
+                .action(trigger.getActionType() == ActionType.ALARM_ASSIGNED ? "assigned" : "unassigned")
                 .assigneeFirstName(assignee != null ? assignee.getFirstName() : null)
                 .assigneeLastName(assignee != null ? assignee.getLastName() : null)
                 .assigneeEmail(assignee != null ? assignee.getEmail() : null)
                 .assigneeId(assignee != null ? assignee.getId() : null)
-                .userEmail(trigger.getMsg().getMetaData().getValue("userEmail"))
+                .userEmail(trigger.getUser().getEmail())
-                .userFirstName(trigger.getMsg().getMetaData().getValue("userFirstName"))
+                .userFirstName(trigger.getUser().getFirstName())
-                .userLastName(trigger.getMsg().getMetaData().getValue("userLastName"))
+                .userLastName(trigger.getUser().getLastName())
                 .alarmId(alarmInfo.getUuidId())
                 .alarmType(alarmInfo.getType())
                 .alarmOriginator(alarmInfo.getOriginator())
@@ -77,9 +73,4 @@ public class AlarmAssignmentTriggerProcessor implements RuleEngineMsgNotificatio
         return NotificationRuleTriggerType.ALARM_ASSIGNMENT;
     }
 
-    @Override
-    public Set<String> getSupportedMsgTypes() {
-        return Set.of(DataConstants.ALARM_ASSIGNED, DataConstants.ALARM_UNASSIGNED);
-    }
-
 }

application/src/main/java/org/thingsboard/server/service/notification/rule/trigger/AlarmCommentTriggerProcessor.java

@@ -15,68 +15,67 @@
  */
 package org.thingsboard.server.service.notification.rule.trigger;
 
+import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
-import org.thingsboard.common.util.JacksonUtil;
-import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.alarm.Alarm;
-import org.thingsboard.server.common.data.alarm.AlarmComment;
 import org.thingsboard.server.common.data.alarm.AlarmCommentType;
 import org.thingsboard.server.common.data.alarm.AlarmInfo;
 import org.thingsboard.server.common.data.alarm.AlarmStatusFilter;
+import org.thingsboard.server.common.data.audit.ActionType;
 import org.thingsboard.server.common.data.notification.info.AlarmCommentNotificationInfo;
 import org.thingsboard.server.common.data.notification.info.RuleOriginatedNotificationInfo;
 import org.thingsboard.server.common.data.notification.rule.trigger.AlarmCommentNotificationRuleTriggerConfig;
 import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
-import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.notification.trigger.AlarmCommentTrigger;
-import org.thingsboard.server.common.msg.notification.trigger.RuleEngineMsgTrigger;
+import org.thingsboard.server.dao.entity.EntityService;
-
-import java.util.Set;
 
 import static org.apache.commons.collections.CollectionUtils.isEmpty;
 import static org.thingsboard.server.common.data.util.CollectionsUtil.emptyOrContains;
 
 @Service
-public class AlarmCommentTriggerProcessor implements RuleEngineMsgNotificationRuleTriggerProcessor<AlarmCommentNotificationRuleTriggerConfig> {
+@RequiredArgsConstructor
+public class AlarmCommentTriggerProcessor implements NotificationRuleTriggerProcessor<AlarmCommentTrigger, AlarmCommentNotificationRuleTriggerConfig> {
+
+    private final EntityService entityService;
 
     @Override
-    public boolean matchesFilter(RuleEngineMsgTrigger trigger, AlarmCommentNotificationRuleTriggerConfig triggerConfig) {
+    public boolean matchesFilter(AlarmCommentTrigger trigger, AlarmCommentNotificationRuleTriggerConfig triggerConfig) {
-        TbMsg msg = trigger.getMsg();
+        if (trigger.getActionType() == ActionType.UPDATED_COMMENT && !triggerConfig.isNotifyOnCommentUpdate()) {
-        if (msg.getMetaData().getValue("comment") == null) {
-            return false;
-        }
-        if (msg.getType().equals(DataConstants.COMMENT_UPDATED) && !triggerConfig.isNotifyOnCommentUpdate()) {
             return false;
         }
         if (triggerConfig.isOnlyUserComments()) {
-            AlarmComment comment = JacksonUtil.fromString(msg.getMetaData().getValue("comment"), AlarmComment.class);
+            if (trigger.getComment().getType() == AlarmCommentType.SYSTEM) {
-            if (comment.getType() == AlarmCommentType.SYSTEM) {
                 return false;
             }
         }
-        Alarm alarm = JacksonUtil.fromString(msg.getData(), Alarm.class);
+        Alarm alarm = trigger.getAlarm();
         return emptyOrContains(triggerConfig.getAlarmTypes(), alarm.getType()) &&
                 emptyOrContains(triggerConfig.getAlarmSeverities(), alarm.getSeverity()) &&
                 (isEmpty(triggerConfig.getAlarmStatuses()) || AlarmStatusFilter.from(triggerConfig.getAlarmStatuses()).matches(alarm));
     }
 
     @Override
-    public RuleOriginatedNotificationInfo constructNotificationInfo(RuleEngineMsgTrigger trigger) {
+    public RuleOriginatedNotificationInfo constructNotificationInfo(AlarmCommentTrigger trigger) {
-        TbMsg msg = trigger.getMsg();
+        Alarm alarm = trigger.getAlarm();
-        AlarmComment comment = JacksonUtil.fromString(msg.getMetaData().getValue("comment"), AlarmComment.class);
+        String originatorName;
-        AlarmInfo alarmInfo = JacksonUtil.fromString(msg.getData(), AlarmInfo.class);
+        if (alarm instanceof AlarmInfo) {
+            originatorName = ((AlarmInfo) alarm).getOriginatorName();
+        } else {
+            originatorName = entityService.fetchEntityName(trigger.getTenantId(), alarm.getOriginator()).orElse("");
+        }
         return AlarmCommentNotificationInfo.builder()
-                .comment(comment.getComment().get("text").asText())
+                .comment(trigger.getComment().getComment().get("text").asText())
-                .action(msg.getType().equals(DataConstants.COMMENT_CREATED) ? "added" : "updated")
+                .action(trigger.getActionType() == ActionType.ADDED_COMMENT ? "added" : "updated")
-                .userEmail(msg.getMetaData().getValue("userEmail"))
+                .userEmail(trigger.getUser().getEmail())
-                .userFirstName(msg.getMetaData().getValue("userFirstName"))
+                .userFirstName(trigger.getUser().getFirstName())
-                .userLastName(msg.getMetaData().getValue("userLastName"))
+                .userLastName(trigger.getUser().getLastName())
-                .alarmId(alarmInfo.getUuidId())
+                .alarmId(alarm.getUuidId())
-                .alarmType(alarmInfo.getType())
+                .alarmType(alarm.getType())
-                .alarmOriginator(alarmInfo.getOriginator())
+                .alarmOriginator(alarm.getOriginator())
-                .alarmOriginatorName(alarmInfo.getOriginatorName())
+                .alarmOriginatorName(originatorName)
-                .alarmSeverity(alarmInfo.getSeverity())
+                .alarmSeverity(alarm.getSeverity())
-                .alarmStatus(alarmInfo.getStatus())
+                .alarmStatus(alarm.getStatus())
-                .alarmCustomerId(alarmInfo.getCustomerId())
+                .alarmCustomerId(alarm.getCustomerId())
                 .build();
     }
 
@@ -85,9 +84,4 @@ public class AlarmCommentTriggerProcessor implements RuleEngineMsgNotificationRu
         return NotificationRuleTriggerType.ALARM_COMMENT;
     }
 
-    @Override
-    public Set<String> getSupportedMsgTypes() {
-        return Set.of(DataConstants.COMMENT_CREATED, DataConstants.COMMENT_UPDATED);
-    }
-
 }

application/src/main/java/org/thingsboard/server/service/notification/rule/trigger/DeviceActivityTriggerProcessor.java

@@ -18,9 +18,7 @@ package org.thingsboard.server.service.notification.rule.trigger;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.collections.CollectionUtils;
 import org.springframework.stereotype.Service;
-import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.DeviceProfile;
-import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.notification.info.DeviceActivityNotificationInfo;
@@ -28,28 +26,22 @@ import org.thingsboard.server.common.data.notification.info.RuleOriginatedNotifi
 import org.thingsboard.server.common.data.notification.rule.trigger.DeviceActivityNotificationRuleTriggerConfig;
 import org.thingsboard.server.common.data.notification.rule.trigger.DeviceActivityNotificationRuleTriggerConfig.DeviceEvent;
 import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
-import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.notification.trigger.DeviceActivityTrigger;
-import org.thingsboard.server.common.msg.notification.trigger.RuleEngineMsgTrigger;
 import org.thingsboard.server.service.profile.TbDeviceProfileCache;
 
-import java.util.Set;
-
 @Service
 @RequiredArgsConstructor
-public class DeviceActivityTriggerProcessor implements RuleEngineMsgNotificationRuleTriggerProcessor<DeviceActivityNotificationRuleTriggerConfig> {
+public class DeviceActivityTriggerProcessor implements NotificationRuleTriggerProcessor<DeviceActivityTrigger, DeviceActivityNotificationRuleTriggerConfig> {
 
     private final TbDeviceProfileCache deviceProfileCache;
 
     @Override
-    public boolean matchesFilter(RuleEngineMsgTrigger trigger, DeviceActivityNotificationRuleTriggerConfig triggerConfig) {
+    public boolean matchesFilter(DeviceActivityTrigger trigger, DeviceActivityNotificationRuleTriggerConfig triggerConfig) {
-        if (trigger.getMsg().getOriginator().getEntityType() != EntityType.DEVICE) {
+        DeviceEvent event = trigger.isActive() ? DeviceEvent.ACTIVE : DeviceEvent.INACTIVE;
-            return false;
-        }
-        DeviceEvent event = trigger.getMsg().getType().equals(DataConstants.ACTIVITY_EVENT) ? DeviceEvent.ACTIVE : DeviceEvent.INACTIVE;
         if (!triggerConfig.getNotifyOn().contains(event)) {
             return false;
         }
-        DeviceId deviceId = (DeviceId) trigger.getMsg().getOriginator();
+        DeviceId deviceId = trigger.getDeviceId();
         if (CollectionUtils.isNotEmpty(triggerConfig.getDevices())) {
             return triggerConfig.getDevices().contains(deviceId.getId());
         } else if (CollectionUtils.isNotEmpty(triggerConfig.getDeviceProfiles())) {
@@ -61,15 +53,14 @@ public class DeviceActivityTriggerProcessor implements RuleEngineMsgNotification
     }
 
     @Override
-    public RuleOriginatedNotificationInfo constructNotificationInfo(RuleEngineMsgTrigger trigger) {
+    public RuleOriginatedNotificationInfo constructNotificationInfo(DeviceActivityTrigger trigger) {
-        TbMsg msg = trigger.getMsg();
         return DeviceActivityNotificationInfo.builder()
-                .eventType(trigger.getMsg().getType().equals(DataConstants.ACTIVITY_EVENT) ? "active" : "inactive")
+                .eventType(trigger.isActive() ? "active" : "inactive")
-                .deviceId(msg.getOriginator().getId())
+                .deviceId(trigger.getDeviceId().getId())
-                .deviceName(msg.getMetaData().getValue("deviceName"))
+                .deviceName(trigger.getDeviceName())
-                .deviceType(msg.getMetaData().getValue("deviceType"))
+                .deviceType(trigger.getDeviceType())
-                .deviceLabel(msg.getMetaData().getValue("deviceLabel"))
+                .deviceLabel(trigger.getDeviceLabel())
-                .deviceCustomerId(msg.getCustomerId())
+                .deviceCustomerId(trigger.getCustomerId())
                 .build();
     }
 
@@ -78,9 +69,4 @@ public class DeviceActivityTriggerProcessor implements RuleEngineMsgNotification
         return NotificationRuleTriggerType.DEVICE_ACTIVITY;
     }
 
-    @Override
-    public Set<String> getSupportedMsgTypes() {
-        return Set.of(DataConstants.ACTIVITY_EVENT, DataConstants.INACTIVITY_EVENT);
-    }
-
 }

application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java

@@ -63,7 +63,7 @@ import org.thingsboard.server.queue.TbQueueConsumer;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
-import org.thingsboard.server.queue.notification.NotificationRuleProcessor;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
 import org.thingsboard.server.queue.provider.TbCoreQueueFactory;
 import org.thingsboard.server.queue.util.AfterStartUp;
 import org.thingsboard.server.queue.util.DataDecodingEncodingService;

application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java

@@ -31,7 +31,6 @@ import org.apache.commons.lang3.tuple.Pair;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Lazy;
-import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.common.util.ThingsBoardExecutors;
@@ -63,6 +62,8 @@ import org.thingsboard.server.common.data.query.EntityListFilter;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgDataType;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
+import org.thingsboard.server.common.msg.notification.trigger.DeviceActivityTrigger;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TbCallback;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
@@ -70,12 +71,10 @@ import org.thingsboard.server.common.stats.TbApiUsageReportClient;
 import org.thingsboard.server.dao.attributes.AttributesService;
 import org.thingsboard.server.dao.device.DeviceService;
 import org.thingsboard.server.dao.sql.query.EntityQueryRepository;
-import org.thingsboard.server.dao.tenant.TenantService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.util.DbTypeInfoComponent;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.discovery.PartitionService;
-import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.partition.AbstractPartitionBasedService;
 import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
@@ -158,6 +157,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
     private final EntityQueryRepository entityQueryRepository;
     private final DbTypeInfoComponent dbTypeInfoComponent;
     private final TbApiUsageReportClient apiUsageReportClient;
+    private final NotificationRuleProcessor notificationRuleProcessor;
     @Autowired @Lazy
     private TelemetrySubscriptionService tsSubService;
 
@@ -254,8 +254,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
             state.setLastActivityTime(lastReportedActivity);
             if (!state.isActive()) {
                 state.setActive(true);
-                save(deviceId, ACTIVITY_STATE, true);
+                onDeviceActivityStatusChange(deviceId, true, stateData);
-                pushRuleEngineMessage(stateData, ACTIVITY_EVENT);
             }
         } else {
             log.debug("updateActivityState - fetched state IN NULL for device {}, lastReportedActivity {}", deviceId, lastReportedActivity);
@@ -491,9 +490,8 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
                 if (partitionService.resolve(ServiceType.TB_CORE, stateData.getTenantId(), deviceId).isMyPartition()) {
                     state.setActive(false);
                     state.setLastInactivityAlarmTime(ts);
-                    save(deviceId, ACTIVITY_STATE, false);
+                    onDeviceActivityStatusChange(deviceId, false, stateData);
                     save(deviceId, INACTIVITY_ALARM_TIME, ts);
-                    pushRuleEngineMessage(stateData, INACTIVITY_EVENT);
                 } else {
                     cleanupEntity(deviceId);
                 }
@@ -533,6 +531,19 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
         }
     }
 
+    private void onDeviceActivityStatusChange(DeviceId deviceId, boolean active, DeviceStateData stateData) {
+        save(deviceId, ACTIVITY_STATE, active);
+        pushRuleEngineMessage(stateData, active ? ACTIVITY_EVENT : INACTIVITY_EVENT);
+        TbMsgMetaData metaData = stateData.getMetaData();
+        notificationRuleProcessor.process(DeviceActivityTrigger.builder()
+                .tenantId(stateData.getTenantId()).customerId(stateData.getCustomerId())
+                .deviceId(deviceId).active(active)
+                .deviceName(metaData.getValue("deviceName"))
+                .deviceType(metaData.getValue("deviceType"))
+                .deviceLabel(metaData.getValue("deviceLabel"))
+                .build());
+    }
+
     private boolean cleanDeviceStateIfBelongsExternalPartition(TenantId tenantId, final DeviceId deviceId) {
         TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId);
         boolean cleanup = !partitionedEntities.containsKey(tpi);

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

@@ -52,7 +52,7 @@ import org.thingsboard.server.common.msg.queue.TbCallback;
 import org.thingsboard.server.common.stats.TbApiUsageReportClient;
 import org.thingsboard.server.dao.alarm.AlarmOperationResult;
 import org.thingsboard.server.dao.alarm.AlarmService;
-import org.thingsboard.server.queue.notification.NotificationRuleProcessor;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
 import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
 import org.thingsboard.server.service.entitiy.alarm.TbAlarmCommentService;
 import org.thingsboard.server.service.subscription.TbSubscriptionUtils;

application/src/main/java/org/thingsboard/server/service/update/DefaultUpdateService.java

@@ -29,7 +29,7 @@ import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.common.data.UpdateMessage;
 import org.thingsboard.server.common.msg.notification.trigger.NewPlatformVersionTrigger;
-import org.thingsboard.server.queue.notification.NotificationRuleProcessor;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
 import org.thingsboard.server.queue.util.AfterStartUp;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 

application/src/main/resources/thingsboard.yml

@@ -1271,6 +1271,11 @@ vc:
 
 notification_system:
   thread_pool_size: "${TB_NOTIFICATION_SYSTEM_THREAD_POOL_SIZE:10}"
+  rules:
+    trigger_types_configs:
+      NEW_PLATFORM_VERSION:
+        # In milliseconds, infinitely by default
+        deduplication_duration: "${NEW_PLATFORM_VERSION_NOTIFICATION_RULE_DEDUPLICATION_DURATION:0}"
 
 management:
   endpoints:

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

@@ -264,7 +264,7 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
         int cntTime = 1;
         testNotificationMsgToEdgeServiceTime(entityId, tenantId, actionType, cntTime);
         testLogEntityAction(entity, originatorId, tenantId, customerId, userId, userName, actionType, cntTime, additionalInfo);
-        tesPushMsgToCoreTime(cntTime);
+        testPushMsgToCoreTime(cntTime);
         Mockito.reset(tbClusterService, auditLogService);
     }
 
@@ -363,13 +363,13 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
                 Mockito.any(entityId.getClass()), Mockito.any(ComponentLifecycleEvent.class));
     }
 
-    private void tesPushMsgToCoreTime(int cntTime) {
+    private void testPushMsgToCoreTime(int cntTime) {
         Mockito.verify(tbClusterService, times(cntTime)).pushMsgToCore(Mockito.any(ToDeviceActorNotificationMsg.class), Mockito.isNull());
     }
 
     protected void testLogEntityAction(HasName entity, EntityId originatorId, TenantId tenantId,
-                                     CustomerId customerId, UserId userId, String userName,
+                                       CustomerId customerId, UserId userId, String userName,
-                                     ActionType actionType, int cntTime, Object... additionalInfo) {
+                                       ActionType actionType, int cntTime, Object... additionalInfo) {
         ArgumentMatcher<HasName> matcherEntityEquals = entity == null ? Objects::isNull : argument -> argument.toString().equals(entity.toString());
         ArgumentMatcher<EntityId> matcherOriginatorId = argument -> argument.equals(originatorId);
         ArgumentMatcher<CustomerId> matcherCustomerId = customerId == null ?
@@ -380,10 +380,10 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
                 actionType, cntTime, extractMatcherAdditionalInfo(additionalInfo));
     }
 
-    private void testLogEntityActionEntityEqClass(HasName entity, EntityId originatorId, TenantId tenantId,
+    protected void testLogEntityActionEntityEqClass(HasName entity, EntityId originatorId, TenantId tenantId,
-                                                  CustomerId customerId, UserId userId, String userName,
+                                                    CustomerId customerId, UserId userId, String userName,
-                                                  ActionType actionType, int cntTime, Object... additionalInfo) {
+                                                    ActionType actionType, int cntTime, Object... additionalInfo) {
-        ArgumentMatcher<HasName> matcherEntityEquals = argument -> argument.getClass().equals(entity.getClass());
+        ArgumentMatcher<HasName> matcherEntityEquals = argument -> entity.getClass().isAssignableFrom(argument.getClass());
         ArgumentMatcher<EntityId> matcherOriginatorId = argument -> argument.equals(originatorId);
         ArgumentMatcher<CustomerId> matcherCustomerId = customerId == null ?
                 argument -> argument.getClass().equals(CustomerId.class) : argument -> argument.equals(customerId);
@@ -600,8 +600,8 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
         return fieldName + " length must be equal or less than 255";
     }
 
-    protected String msgErrorNoFound(String entityClassName, String assetIdStr) {
+    protected String msgErrorNoFound(String entityClassName, String entityIdStr) {
-        return entityClassName + " with id [" + assetIdStr + "] is not found";
+        return entityClassName + " with id [" + entityIdStr + "] is not found";
     }
 
     private String entityClassToEntityTypeName(HasName entity) {

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

@@ -46,6 +46,7 @@ import java.util.LinkedList;
 import java.util.List;
 
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
 @Slf4j
@@ -104,7 +105,7 @@ public class AlarmCommentControllerTest extends AbstractControllerTest {
 
         AlarmComment createdComment = createAlarmComment(alarm.getId());
 
-        testLogEntityAction(alarm, alarm.getId(), tenantId, customerId, customerUserId, CUSTOMER_USER_EMAIL, ActionType.ADDED_COMMENT, 1, createdComment);
+        testLogEntityActionEntityEqClass(alarm, alarm.getId(), tenantId, customerId, customerUserId, CUSTOMER_USER_EMAIL, ActionType.ADDED_COMMENT, 1, createdComment);
     }
 
     @Test
@@ -116,7 +117,7 @@ public class AlarmCommentControllerTest extends AbstractControllerTest {
         AlarmComment createdComment = createAlarmComment(alarm.getId());
         Assert.assertEquals(AlarmCommentType.OTHER, createdComment.getType());
 
-        testLogEntityAction(alarm, alarm.getId(), tenantId, customerId, tenantAdminUserId, TENANT_ADMIN_EMAIL, ActionType.ADDED_COMMENT, 1, createdComment);
+        testLogEntityActionEntityEqClass(alarm, alarm.getId(), tenantId, customerId, tenantAdminUserId, TENANT_ADMIN_EMAIL, ActionType.ADDED_COMMENT, 1, createdComment);
     }
 
     @Test
@@ -135,7 +136,7 @@ public class AlarmCommentControllerTest extends AbstractControllerTest {
         Assert.assertEquals("true", updatedAlarmComment.getComment().get("edited").asText());
         Assert.assertNotNull(updatedAlarmComment.getComment().get("editedOn"));
 
-        testLogEntityAction(alarm, alarm.getId(), tenantId, customerId, customerUserId, CUSTOMER_USER_EMAIL, ActionType.UPDATED_COMMENT, 1, savedComment);
+        testLogEntityActionEntityEqClass(alarm, alarm.getId(), tenantId, customerId, customerUserId, CUSTOMER_USER_EMAIL, ActionType.UPDATED_COMMENT, 1, savedComment);
     }
 
     @Test
@@ -154,7 +155,7 @@ public class AlarmCommentControllerTest extends AbstractControllerTest {
         Assert.assertEquals("true", updatedAlarmComment.getComment().get("edited").asText());
         Assert.assertNotNull(updatedAlarmComment.getComment().get("editedOn"));
 
-        testLogEntityAction(alarm, alarm.getId(), tenantId, customerId, tenantAdminUserId, TENANT_ADMIN_EMAIL, ActionType.UPDATED_COMMENT, 1, updatedAlarmComment);
+        testLogEntityActionEntityEqClass(alarm, alarm.getId(), tenantId, customerId, tenantAdminUserId, TENANT_ADMIN_EMAIL, ActionType.UPDATED_COMMENT, 1, updatedAlarmComment);
     }
 
     @Test
@@ -169,8 +170,8 @@ public class AlarmCommentControllerTest extends AbstractControllerTest {
         savedComment.setComment(newComment);
 
         doPost("/api/alarm/" + alarm.getId() + "/comment", savedComment)
-                .andExpect(status().isForbidden())
+                .andExpect(status().isNotFound())
-                .andExpect(statusReason(containsString(msgErrorPermission)));
+                .andExpect(statusReason(equalTo(msgErrorNoFound("Alarm", alarm.getId().toString()))));
 
         testNotifyEntityNever(alarm.getId(), savedComment);
     }
@@ -209,7 +210,7 @@ public class AlarmCommentControllerTest extends AbstractControllerTest {
                 .comment(JacksonUtil.newObjectNode().put("text", String.format("User %s deleted his comment",
                         CUSTOMER_USER_EMAIL)))
                 .build();
-        testLogEntityAction(alarm, alarm.getId(), tenantId, customerId, customerUserId, CUSTOMER_USER_EMAIL, ActionType.DELETED_COMMENT, 1, expectedAlarmComment);
+        testLogEntityActionEntityEqClass(alarm, alarm.getId(), tenantId, customerId, customerUserId, CUSTOMER_USER_EMAIL, ActionType.DELETED_COMMENT, 1, expectedAlarmComment);
     }
 
     @Test
@@ -228,7 +229,7 @@ public class AlarmCommentControllerTest extends AbstractControllerTest {
                 .comment(JacksonUtil.newObjectNode().put("text", String.format("User %s deleted his comment",
                         TENANT_ADMIN_EMAIL)))
                 .build();
-        testLogEntityAction(alarm, alarm.getId(), tenantId, customerId, tenantAdminUserId, TENANT_ADMIN_EMAIL, ActionType.DELETED_COMMENT, 1, expectedAlarmComment);
+        testLogEntityActionEntityEqClass(alarm, alarm.getId(), tenantId, customerId, tenantAdminUserId, TENANT_ADMIN_EMAIL, ActionType.DELETED_COMMENT, 1, expectedAlarmComment);
     }
 
     @Test
@@ -355,16 +356,18 @@ public class AlarmCommentControllerTest extends AbstractControllerTest {
         Assert.assertTrue("Created alarm doesn't match the found one!", equals);
     }
 
-    private AlarmComment createAlarmComment(AlarmId alarmId, String text)  {
+    private AlarmComment createAlarmComment(AlarmId alarmId, String text) {
         AlarmComment alarmComment = AlarmComment.builder()
                 .comment(JacksonUtil.newObjectNode().set("text", new TextNode(text)))
                 .build();
 
         return saveAlarmComment(alarmId, alarmComment);
     }
-    private AlarmComment createAlarmComment(AlarmId alarmId)  {
+
+    private AlarmComment createAlarmComment(AlarmId alarmId) {
         return createAlarmComment(alarmId, "Please take a look");
     }
+
     private AlarmComment saveAlarmComment(AlarmId alarmId, AlarmComment alarmComment) {
         alarmComment = doPost("/api/alarm/" + alarmId + "/comment", alarmComment, AlarmComment.class);
         Assert.assertNotNull(alarmComment);

application/src/test/java/org/thingsboard/server/service/notification/AbstractNotificationApiTest.java

@@ -17,6 +17,7 @@ package org.thingsboard.server.service.notification;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import org.apache.commons.lang3.RandomStringUtils;
+import org.junit.After;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.mock.mockito.MockBean;
 import org.springframework.data.util.Pair;
@@ -26,6 +27,7 @@ import org.thingsboard.server.common.data.User;
 import org.thingsboard.server.common.data.id.NotificationRequestId;
 import org.thingsboard.server.common.data.id.NotificationTargetId;
 import org.thingsboard.server.common.data.id.NotificationTemplateId;
+import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.id.UUIDBased;
 import org.thingsboard.server.common.data.id.UserId;
 import org.thingsboard.server.common.data.notification.Notification;
@@ -43,6 +45,7 @@ import org.thingsboard.server.common.data.notification.settings.NotificationSett
 import org.thingsboard.server.common.data.notification.targets.NotificationTarget;
 import org.thingsboard.server.common.data.notification.targets.platform.PlatformUsersNotificationTargetConfig;
 import org.thingsboard.server.common.data.notification.targets.platform.UserListFilter;
+import org.thingsboard.server.common.data.notification.targets.platform.UsersFilter;
 import org.thingsboard.server.common.data.notification.template.DeliveryMethodNotificationTemplate;
 import org.thingsboard.server.common.data.notification.template.EmailDeliveryMethodNotificationTemplate;
 import org.thingsboard.server.common.data.notification.template.NotificationTemplate;
@@ -54,6 +57,10 @@ import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.common.data.security.Authority;
 import org.thingsboard.server.controller.AbstractControllerTest;
 import org.thingsboard.server.dao.DaoUtil;
+import org.thingsboard.server.dao.notification.NotificationRequestService;
+import org.thingsboard.server.dao.notification.NotificationRuleService;
+import org.thingsboard.server.dao.notification.NotificationTargetService;
+import org.thingsboard.server.dao.notification.NotificationTemplateService;
 
 import java.net.URISyntaxException;
 import java.util.Arrays;
@@ -72,21 +79,40 @@ public abstract class AbstractNotificationApiTest extends AbstractControllerTest
 
     @MockBean
     protected SlackService slackService;
-
     @Autowired
     protected MailService mailService;
 
+    @Autowired
+    protected NotificationRuleService notificationRuleService;
+    @Autowired
+    protected NotificationTemplateService notificationTemplateService;
+    @Autowired
+    protected NotificationTargetService notificationTargetService;
+    @Autowired
+    protected NotificationRequestService notificationRequestService;
+
     public static final String DEFAULT_NOTIFICATION_SUBJECT = "Just a test";
     public static final NotificationType DEFAULT_NOTIFICATION_TYPE = NotificationType.GENERAL;
 
+    @After
+    public void afterEach() {
+        notificationRequestService.deleteNotificationRequestsByTenantId(TenantId.SYS_TENANT_ID);
+        notificationRuleService.deleteNotificationRulesByTenantId(TenantId.SYS_TENANT_ID);
+        notificationTemplateService.deleteNotificationTemplatesByTenantId(TenantId.SYS_TENANT_ID);
+        notificationTargetService.deleteNotificationTargetsByTenantId(TenantId.SYS_TENANT_ID);
+    }
+
     protected NotificationTarget createNotificationTarget(UserId... usersIds) {
-        NotificationTarget notificationTarget = new NotificationTarget();
-        notificationTarget.setTenantId(tenantId);
-        notificationTarget.setName("Users " + List.of(usersIds));
-        PlatformUsersNotificationTargetConfig targetConfig = new PlatformUsersNotificationTargetConfig();
         UserListFilter filter = new UserListFilter();
         filter.setUsersIds(DaoUtil.toUUIDs(List.of(usersIds)));
-        targetConfig.setUsersFilter(filter);
+        return createNotificationTarget(filter);
+    }
+
+    protected NotificationTarget createNotificationTarget(UsersFilter usersFilter) {
+        NotificationTarget notificationTarget = new NotificationTarget();
+        notificationTarget.setName(usersFilter.toString());
+        PlatformUsersNotificationTargetConfig targetConfig = new PlatformUsersNotificationTargetConfig();
+        targetConfig.setUsersFilter(usersFilter);
         notificationTarget.setConfiguration(targetConfig);
         return saveNotificationTarget(notificationTarget);
     }

application/src/test/java/org/thingsboard/server/service/notification/NotificationRuleApiTest.java

@@ -17,12 +17,14 @@ package org.thingsboard.server.service.notification;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.BooleanNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.function.ThrowingRunnable;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.data.util.Pair;
+import org.springframework.test.context.TestPropertySource;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.Device;
@@ -31,10 +33,15 @@ import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.UpdateMessage;
 import org.thingsboard.server.common.data.User;
 import org.thingsboard.server.common.data.alarm.Alarm;
+import org.thingsboard.server.common.data.alarm.AlarmComment;
+import org.thingsboard.server.common.data.alarm.AlarmCommentType;
 import org.thingsboard.server.common.data.alarm.AlarmSearchStatus;
 import org.thingsboard.server.common.data.alarm.AlarmSeverity;
 import org.thingsboard.server.common.data.alarm.AlarmStatus;
 import org.thingsboard.server.common.data.asset.Asset;
+import org.thingsboard.server.common.data.device.data.DefaultDeviceConfiguration;
+import org.thingsboard.server.common.data.device.data.DefaultDeviceTransportConfiguration;
+import org.thingsboard.server.common.data.device.data.DeviceData;
 import org.thingsboard.server.common.data.device.profile.AlarmCondition;
 import org.thingsboard.server.common.data.device.profile.AlarmConditionFilter;
 import org.thingsboard.server.common.data.device.profile.AlarmConditionFilterKey;
@@ -42,6 +49,8 @@ import org.thingsboard.server.common.data.device.profile.AlarmConditionKeyType;
 import org.thingsboard.server.common.data.device.profile.AlarmRule;
 import org.thingsboard.server.common.data.device.profile.DeviceProfileAlarm;
 import org.thingsboard.server.common.data.device.profile.SimpleAlarmConditionSpec;
+import org.thingsboard.server.common.data.id.AlarmId;
+import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.notification.Notification;
 import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod;
 import org.thingsboard.server.common.data.notification.NotificationRequest;
@@ -52,8 +61,11 @@ import org.thingsboard.server.common.data.notification.rule.DefaultNotificationR
 import org.thingsboard.server.common.data.notification.rule.EscalatedNotificationRuleRecipientsConfig;
 import org.thingsboard.server.common.data.notification.rule.NotificationRule;
 import org.thingsboard.server.common.data.notification.rule.NotificationRuleInfo;
+import org.thingsboard.server.common.data.notification.rule.trigger.AlarmAssignmentNotificationRuleTriggerConfig;
+import org.thingsboard.server.common.data.notification.rule.trigger.AlarmCommentNotificationRuleTriggerConfig;
 import org.thingsboard.server.common.data.notification.rule.trigger.AlarmNotificationRuleTriggerConfig;
 import org.thingsboard.server.common.data.notification.rule.trigger.AlarmNotificationRuleTriggerConfig.AlarmAction;
+import org.thingsboard.server.common.data.notification.rule.trigger.DeviceActivityNotificationRuleTriggerConfig;
 import org.thingsboard.server.common.data.notification.rule.trigger.EntitiesLimitNotificationRuleTriggerConfig;
 import org.thingsboard.server.common.data.notification.rule.trigger.EntityActionNotificationRuleTriggerConfig;
 import org.thingsboard.server.common.data.notification.rule.trigger.NewPlatformVersionNotificationRuleTriggerConfig;
@@ -68,13 +80,16 @@ import org.thingsboard.server.common.data.query.FilterPredicateValue;
 import org.thingsboard.server.common.data.rule.RuleChain;
 import org.thingsboard.server.common.data.rule.RuleChainMetaData;
 import org.thingsboard.server.common.data.security.Authority;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
 import org.thingsboard.server.common.msg.notification.trigger.NewPlatformVersionTrigger;
+import org.thingsboard.server.dao.notification.DefaultNotifications;
 import org.thingsboard.server.dao.notification.NotificationRequestService;
 import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.dao.service.DaoSqlTest;
 import org.thingsboard.server.dao.util.limits.LimitedApi;
 import org.thingsboard.server.dao.util.limits.RateLimitService;
-import org.thingsboard.server.queue.notification.NotificationRuleProcessor;
+import org.thingsboard.server.service.notification.rule.cache.DefaultNotificationRulesCache;
+import org.thingsboard.server.service.state.DeviceStateService;
 import org.thingsboard.server.service.telemetry.AlarmSubscriptionService;
 
 import java.util.ArrayList;
@@ -94,8 +109,15 @@ import static org.assertj.core.api.Assertions.offset;
 import static org.assertj.core.api.InstanceOfAssertFactories.type;
 import static org.awaitility.Awaitility.await;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+import static org.thingsboard.server.common.data.notification.rule.trigger.AlarmAssignmentNotificationRuleTriggerConfig.Action.ASSIGNED;
+import static org.thingsboard.server.common.data.notification.rule.trigger.AlarmAssignmentNotificationRuleTriggerConfig.Action.UNASSIGNED;
+import static org.thingsboard.server.common.data.notification.rule.trigger.DeviceActivityNotificationRuleTriggerConfig.DeviceEvent.ACTIVE;
+import static org.thingsboard.server.common.data.notification.rule.trigger.DeviceActivityNotificationRuleTriggerConfig.DeviceEvent.INACTIVE;
 
 @DaoSqlTest
+@TestPropertySource(properties = {
+        "transport.http.enabled=true"
+})
 public class NotificationRuleApiTest extends AbstractNotificationApiTest {
 
     @SpyBean
@@ -108,6 +130,12 @@ public class NotificationRuleApiTest extends AbstractNotificationApiTest {
     private RuleChainService ruleChainService;
     @Autowired
     private NotificationRuleProcessor notificationRuleProcessor;
+    @Autowired
+    private DefaultNotifications defaultNotifications;
+    @Autowired
+    private DefaultNotificationRulesCache notificationRulesCache;
+    @Autowired
+    private DeviceStateService deviceStateService;
 
     @Before
     public void beforeEach() throws Exception {
@@ -297,7 +325,9 @@ public class NotificationRuleApiTest extends AbstractNotificationApiTest {
         notification = getWsClient().getLastDataUpdate().getUpdate();
         assertThat(notification.getSubject()).isEqualTo("critical alarm '" + alarmType + "' is CLEARED_UNACK");
 
-        assertThat(findNotificationRequests(EntityType.ALARM).getData()).filteredOn(NotificationRequest::isScheduled).isEmpty();
+        await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+            assertThat(findNotificationRequests(EntityType.ALARM).getData()).filteredOn(NotificationRequest::isScheduled).isEmpty();
+        });
     }
 
     @Test
@@ -370,6 +400,122 @@ public class NotificationRuleApiTest extends AbstractNotificationApiTest {
         });
     }
 
+    @Test
+    public void testNotificationRuleProcessing_alarmAssignment() throws Exception {
+        AlarmAssignmentNotificationRuleTriggerConfig triggerConfig = AlarmAssignmentNotificationRuleTriggerConfig.builder()
+                .alarmTypes(Set.of("test"))
+                .notifyOn(Set.of(ASSIGNED, UNASSIGNED))
+                .build();
+        NotificationTarget target = createNotificationTarget(tenantAdminUserId);
+        String template = "${userEmail} ${action} alarm on ${alarmOriginatorEntityType} '${alarmOriginatorName}'. Assignee: ${assigneeEmail}";
+        createNotificationRule(triggerConfig, "Test", template, target.getId());
+
+        Device device = createDevice("Device A", "123");
+        Alarm alarm = Alarm.builder()
+                .tenantId(tenantId)
+                .originator(device.getId())
+                .cleared(false)
+                .acknowledged(false)
+                .severity(AlarmSeverity.CRITICAL)
+                .type("test")
+                .startTs(System.currentTimeMillis())
+                .build();
+        alarm = doPost("/api/alarm", alarm, Alarm.class);
+        AlarmId alarmId = alarm.getId();
+
+        checkNotificationAfter(() -> {
+            doPost("/api/alarm/" + alarmId + "/assign/" + tenantAdminUserId).andExpect(status().isOk());
+        }, notification -> {
+            assertThat(notification.getText()).isEqualTo(
+                    TENANT_ADMIN_EMAIL + " assigned alarm on Device 'Device A'. Assignee: " + TENANT_ADMIN_EMAIL
+            );
+        });
+
+        checkNotificationAfter(() -> {
+            doDelete("/api/alarm/" + alarmId + "/assign").andExpect(status().isOk());
+        }, notification -> {
+            assertThat(notification.getText()).isEqualTo(
+                    TENANT_ADMIN_EMAIL + " unassigned alarm on Device 'Device A'. Assignee: "
+            );
+        });
+    }
+
+    @Test
+    public void testNotificationRuleProcessing_alarmComment() throws Exception {
+        AlarmCommentNotificationRuleTriggerConfig triggerConfig = AlarmCommentNotificationRuleTriggerConfig.builder()
+                .alarmTypes(Set.of("test"))
+                .onlyUserComments(true)
+                .notifyOnCommentUpdate(true)
+                .build();
+        NotificationTarget target = createNotificationTarget(tenantAdminUserId);
+        String template = "${userEmail} ${action} comment on alarm ${alarmType}: ${comment}";
+        createNotificationRule(triggerConfig, "Test", template, target.getId());
+
+        Device device = createDevice("Device A", "123");
+        Alarm alarm = Alarm.builder()
+                .tenantId(tenantId)
+                .originator(device.getId())
+                .cleared(false)
+                .acknowledged(false)
+                .severity(AlarmSeverity.CRITICAL)
+                .type("test")
+                .startTs(System.currentTimeMillis())
+                .build();
+        alarm = doPost("/api/alarm", alarm, Alarm.class);
+        AlarmId alarmId = alarm.getId();
+
+        AlarmComment comment = checkNotificationAfter(() -> {
+            return doPost("/api/alarm/" + alarmId + "/comment",
+                    AlarmComment.builder()
+                            .type(AlarmCommentType.OTHER)
+                            .comment(JacksonUtil.newObjectNode()
+                                    .put("text", "this is bad"))
+                            .build(), AlarmComment.class);
+        }, (notification, r) -> {
+            assertThat(notification.getText()).isEqualTo(
+                    TENANT_ADMIN_EMAIL + " added comment on alarm test: this is bad"
+            );
+        });
+
+        checkNotificationAfter(() -> {
+            ((ObjectNode) comment.getComment()).put("text", "this is very bad");
+            doPost("/api/alarm/" + alarmId + "/comment", comment);
+        }, notification -> {
+            assertThat(notification.getText()).isEqualTo(
+                    TENANT_ADMIN_EMAIL + " updated comment on alarm test: this is very bad"
+            );
+        });
+    }
+
+    @Test
+    public void testNotificationRuleProcessing_deviceActivity() throws Exception {
+        DeviceActivityNotificationRuleTriggerConfig triggerConfig = DeviceActivityNotificationRuleTriggerConfig.builder()
+                .notifyOn(Set.of(ACTIVE, INACTIVE))
+                .build();
+        NotificationTarget target = createNotificationTarget(tenantAdminUserId);
+        String template = "Device ${deviceName} (${deviceLabel}) of type ${deviceType} is now ${eventType}";
+        createNotificationRule(triggerConfig, "Test", template, target.getId());
+
+        Device device = new Device();
+        device.setName("A");
+        device.setLabel("Test Device A");
+        device.setType("test");
+        DeviceData deviceData = new DeviceData();
+        deviceData.setTransportConfiguration(new DefaultDeviceTransportConfiguration());
+        deviceData.setConfiguration(new DefaultDeviceConfiguration());
+        device.setDeviceData(deviceData);
+        device = doPost("/api/device", device, Device.class);
+        DeviceId deviceId = device.getId();
+
+        checkNotificationAfter(() -> {
+            deviceStateService.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis());
+        }, notification -> {
+            assertThat(notification.getText()).isEqualTo(
+                    "Device A (Test Device A) of type test is now active"
+            );
+        });
+    }
+
     @Test
     public void testNotificationRuleInfo() throws Exception {
         NotificationDeliveryMethod[] deliveryMethods = {NotificationDeliveryMethod.WEB, NotificationDeliveryMethod.EMAIL};
@@ -444,7 +590,7 @@ public class NotificationRuleApiTest extends AbstractNotificationApiTest {
     }
 
     @Test
-    public void testNotificationsDeduplication() throws Exception {
+    public void testNotificationsDeduplication_newPlatformVersion() throws Exception {
         loginSysAdmin();
         NewPlatformVersionNotificationRuleTriggerConfig triggerConfig = new NewPlatformVersionNotificationRuleTriggerConfig();
         createNotificationRule(triggerConfig, "Test", "Test", createNotificationTarget(tenantAdminUserId).getId());
@@ -477,6 +623,7 @@ public class NotificationRuleApiTest extends AbstractNotificationApiTest {
         triggerConfig.setEntityTypes(Set.of(EntityType.DEVICE));
         triggerConfig.setCreated(true);
         NotificationRule rule = createNotificationRule(triggerConfig, "Created", "Created", createNotificationTarget(tenantAdminUserId).getId());
+        notificationRulesCache.evict(tenantId);
 
         assertThat(getMyNotifications(false, 100)).size().isZero();
         createDevice("Device 1", "default", "111");
@@ -487,6 +634,7 @@ public class NotificationRuleApiTest extends AbstractNotificationApiTest {
 
         rule.setEnabled(false);
         saveNotificationRule(rule);
+        notificationRulesCache.evict(tenantId);
 
         createDevice("Device 2", "default", "222");
         TimeUnit.SECONDS.sleep(5);
@@ -494,7 +642,7 @@ public class NotificationRuleApiTest extends AbstractNotificationApiTest {
 
         rule.setEnabled(true);
         saveNotificationRule(rule);
-        TimeUnit.SECONDS.sleep(2); // for rule update event to reach rules cache
+        notificationRulesCache.evict(tenantId);
 
         createDevice("Device 3", "default", "333");
         await().atMost(30, TimeUnit.SECONDS)

application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java

@@ -29,12 +29,11 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.query.EntityData;
 import org.thingsboard.server.common.data.query.EntityKeyType;
 import org.thingsboard.server.common.data.query.TsValue;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
 import org.thingsboard.server.dao.attributes.AttributesService;
 import org.thingsboard.server.dao.device.DeviceService;
-import org.thingsboard.server.dao.tenant.TenantService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.queue.discovery.PartitionService;
-import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 
 import java.util.Map;
 import java.util.UUID;
@@ -42,6 +41,7 @@ import java.util.UUID;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.mockito.BDDMockito.willReturn;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
@@ -69,7 +69,7 @@ public class DefaultDeviceStateServiceTest {
 
     @Before
     public void setUp() {
-        service = spy(new DefaultDeviceStateService(deviceService, attributesService, tsService, clusterService, partitionService, null, null, null));
+        service = spy(new DefaultDeviceStateService(deviceService, attributesService, tsService, clusterService, partitionService, null, null, null, mock(NotificationRuleProcessor.class)));
     }
 
     @Test

common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestStats.java

@@ -62,6 +62,9 @@ public class NotificationRequestStats {
             return;
         }
         String errorMessage = error.getMessage();
+        if (errorMessage == null) {
+            errorMessage = error.getClass().getSimpleName();
+        }
         errors.computeIfAbsent(deliveryMethod, k -> new ConcurrentHashMap<>()).put(recipient.getTitle(), errorMessage);
     }
 

common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NotificationRule.java

@@ -36,6 +36,8 @@ import javax.validation.constraints.AssertTrue;
 import javax.validation.constraints.NotBlank;
 import javax.validation.constraints.NotNull;
 import java.io.Serializable;
+import java.util.List;
+import java.util.stream.Collectors;
 
 @Data
 @NoArgsConstructor
@@ -84,4 +86,12 @@ public class NotificationRule extends BaseData<NotificationRuleId> implements Ha
                 triggerType == recipientsConfig.getTriggerType();
     }
 
+    @JsonIgnore
+    public String getDeduplicationKey() {
+        String targets = recipientsConfig.getTargetsTable().values().stream()
+                .flatMap(List::stream).sorted().map(Object::toString)
+                .collect(Collectors.joining(","));
+        return String.join(":", targets, triggerConfig.getDeduplicationKey());
+    }
+
 }

common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/NotificationRuleTriggerConfig.java

@@ -15,6 +15,7 @@
  */
 package org.thingsboard.server.common.data.notification.rule.trigger;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonSubTypes.Type;
@@ -39,4 +40,9 @@ public interface NotificationRuleTriggerConfig extends Serializable {
 
     NotificationRuleTriggerType getTriggerType();
 
+    @JsonIgnore
+    default String getDeduplicationKey() {
+        return "#";
+    }
+
 }

common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/NotificationRuleTriggerType.java

@@ -16,10 +16,8 @@
 package org.thingsboard.server.common.data.notification.rule.trigger;
 
 import lombok.Getter;
-import lombok.RequiredArgsConstructor;
 
 @Getter
-@RequiredArgsConstructor
 public enum NotificationRuleTriggerType {
 
     ENTITY_ACTION,
@@ -28,15 +26,18 @@ public enum NotificationRuleTriggerType {
     ALARM_ASSIGNMENT,
     DEVICE_ACTIVITY,
     RULE_ENGINE_COMPONENT_LIFECYCLE_EVENT,
-    NEW_PLATFORM_VERSION(false, true),
+    NEW_PLATFORM_VERSION(false),
-    ENTITIES_LIMIT(false, false),
+    ENTITIES_LIMIT(false),
-    API_USAGE_LIMIT(false, false);
+    API_USAGE_LIMIT(false);
 
     private final boolean tenantLevel;
-    private final boolean deduplicate;
 
     NotificationRuleTriggerType() {
-        this(true, false);
+        this(true);
+    }
+
+    NotificationRuleTriggerType(boolean tenantLevel) {
+        this.tenantLevel = tenantLevel;
     }
 
 }

application/src/main/java/org/thingsboard/server/service/notification/rule/trigger/RuleEngineMsgNotificationRuleTriggerProcessor.java → common/data/src/main/java/org/thingsboard/server/common/data/notification/settings/TriggerTypeConfig.java

@@ -13,15 +13,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.thingsboard.server.service.notification.rule.trigger;
+package org.thingsboard.server.common.data.notification.settings;
 
-import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerConfig;
+import lombok.Data;
-import org.thingsboard.server.common.msg.notification.trigger.RuleEngineMsgTrigger;
-
-import java.util.Set;
-
-public interface RuleEngineMsgNotificationRuleTriggerProcessor<C extends NotificationRuleTriggerConfig> extends NotificationRuleTriggerProcessor<RuleEngineMsgTrigger, C> {
-
-    Set<String> getSupportedMsgTypes();
 
+@Data
+public class TriggerTypeConfig {
+    private long deduplicationDuration;
 }

common/data/src/main/java/org/thingsboard/server/common/data/util/CollectionsUtil.java

@@ -16,7 +16,6 @@
 package org.thingsboard.server.common.data.util;
 
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
@@ -51,20 +50,19 @@ public class CollectionsUtil {
     }
 
     @SuppressWarnings("unchecked")
-    public static <K, V> Map<K, V> mapOf(Object... kvs) {
+    public static <T> Map<T, T> mapOf(T... kvs) {
-        Map<K, V> map = new HashMap<>();
+        if (kvs.length % 2 != 0) {
+            throw new IllegalArgumentException("Invalid number of parameters");
+        }
+        Map<T, T> map = new HashMap<>();
         for (int i = 0; i < kvs.length; i += 2) {
-            K key = (K) kvs[i];
+            T key = kvs[i];
-            V value = (V) kvs[i + 1];
+            T value = kvs[i + 1];
             map.put(key, value);
         }
         return map;
     }
 
-    public static <K, V> Map<K, V> unmodifiableMapOf(Object... kvs) {
-        return Collections.unmodifiableMap(mapOf(kvs));
-    }
-
     public static <V> boolean emptyOrContains(Collection<V> collection, V element) {
         return isEmpty(collection) || collection.contains(element);
     }

common/queue/src/main/java/org/thingsboard/server/queue/notification/NotificationRuleProcessor.java → common/message/src/main/java/org/thingsboard/server/common/msg/notification/NotificationRuleProcessor.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.thingsboard.server.queue.notification;
+package org.thingsboard.server.common.msg.notification;
 
 import org.thingsboard.server.common.msg.notification.trigger.NotificationRuleTrigger;
 

common/message/src/main/java/org/thingsboard/server/common/msg/notification/trigger/RuleEngineMsgTrigger.java → common/message/src/main/java/org/thingsboard/server/common/msg/notification/trigger/AlarmAssignmentTrigger.java

@@ -17,30 +17,30 @@ package org.thingsboard.server.common.msg.notification.trigger;
 
 import lombok.Builder;
 import lombok.Data;
+import org.thingsboard.server.common.data.User;
+import org.thingsboard.server.common.data.alarm.AlarmInfo;
+import org.thingsboard.server.common.data.audit.ActionType;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
-import org.thingsboard.server.common.msg.TbMsg;
-
-import java.util.Map;
 
 @Data
 @Builder
-public class RuleEngineMsgTrigger implements NotificationRuleTrigger {
+public class AlarmAssignmentTrigger implements NotificationRuleTrigger {
 
     private final TenantId tenantId;
-    private final TbMsg msg;
+    private final AlarmInfo alarmInfo;
-
+    private final ActionType actionType;
-    public static Map<String, NotificationRuleTriggerType> msgTypeToTriggerType; // set on init by DefaultNotificationRuleProcessor
+    private final User user;
 
     @Override
-    public NotificationRuleTriggerType getType() {
+    public EntityId getOriginatorEntityId() {
-        return msgTypeToTriggerType != null ? msgTypeToTriggerType.get(msg.getType()) : null;
+        return alarmInfo.getOriginator();
     }
 
     @Override
-    public EntityId getOriginatorEntityId() {
+    public NotificationRuleTriggerType getType() {
-        return msg.getOriginator();
+        return NotificationRuleTriggerType.ALARM_ASSIGNMENT;
     }
 
 }

common/message/src/main/java/org/thingsboard/server/common/msg/notification/trigger/AlarmCommentTrigger.java

@@ -0,0 +1,48 @@
+/**
+ * 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.server.common.msg.notification.trigger;
+
+import lombok.Builder;
+import lombok.Data;
+import org.thingsboard.server.common.data.User;
+import org.thingsboard.server.common.data.alarm.Alarm;
+import org.thingsboard.server.common.data.alarm.AlarmComment;
+import org.thingsboard.server.common.data.audit.ActionType;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
+
+@Data
+@Builder
+public class AlarmCommentTrigger implements NotificationRuleTrigger {
+
+    private final TenantId tenantId;
+    private final AlarmComment comment;
+    private final Alarm alarm;
+    private final ActionType actionType;
+    private final User user;
+
+    @Override
+    public NotificationRuleTriggerType getType() {
+        return NotificationRuleTriggerType.ALARM_COMMENT;
+    }
+
+    @Override
+    public EntityId getOriginatorEntityId() {
+        return alarm.getId();
+    }
+
+}

common/message/src/main/java/org/thingsboard/server/common/msg/notification/trigger/ApiUsageLimitTrigger.java

@@ -36,11 +36,6 @@ public class ApiUsageLimitTrigger implements NotificationRuleTrigger {
         return NotificationRuleTriggerType.API_USAGE_LIMIT;
     }
 
-    @Override
-    public TenantId getTenantId() {
-        return tenantId;
-    }
-
     @Override
     public EntityId getOriginatorEntityId() {
         return tenantId;

common/message/src/main/java/org/thingsboard/server/common/msg/notification/trigger/DeviceActivityTrigger.java

@@ -0,0 +1,49 @@
+/**
+ * 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.server.common.msg.notification.trigger;
+
+import lombok.Builder;
+import lombok.Data;
+import org.thingsboard.server.common.data.id.CustomerId;
+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.notification.rule.trigger.NotificationRuleTriggerType;
+
+@Data
+@Builder
+public class DeviceActivityTrigger implements NotificationRuleTrigger {
+
+    private final TenantId tenantId;
+    private final CustomerId customerId;
+    private final DeviceId deviceId;
+    private final boolean active;
+
+    private final String deviceName;
+    private final String deviceType;
+    private final String deviceLabel;
+
+    @Override
+    public EntityId getOriginatorEntityId() {
+        return deviceId;
+    }
+
+    @Override
+    public NotificationRuleTriggerType getType() {
+        return NotificationRuleTriggerType.DEVICE_ACTIVITY;
+    }
+
+}

common/message/src/main/java/org/thingsboard/server/common/msg/notification/trigger/NewPlatformVersionTrigger.java

@@ -43,4 +43,21 @@ public class NewPlatformVersionTrigger implements NotificationRuleTrigger {
         return TenantId.SYS_TENANT_ID;
     }
 
+
+    @Override
+    public boolean deduplicate() {
+        return true;
+    }
+
+    @Override
+    public String getDeduplicationKey() {
+        return String.join(":", NotificationRuleTrigger.super.getDeduplicationKey(),
+                updateInfo.getCurrentVersion(), updateInfo.getLatestVersion());
+    }
+
+    @Override
+    public long getDefaultDeduplicationDuration() {
+        return 0;
+    }
+
 }

common/message/src/main/java/org/thingsboard/server/common/msg/notification/trigger/NotificationRuleTrigger.java

@@ -29,4 +29,18 @@ public interface NotificationRuleTrigger extends Serializable {
 
     EntityId getOriginatorEntityId();
 
+
+    default boolean deduplicate() {
+        return false;
+    }
+
+    default String getDeduplicationKey() {
+        EntityId originatorEntityId = getOriginatorEntityId();
+        return String.join(":", getType().toString(), originatorEntityId.getEntityType().toString(), originatorEntityId.getId().toString());
+    }
+
+    default long getDefaultDeduplicationDuration() {
+        return 0;
+    }
+
 }

common/queue/src/main/java/org/thingsboard/server/queue/notification/RemoteNotificationRuleProcessor.java

@@ -19,7 +19,12 @@ import com.google.protobuf.ByteString;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.stereotype.Service;
+import org.springframework.util.ConcurrentReferenceHashMap;
+import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
+import org.thingsboard.server.common.data.notification.settings.TriggerTypeConfig;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
 import org.thingsboard.server.common.msg.notification.trigger.NotificationRuleTrigger;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
@@ -30,10 +35,17 @@ import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
 import org.thingsboard.server.queue.util.DataDecodingEncodingService;
 
+import java.util.EnumMap;
+import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.springframework.util.ConcurrentReferenceHashMap.ReferenceType.SOFT;
 
 @Service
 @ConditionalOnMissingBean(value = NotificationRuleProcessor.class, ignored = RemoteNotificationRuleProcessor.class)
+@ConfigurationProperties(prefix = "notification-system.rules")
 @RequiredArgsConstructor
 @Slf4j
 public class RemoteNotificationRuleProcessor implements NotificationRuleProcessor {
@@ -43,10 +55,16 @@ public class RemoteNotificationRuleProcessor implements NotificationRuleProcesso
     private final PartitionService partitionService;
     private final DataDecodingEncodingService encodingService;
 
+    private Map<NotificationRuleTriggerType, TriggerTypeConfig> triggerTypesConfigs;
+    private final ConcurrentMap<String, Long> submittedTriggers = new ConcurrentReferenceHashMap<>(16, SOFT);
+
     @Override
     public void process(NotificationRuleTrigger trigger) {
+        if (trigger.deduplicate() && alreadySubmitted(trigger)) {
+            return;
+        }
         try {
-            log.trace("Submitting notification rule trigger: {}", trigger);
+            log.debug("Submitting notification rule trigger: {}", trigger);
             TransportProtos.NotificationRuleProcessorMsg.Builder msg = TransportProtos.NotificationRuleProcessorMsg.newBuilder()
                     .setTrigger(ByteString.copyFrom(encodingService.encode(trigger)));
 
@@ -57,9 +75,52 @@ public class RemoteNotificationRuleProcessor implements NotificationRuleProcesso
                                 .setNotificationRuleProcessorMsg(msg)
                                 .build()), null);
             });
-        } catch (Exception e) {
+        } catch (Throwable e) {
             log.error("Failed to submit notification rule trigger: {}", trigger, e);
         }
     }
 
+    private boolean alreadySubmitted(NotificationRuleTrigger trigger) {
+        String deduplicationKey = trigger.getDeduplicationKey();
+
+        AtomicBoolean alreadySubmitted = new AtomicBoolean(false);
+        submittedTriggers.compute(deduplicationKey, (key, lastSubmittedTs) -> {
+            long currentTs = System.currentTimeMillis();
+            if (lastSubmittedTs == null) {
+                return currentTs;
+            } else {
+                long deduplicationDuration = getDeduplicationDuration(trigger);
+                long passed = currentTs - lastSubmittedTs;
+                if (deduplicationDuration == 0 || passed <= deduplicationDuration) {
+                    log.trace("Notification rule trigger {} was already submitted {} ms ago, deduplication duration is {} ms. Key: '{}'",
+                            trigger.getType(), passed, deduplicationDuration, deduplicationKey);
+                    alreadySubmitted.set(true);
+                    return lastSubmittedTs;
+                } else {
+                    return currentTs;
+                }
+            }
+        });
+        return alreadySubmitted.get();
+    }
+
+    private long getDeduplicationDuration(NotificationRuleTrigger trigger) {
+        if (triggerTypesConfigs == null) {
+            triggerTypesConfigs = new EnumMap<>(NotificationRuleTriggerType.class);
+        }
+        TriggerTypeConfig triggerTypeConfig = triggerTypesConfigs.computeIfAbsent(trigger.getType(), triggerType -> {
+            TriggerTypeConfig config = new TriggerTypeConfig();
+            config.setDeduplicationDuration(trigger.getDefaultDeduplicationDuration());
+            return config;
+        });
+        return triggerTypeConfig.getDeduplicationDuration();
+    }
+
+    // set from ConfigurationProperties
+    public void setTriggerTypesConfigs(Map<NotificationRuleTriggerType, TriggerTypeConfig> triggerTypesConfigs) {
+        if (triggerTypesConfigs != null) {
+            this.triggerTypesConfigs = new EnumMap<>(triggerTypesConfigs);
+        }
+    }
+
 }

msa/vc-executor/src/main/resources/tb-vc-executor.yml

@@ -202,4 +202,4 @@ management:
 service:
   type: "${TB_SERVICE_TYPE:tb-vc-executor}"
   # Unique id for this service (autogenerated if empty)
-  id: "${TB_SERVICE_ID:}"
+  id: "${TB_SERVICE_ID:}"