Merge pull request #11762 from thingsboard/hotfix/3.7

hotfix/3.7

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

@@ -95,7 +95,7 @@ public abstract class RuleChainManagerActor extends ContextAwareActor {
                 () -> {
                     RuleChain ruleChain = provider.apply(ruleChainId);
                     if (ruleChain == null) {
-                        return new RuleChainErrorActor.ActorCreator(systemContext, tenantId,
+                        return new RuleChainErrorActor.ActorCreator(systemContext, tenantId, ruleChainId,
                                 new RuleEngineException("Rule Chain with id: " + ruleChainId + " not found!"));
                     } else {
                         return new RuleChainActor.ActorCreator(systemContext, tenantId, ruleChain);

application/src/main/java/org/thingsboard/server/actors/shared/RuleChainErrorActor.java

@@ -19,16 +19,15 @@ import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.TbActor;
 import org.thingsboard.server.actors.TbActorId;
-import org.thingsboard.server.actors.TbStringActorId;
+import org.thingsboard.server.actors.TbEntityActorId;
 import org.thingsboard.server.actors.service.ContextAwareActor;
 import org.thingsboard.server.actors.service.ContextBasedCreator;
+import org.thingsboard.server.common.data.id.RuleChainId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.msg.TbActorMsg;
 import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg;
 import org.thingsboard.server.common.msg.queue.RuleEngineException;
 
-import java.util.UUID;
-
 @Slf4j
 public class RuleChainErrorActor extends ContextAwareActor {
 
@@ -43,9 +42,8 @@ public class RuleChainErrorActor extends ContextAwareActor {
 
     @Override
     protected boolean doProcess(TbActorMsg msg) {
-        if (msg instanceof RuleChainAwareMsg) {
+        if (msg instanceof RuleChainAwareMsg rcMsg) {
             log.debug("[{}] Reply with {} for message {}", tenantId, error.getMessage(), msg);
-            var rcMsg = (RuleChainAwareMsg) msg;
             rcMsg.getMsg().getCallback().onFailure(error);
             return true;
         } else {
@@ -56,17 +54,19 @@ public class RuleChainErrorActor extends ContextAwareActor {
     public static class ActorCreator extends ContextBasedCreator {
 
         private final TenantId tenantId;
+        private final RuleChainId ruleChainId;
         private final RuleEngineException error;
 
-        public ActorCreator(ActorSystemContext context, TenantId tenantId, RuleEngineException error) {
+        public ActorCreator(ActorSystemContext context, TenantId tenantId, RuleChainId ruleChainId, RuleEngineException error) {
             super(context);
             this.tenantId = tenantId;
+            this.ruleChainId = ruleChainId;
             this.error = error;
         }
 
         @Override
         public TbActorId createActorId() {
-            return new TbStringActorId(UUID.randomUUID().toString());
+            return new TbEntityActorId(ruleChainId);
         }
 
         @Override

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

@@ -20,6 +20,7 @@ import lombok.EqualsAndHashCode;
 import lombok.RequiredArgsConstructor;
 import lombok.ToString;
 
+import java.util.HashSet;
 import java.util.Set;
 
 /**
@@ -48,7 +49,7 @@ public class TbSubscriptionsInfo {
     }
 
     protected TbSubscriptionsInfo copy(int seqNumber) {
-        return new TbSubscriptionsInfo(notifications, alarms, tsAllKeys, tsKeys, attrAllKeys, attrKeys, seqNumber);
+        return new TbSubscriptionsInfo(notifications, alarms, tsAllKeys, tsKeys != null ? new HashSet<>(tsKeys) : null, attrAllKeys, attrKeys != null ? new HashSet<>(attrKeys) : null, seqNumber);
     }
 
 }

application/src/test/java/org/thingsboard/server/actors/tenant/TenantActorTest.java

@@ -18,57 +18,130 @@ package org.thingsboard.server.actors.tenant;
 import org.junit.Before;
 import org.junit.Test;
 import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.DefaultTbActorSystem;
 import org.thingsboard.server.actors.TbActorCtx;
+import org.thingsboard.server.actors.TbActorMailbox;
 import org.thingsboard.server.actors.TbActorRef;
+import org.thingsboard.server.actors.TbActorSystem;
+import org.thingsboard.server.actors.TbActorSystemSettings;
+import org.thingsboard.server.actors.TbEntityActorId;
+import org.thingsboard.server.actors.ruleChain.RuleChainActor;
+import org.thingsboard.server.actors.ruleChain.RuleChainToRuleChainMsg;
+import org.thingsboard.server.actors.shared.RuleChainErrorActor;
+import org.thingsboard.server.common.data.ApiUsageState;
 import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.id.RuleChainId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
+import org.thingsboard.server.common.data.rule.RuleChain;
+import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.common.msg.rule.engine.DeviceDeleteMsg;
+import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.dao.tenant.TenantService;
+import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
+import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
 
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
+import static org.thingsboard.server.actors.service.DefaultActorService.RULE_DISPATCHER_NAME;
 
 public class TenantActorTest {
 
     TenantActor tenantActor;
-    TbActorCtx ctx;
     ActorSystemContext systemContext;
+    RuleChainService ruleChainService;
+    PartitionService partitionService;
     TenantId tenantId = TenantId.SYS_TENANT_ID;
     DeviceId deviceId = DeviceId.fromString("78bf9b26-74ef-4af2-9cfb-ad6cf24ad2ec");
+    RuleChainId ruleChainId = new RuleChainId(UUID.fromString("48cfa2b0-3dca-11ef-8d1a-37c2894cc59c"));
 
     @Before
     public void setUp() throws Exception {
         systemContext = mock(ActorSystemContext.class);
-        ctx = mock(TbActorCtx.class);
+        ruleChainService = mock(RuleChainService.class);
+        partitionService = mock();
+
+        TbServiceInfoProvider serviceInfoProvider = mock(TbServiceInfoProvider.class);
+        TbApiUsageStateService apiUsageService = mock(TbApiUsageStateService.class);
+        TenantService tenantService = mock(TenantService.class);
+
+        when(systemContext.getRuleChainService()).thenReturn(ruleChainService);
         tenantActor = (TenantActor) new TenantActor.ActorCreator(systemContext, tenantId).createActor();
-        when(systemContext.getTenantService()).thenReturn(mock(TenantService.class));
+
-        tenantActor.init(ctx);
+        when(tenantService.findTenantById(tenantId)).thenReturn(mock());
-        tenantActor.cantFindTenant = false;
+        when(systemContext.getTenantService()).thenReturn(tenantService);
+        when(serviceInfoProvider.isService(ServiceType.TB_CORE)).thenReturn(true);
+        when(serviceInfoProvider.isService(ServiceType.TB_RULE_ENGINE)).thenReturn(true);
+        when(systemContext.getServiceInfoProvider()).thenReturn(serviceInfoProvider);
+        when(partitionService.isManagedByCurrentService(tenantId)).thenReturn(true);
+        when(systemContext.getPartitionService()).thenReturn(partitionService);
+        when(systemContext.getApiUsageStateService()).thenReturn(apiUsageService);
+        when(apiUsageService.getApiUsageState(tenantId)).thenReturn(new ApiUsageState());
     }
 
     @Test
-    public void deleteDeviceTest() {
+    public void deleteDeviceTest() throws Exception {
+        TbActorCtx ctx = mock(TbActorCtx.class);
+        tenantActor.init(ctx);
         TbActorRef deviceActorRef = mock(TbActorRef.class);
         when(systemContext.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(new TopicPartitionInfo("Main", tenantId, 0, true));
         when(ctx.getOrCreateChildActor(any(), any(), any(), any())).thenReturn(deviceActorRef);
+
         ComponentLifecycleMsg componentLifecycleMsg = new ComponentLifecycleMsg(tenantId, deviceId, ComponentLifecycleEvent.DELETED);
         tenantActor.doProcess(componentLifecycleMsg);
         verify(deviceActorRef).tellWithHighPriority(eq(new DeviceDeleteMsg(tenantId, deviceId)));
-
         reset(ctx, deviceActorRef);
+
         when(systemContext.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(new TopicPartitionInfo("Main", tenantId, 1, false));
         tenantActor.doProcess(componentLifecycleMsg);
         verify(ctx, never()).getOrCreateChildActor(any(), any(), any(), any());
         verify(deviceActorRef, never()).tellWithHighPriority(any());
     }
 
+    @Test
+    public void ruleChainErrorActorTest() throws Exception {
+        TbActorSystemSettings settings = new TbActorSystemSettings(0, 0, 0);
+        TbActorSystem system = spy(new DefaultTbActorSystem(settings));
+        system.createDispatcher(RULE_DISPATCHER_NAME, mock());
+        TbActorMailbox tenantCtx = new TbActorMailbox(system, settings, null, mock(), mock(), null);
+        tenantActor.init(tenantCtx);
+
+        TbMsg msg = mock(TbMsg.class);
+
+        when(ruleChainService.findRuleChainById(tenantId, ruleChainId)).thenReturn(new RuleChain(ruleChainId));
+
+        RuleChainToRuleChainMsg ruleChainMsg = new RuleChainToRuleChainMsg(ruleChainId, null, msg, null);
+        tenantActor.doProcess(ruleChainMsg);
+        verify(system).createChildActor(eq(RULE_DISPATCHER_NAME), any(RuleChainActor.ActorCreator.class), any());
+        reset(system);
+        tenantActor.doProcess(ruleChainMsg);
+        verify(system, never()).createChildActor(any(), any(), any());
+
+        //Delete rule-chain
+        TbActorRef ruleChainActor = system.getActor(new TbEntityActorId(ruleChainId));
+        assertNotNull(ruleChainActor);
+        system.stop(ruleChainActor);
+        when(ruleChainService.findRuleChainById(tenantId, ruleChainId)).thenReturn(null);
+
+        tenantActor.doProcess(ruleChainMsg);
+        verify(system).createChildActor(eq(RULE_DISPATCHER_NAME), any(RuleChainErrorActor.ActorCreator.class), any());
+        reset(system);
+        tenantActor.doProcess(ruleChainMsg);
+        verify(system, never()).createChildActor(any(), any(), any());
+        system.stop();
+    }
+
 }

application/src/test/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionServiceTest.java

@@ -0,0 +1,126 @@
+/**
+ * Copyright © 2016-2024 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.service.subscription;
+
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.thingsboard.server.cache.limits.RateLimitService;
+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.limit.LimitedApi;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
+import org.thingsboard.server.gen.transport.TransportProtos;
+import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.service.ws.WebSocketSessionRef;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.Executors;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.nullable;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class DefaultTbLocalSubscriptionServiceTest {
+
+    ListAppender<ILoggingEvent> testLogAppender;
+    TbLocalSubscriptionService subscriptionService;
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        Logger logger = (Logger) LoggerFactory.getLogger(DefaultTbLocalSubscriptionService.class);
+        testLogAppender = new ListAppender<>();
+        testLogAppender.start();
+        logger.addAppender(testLogAppender);
+
+        RateLimitService rateLimitService = mock();
+        when(rateLimitService.checkRateLimit(eq(LimitedApi.WS_SUBSCRIPTIONS), any(Object.class), nullable(String.class))).thenReturn(true);
+        PartitionService partitionService = mock();
+        when(partitionService.resolve(any(), any(), any())).thenReturn(TopicPartitionInfo.builder().build());
+        subscriptionService = new DefaultTbLocalSubscriptionService(mock(), mock(), mock(), partitionService, mock(), mock(), mock(), rateLimitService);
+        ReflectionTestUtils.setField(subscriptionService, "serviceId", "serviceId");
+    }
+
+    @AfterEach
+    public void tearDown() {
+        if (testLogAppender != null) {
+            testLogAppender.stop();
+            Logger logger = (Logger) LoggerFactory.getLogger(DefaultTbLocalSubscriptionService.class);
+            logger.detachAppender(testLogAppender);
+        }
+    }
+
+    @Test
+    public void addSubscriptionConcurrentModificationTest() throws Exception {
+        ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
+        TenantId tenantId = new TenantId(UUID.randomUUID());
+        DeviceId deviceId = new DeviceId(UUID.randomUUID());
+        WebSocketSessionRef sessionRef = mock();
+        ReflectionTestUtils.setField(subscriptionService, "subscriptionUpdateExecutor", executorService);
+
+        List<ListenableFuture<?>> futures = new ArrayList<>();
+
+        try {
+            subscriptionService.onCoreStartupMsg(TransportProtos.CoreStartupMsg.newBuilder().addAllPartitions(List.of(0)).getDefaultInstanceForType());
+            for (int i = 0; i < 50; i++) {
+                futures.add(executorService.submit(() -> subscriptionService.addSubscription(createSubscription(tenantId, deviceId), sessionRef)));
+            }
+            Futures.allAsList(futures).get();
+        } finally {
+            executorService.shutdownNow();
+        }
+
+        List<ILoggingEvent> logs = testLogAppender.list;
+        boolean exceptionLogged = logs.stream()
+                .filter(event -> event.getThrowableProxy() != null)
+                .map(event -> event.getThrowableProxy().getClassName())
+                .anyMatch(log -> log.equals("java.util.ConcurrentModificationException"));
+
+        assertFalse(exceptionLogged, "Detected ConcurrentModificationException!");
+    }
+
+    private TbSubscription<?> createSubscription(TenantId tenantId, EntityId entityId) {
+        Map<String, Long> keys = new HashMap<>();
+        for (int i = 0; i < 50; i++) {
+            keys.put(RandomStringUtils.randomAlphanumeric(5), 1L);
+        }
+        return TbAttributeSubscription.builder()
+                .tenantId(tenantId)
+                .entityId(entityId)
+                .subscriptionId(1)
+                .sessionId(RandomStringUtils.randomAlphanumeric(5))
+                .keyStates(keys)
+                .build();
+    }
+}