diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index b60f960498..1ed919e922 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -141,6 +141,7 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -856,9 +857,9 @@ public class ActorSystemContext { appActor.tellWithHighPriority(tbActorMsg); } - public void schedulePeriodicMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs, long periodInMs) { + public ScheduledFuture schedulePeriodicMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs, long periodInMs) { log.debug("Scheduling periodic msg {} every {} ms with delay {} ms", msg, periodInMs, delayInMs); - getScheduler().scheduleWithFixedDelay(() -> ctx.tell(msg), delayInMs, periodInMs, TimeUnit.MILLISECONDS); + return getScheduler().scheduleWithFixedDelay(() -> ctx.tell(msg), delayInMs, periodInMs, TimeUnit.MILLISECONDS); } public void scheduleMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs) { diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java index b0185fd555..2d182510e9 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java @@ -22,6 +22,7 @@ import org.thingsboard.server.actors.TbActorRef; import org.thingsboard.server.actors.TbCalculatedFieldEntityActorId; import org.thingsboard.server.actors.service.DefaultActorService; import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; @@ -35,6 +36,7 @@ import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg; import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; +import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.dao.cf.CalculatedFieldService; import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; @@ -121,7 +123,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware if (calculatedField != null) { msg.getState().setRequiredArguments(calculatedField.getArgNames()); - log.info("Pushing CF state restore msg to specific actor [{}]", msg.getId().entityId()); + log.debug("Pushing CF state restore msg to specific actor [{}]", msg.getId().entityId()); getOrCreateActor(msg.getId().entityId()).tell(msg); } else { cfStateService.removeState(msg.getId(), msg.getCallback()); @@ -178,6 +180,9 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware EntityId entityId = msg.getEntityId(); EntityId profileId = getProfileId(tenantId, entityId); cfEntityCache.add(tenantId, profileId, entityId); + if (!isMyPartition(entityId, callback)) { + return; + } var entityIdFields = getCalculatedFieldsByEntityId(entityId); var profileIdFields = getCalculatedFieldsByEntityId(profileId); var fieldsCount = entityIdFields.size() + profileIdFields.size(); @@ -193,6 +198,9 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private void onEntityUpdated(ComponentLifecycleMsg msg, TbCallback callback) { if (msg.getOldProfileId() != null && msg.getOldProfileId() != msg.getProfileId()) { cfEntityCache.update(tenantId, msg.getOldProfileId(), msg.getProfileId(), msg.getEntityId()); + if (!isMyPartition(msg.getEntityId(), callback)) { + return; + } var oldProfileCfs = getCalculatedFieldsByEntityId(msg.getOldProfileId()); var newProfileCfs = getCalculatedFieldsByEntityId(msg.getProfileId()); var fieldsCount = oldProfileCfs.size() + newProfileCfs.size(); @@ -209,8 +217,10 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) { cfEntityCache.evict(tenantId, msg.getEntityId()); - log.info("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId()); - getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); + if (isMyPartition(msg.getEntityId(), callback)) { + log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId()); + getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); + } } private void onCfCreated(ComponentLifecycleMsg msg, TbCallback callback) throws CalculatedFieldException { @@ -311,7 +321,9 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware callback.onSuccess(); } } else { - deleteCfForEntity(entityId, cfId, callback); + if (isMyPartition(entityId, callback)) { + deleteCfForEntity(entityId, cfId, callback); + } } } } @@ -418,20 +430,31 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware callback.onSuccess(); } } else { - initCfForEntity(entityId, cfCtx, forceStateReinit, callback); + if (isMyPartition(entityId, callback)) { + initCfForEntity(entityId, cfCtx, forceStateReinit, callback); + } } } private void deleteCfForEntity(EntityId entityId, CalculatedFieldId cfId, TbCallback callback) { - log.info("Pushing delete CF msg to specific actor [{}]", entityId); + log.debug("Pushing delete CF msg to specific actor [{}]", entityId); getOrCreateActor(entityId).tell(new CalculatedFieldEntityDeleteMsg(tenantId, cfId, callback)); } private void initCfForEntity(EntityId entityId, CalculatedFieldCtx cfCtx, boolean forceStateReinit, TbCallback callback) { - log.info("Pushing entity init CF msg to specific actor [{}]", entityId); + log.debug("Pushing entity init CF msg to specific actor [{}]", entityId); getOrCreateActor(entityId).tell(new EntityInitCalculatedFieldMsg(tenantId, cfCtx, callback, forceStateReinit)); } + private boolean isMyPartition(EntityId entityId, TbCallback callback) { + if (!systemContext.getPartitionService().resolve(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME, tenantId, entityId).isMyPartition()) { + log.debug("[{}] Entity belongs to external partition.", entityId); + callback.onSuccess(); + return false; + } + return true; + } + private static boolean isProfileEntity(EntityType entityType) { return EntityType.DEVICE_PROFILE.equals(entityType) || EntityType.ASSET_PROFILE.equals(entityType); } diff --git a/application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java b/application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java index 5778f02663..7b390cd9f3 100644 --- a/application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java @@ -29,6 +29,9 @@ import org.thingsboard.server.common.msg.TbActorStopReason; import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; +import java.util.Optional; +import java.util.concurrent.ScheduledFuture; + /** * @author Andrew Shvayka */ @@ -41,6 +44,7 @@ public abstract class ComponentActor statsScheduledFuture = null; public ComponentActor(ActorSystemContext systemContext, TenantId tenantId, T id) { super(systemContext); @@ -73,9 +77,9 @@ public abstract class ComponentActor x.cancel(false)); + statsScheduledFuture = null; } catch (Exception e) { log.warn("[{}][{}] Failed to stop {} processor: {}", tenantId, id, id.getEntityType(), e.getMessage()); logAndPersist("OnStop", e, true); diff --git a/application/src/main/java/org/thingsboard/server/actors/shared/AbstractContextAwareMsgProcessor.java b/application/src/main/java/org/thingsboard/server/actors/shared/AbstractContextAwareMsgProcessor.java index 838bf0ea36..17f432eca5 100644 --- a/application/src/main/java/org/thingsboard/server/actors/shared/AbstractContextAwareMsgProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/shared/AbstractContextAwareMsgProcessor.java @@ -21,6 +21,7 @@ import org.thingsboard.server.actors.TbActorCtx; import org.thingsboard.server.common.msg.TbActorMsg; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; @Slf4j public abstract class AbstractContextAwareMsgProcessor { @@ -36,8 +37,8 @@ public abstract class AbstractContextAwareMsgProcessor { return systemContext.getScheduler(); } - protected void schedulePeriodicMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs, long periodInMs) { - systemContext.schedulePeriodicMsgWithDelay(ctx, msg, delayInMs, periodInMs); + protected ScheduledFuture schedulePeriodicMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs, long periodInMs) { + return systemContext.schedulePeriodicMsgWithDelay(ctx, msg, delayInMs, periodInMs); } protected void scheduleMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs) { diff --git a/application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java b/application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java index d48a8f7684..76e5ce7899 100644 --- a/application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java @@ -27,6 +27,8 @@ import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; import org.thingsboard.server.common.msg.queue.RuleNodeException; +import java.util.concurrent.ScheduledFuture; + @Slf4j public abstract class ComponentMsgProcessor extends AbstractContextAwareMsgProcessor { @@ -77,8 +79,8 @@ public abstract class ComponentMsgProcessor extends Abstract start(context); } - public void scheduleStatsPersistTick(TbActorCtx context, long statsPersistFrequency) { - schedulePeriodicMsgWithDelay(context, StatsPersistTick.INSTANCE, statsPersistFrequency, statsPersistFrequency); + public ScheduledFuture scheduleStatsPersistTick(TbActorCtx context, long statsPersistFrequency) { + return schedulePeriodicMsgWithDelay(context, StatsPersistTick.INSTANCE, statsPersistFrequency, statsPersistFrequency); } protected boolean checkMsgValid(TbMsg tbMsg) { diff --git a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java index f9eec324fc..c1a0980623 100644 --- a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java @@ -112,7 +112,7 @@ public class TenantActor extends RuleChainManagerActor { cantFindTenant = true; } } else { - log.info("Tenant {} is not managed by current service, skipping rule chains init", tenantId); + log.info("Tenant {} is not managed by current service, skipping rule chains and cf actor init", tenantId); } } log.debug("[{}] Tenant actor started.", tenantId); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java index ad248cc3d4..3232a9b89f 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java @@ -15,15 +15,12 @@ */ package org.thingsboard.server.service.queue; -import com.google.common.util.concurrent.ListeningExecutorService; -import com.google.common.util.concurrent.MoreExecutors; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; -import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.calculatedField.CalculatedFieldLinkedTelemetryMsg; import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg; @@ -35,7 +32,7 @@ import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; @@ -60,7 +57,6 @@ import org.thingsboard.server.service.queue.processing.IdMsgPair; import org.thingsboard.server.service.security.auth.jwt.settings.JwtSettingsService; import java.util.List; -import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -220,7 +216,7 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer protected void handleNotification(UUID id, TbProtoQueueMsg msg, TbCallback callback) { ToCalculatedFieldNotificationMsg toCfNotification = msg.getValue(); if (toCfNotification.hasComponentLifecycleMsg()) { - // from upstream (maybe removed since we don't need to init state for each partition) + handleComponentLifecycleMsg(id, ProtoUtils.fromProto(toCfNotification.getComponentLifecycleMsg())); log.trace("[{}] Forwarding component lifecycle message for processing {}", id, toCfNotification.getComponentLifecycleMsg()); forwardToActorSystem(toCfNotification.getComponentLifecycleMsg(), callback); } else if (toCfNotification.hasLinkedTelemetryMsg()) { diff --git a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java index 2c8c942ba3..7fee411704 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java @@ -351,7 +351,7 @@ public class DefaultTransportApiService implements TransportApiService { device.setAdditionalInfo(additionalInfo); device = deviceService.saveDevice(device); - relationService.saveRelation(TenantId.SYS_TENANT_ID, new EntityRelation(gateway.getId(), device.getId(), "Created")); + relationService.saveRelation(tenantId, new EntityRelation(gateway.getId(), device.getId(), "Created")); TbMsgMetaData metaData = new TbMsgMetaData(); CustomerId customerId = gateway.getCustomerId(); diff --git a/application/src/test/java/org/thingsboard/server/actors/service/ComponentActorTest.java b/application/src/test/java/org/thingsboard/server/actors/service/ComponentActorTest.java new file mode 100644 index 0000000000..49bf3c8670 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/actors/service/ComponentActorTest.java @@ -0,0 +1,74 @@ +/** + * Copyright © 2016-2025 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.actors.service; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.BDDMockito; +import org.mockito.Mockito; +import org.springframework.test.util.ReflectionTestUtils; +import org.thingsboard.server.actors.ActorSystemContext; +import org.thingsboard.server.actors.shared.ComponentMsgProcessor; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.msg.TbActorStopReason; + +import java.util.concurrent.ScheduledFuture; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; + +class ComponentActorTest { + ComponentActor componentActor; + + @BeforeEach + void setUp() { + componentActor = Mockito.mock(ComponentActor.class); + } + + @Test + void scheduleStatsPersistTickTest() { + Assertions.assertNull(componentActor.statsScheduledFuture); + ScheduledFuture statsScheduledFuture = Mockito.mock(ScheduledFuture.class); + ActorSystemContext systemContext = Mockito.mock(ActorSystemContext.class); + ReflectionTestUtils.setField(componentActor, "systemContext", systemContext); + ComponentMsgProcessor processor = Mockito.mock(ComponentMsgProcessor.class); + componentActor.processor = processor; + BDDMockito.willReturn(statsScheduledFuture).given(processor).scheduleStatsPersistTick(any(), anyLong()); + BDDMockito.willCallRealMethod().given(componentActor).scheduleStatsPersistTick(); + + componentActor.scheduleStatsPersistTick(); + + Assertions.assertNotNull(componentActor.statsScheduledFuture); + } + + @Test + void destroyTest() { + ScheduledFuture statsScheduledFuture = Mockito.mock(ScheduledFuture.class); + componentActor.statsScheduledFuture = statsScheduledFuture; + Assertions.assertNotNull(componentActor.statsScheduledFuture); + Throwable cause = new Throwable(); + EntityId id = Mockito.mock(EntityId.class); + ReflectionTestUtils.setField(componentActor, "id", id); + BDDMockito.willCallRealMethod().given(componentActor).destroy(any(), any()); + + componentActor.destroy(TbActorStopReason.STOPPED, cause); + + Mockito.verify(statsScheduledFuture).cancel(false); + Assertions.assertNull(componentActor.statsScheduledFuture); + } + +} diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java index e90aecb41e..33f2209eca 100644 --- a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java @@ -38,6 +38,7 @@ import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.mock.mockito.MockBean; import org.springframework.boot.test.mock.mockito.SpyBean; import org.springframework.cache.Cache; import org.springframework.cache.CacheManager; @@ -143,6 +144,7 @@ import org.thingsboard.server.dao.device.ClaimDevicesService; import org.thingsboard.server.dao.tenant.TenantProfileService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.queue.memory.InMemoryStorage; +import org.thingsboard.server.service.cf.CfRocksDb; import org.thingsboard.server.service.entitiy.tenant.profile.TbTenantProfileService; import org.thingsboard.server.service.security.auth.jwt.RefreshTokenRequest; import org.thingsboard.server.service.security.auth.rest.LoginRequest; @@ -276,6 +278,9 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest { @Autowired protected InMemoryStorage storage; + @MockBean + protected CfRocksDb cfRocksDb; + @Rule public TestRule watcher = new TestWatcher() { protected void starting(Description description) { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/edqs/fields/EntityFields.java b/common/data/src/main/java/org/thingsboard/server/common/data/edqs/fields/EntityFields.java index 532c4a92ac..1b0975542c 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/edqs/fields/EntityFields.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/edqs/fields/EntityFields.java @@ -148,6 +148,7 @@ public interface EntityFields { default String getAsString(String key) { return switch (key) { case "createdTime" -> Long.toString(getCreatedTime()); + case "title" -> getName(); case "type" -> getType(); case "label" -> getLabel(); case "additionalInfo" -> getAdditionalInfo();