dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #12736 from smatvienko-tb/hotfix/stats_scheduled_future_cancel_on_actor_destroy

Browse Source 
StatsPersistTick scheduled future cancel on actor destroy
This commit is contained in:
Viacheslav Klimov 2025-03-13 11:58:23 +02:00 committed by GitHub
commit 6ad2b1b4db
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 17 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
View File

@ -127,6 +127,7 @@ import java.io.StringWriter;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@Slf4j @Slf4j
@ -763,9 +764,9 @@ public class ActorSystemContext {
appActor.tellWithHighPriority(tbActorMsg); 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); 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) { public void scheduleMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs) {

8
application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java
View File

@ -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.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import java.util.Optional;
import java.util.concurrent.ScheduledFuture;
/** /**
* @author Andrew Shvayka * @author Andrew Shvayka
*/ */
@ -41,6 +44,7 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
protected P processor; protected P processor;
private long messagesProcessed; private long messagesProcessed;
private long errorsOccurred; private long errorsOccurred;
ScheduledFuture<?> statsScheduledFuture = null;
public ComponentActor(ActorSystemContext systemContext, TenantId tenantId, T id) { public ComponentActor(ActorSystemContext systemContext, TenantId tenantId, T id) {
super(systemContext); super(systemContext);
@ -75,7 +79,7 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
private void scheduleStatsPersistTick() { private void scheduleStatsPersistTick() {
try { try {
processor.scheduleStatsPersistTick(ctx, systemContext.getStatisticsPersistFrequency()); this.statsScheduledFuture = processor.scheduleStatsPersistTick(ctx, systemContext.getStatisticsPersistFrequency());
} catch (Exception e) { } catch (Exception e) {
log.error("[{}][{}] Failed to schedule statistics store message. No statistics is going to be stored: {}", tenantId, id, e.getMessage()); log.error("[{}][{}] Failed to schedule statistics store message. No statistics is going to be stored: {}", tenantId, id, e.getMessage());
logAndPersist("onScheduleStatsPersistMsg", e); logAndPersist("onScheduleStatsPersistMsg", e);
@ -90,6 +94,8 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
processor.stop(ctx); processor.stop(ctx);
} }
logLifecycleEvent(ComponentLifecycleEvent.STOPPED); logLifecycleEvent(ComponentLifecycleEvent.STOPPED);
Optional.ofNullable(statsScheduledFuture).ifPresent(x -> x.cancel(false));
statsScheduledFuture = null;
} catch (Exception e) { } catch (Exception e) {
log.warn("[{}][{}] Failed to stop {} processor: {}", tenantId, id, id.getEntityType(), e.getMessage()); log.warn("[{}][{}] Failed to stop {} processor: {}", tenantId, id, id.getEntityType(), e.getMessage());
logAndPersist("OnStop", e, true); logAndPersist("OnStop", e, true);

5
application/src/main/java/org/thingsboard/server/actors/shared/AbstractContextAwareMsgProcessor.java
View File

@ -21,6 +21,7 @@ import org.thingsboard.server.actors.TbActorCtx;
import org.thingsboard.server.common.msg.TbActorMsg; import org.thingsboard.server.common.msg.TbActorMsg;
import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
@Slf4j @Slf4j
public abstract class AbstractContextAwareMsgProcessor { public abstract class AbstractContextAwareMsgProcessor {
@ -36,8 +37,8 @@ public abstract class AbstractContextAwareMsgProcessor {
return systemContext.getScheduler(); return systemContext.getScheduler();
} }
protected void schedulePeriodicMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs, long periodInMs) { protected ScheduledFuture<?> schedulePeriodicMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs, long periodInMs) {
systemContext.schedulePeriodicMsgWithDelay(ctx, msg, delayInMs, periodInMs); return systemContext.schedulePeriodicMsgWithDelay(ctx, msg, delayInMs, periodInMs);
} }
protected void scheduleMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs) { protected void scheduleMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs) {

6
application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java
View File

@ -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.PartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.RuleNodeException; import org.thingsboard.server.common.msg.queue.RuleNodeException;
import java.util.concurrent.ScheduledFuture;
@Slf4j @Slf4j
public abstract class ComponentMsgProcessor<T extends EntityId> extends AbstractContextAwareMsgProcessor { public abstract class ComponentMsgProcessor<T extends EntityId> extends AbstractContextAwareMsgProcessor {
@ -77,8 +79,8 @@ public abstract class ComponentMsgProcessor<T extends EntityId> extends Abstract
start(context); start(context);
} }
public void scheduleStatsPersistTick(TbActorCtx context, long statsPersistFrequency) { public ScheduledFuture<?> scheduleStatsPersistTick(TbActorCtx context, long statsPersistFrequency) {
schedulePeriodicMsgWithDelay(context, StatsPersistTick.INSTANCE, statsPersistFrequency, statsPersistFrequency); return schedulePeriodicMsgWithDelay(context, StatsPersistTick.INSTANCE, statsPersistFrequency, statsPersistFrequency);
} }
protected boolean checkMsgValid(TbMsg tbMsg) { protected boolean checkMsgValid(TbMsg tbMsg) {