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

HotFix - fixed init of rule chains - init only on APP_INIT msg

Browse Source
This commit is contained in:
Volodymyr Babak 2023-06-22 16:37:03 +03:00
parent b14459dd9b
commit 371cab26d2
5 changed files with 49 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
View File

@ -73,10 +73,14 @@ public class AppActor extends ContextAwareActor {
@Override
protected boolean doProcess(TbActorMsg msg) {
if (!ruleChainsInitialized) {
initTenantActors();
ruleChainsInitialized = true;
if (msg.getMsgType() != MsgType.APP_INIT_MSG && msg.getMsgType() != MsgType.PARTITION_CHANGE_MSG) {
log.warn("Rule Chains initialized by unexpected message: {}", msg);
if (MsgType.APP_INIT_MSG.equals(msg.getMsgType())) {
initTenantActors();
ruleChainsInitialized = true;
} else {
if (!msg.getMsgType().isIgnoreOnStart()) {
log.warn("Attempt to initialize Rule Chains by unexpected message: {}", msg);
}
return true;
}
}
switch (msg.getMsgType()) {

16
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java
View File

@ -259,7 +259,21 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
}
void launchConsumer(TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer, Queue configuration, TbRuleEngineConsumerStats stats, String threadSuffix) {
consumersExecutor.execute(() -> consumerLoop(consumer, configuration, stats, threadSuffix));
if (isReady) {
consumersExecutor.execute(() -> consumerLoop(consumer, configuration, stats, threadSuffix));
} else {
scheduleLaunchConsumer(consumer, configuration, stats, threadSuffix);
}
}
private void scheduleLaunchConsumer(TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer, Queue configuration, TbRuleEngineConsumerStats stats, String threadSuffix) {
repartitionExecutor.schedule(() -> {
if (isReady) {
consumersExecutor.execute(() -> consumerLoop(consumer, configuration, stats, threadSuffix));
} else {
scheduleLaunchConsumer(consumer, configuration, stats, threadSuffix);
}
}, 10, TimeUnit.SECONDS);
}
void consumerLoop(TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer, org.thingsboard.server.common.data.queue.Queue configuration, TbRuleEngineConsumerStats stats, String threadSuffix) {

3
application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java
View File

@ -68,7 +68,7 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
protected volatile ExecutorService consumersExecutor;
protected volatile ExecutorService notificationsConsumerExecutor;
protected volatile boolean stopped = false;
protected volatile boolean isReady = false;
protected final ActorSystemContext actorContext;
protected final DataDecodingEncodingService encodingService;
protected final TbTenantProfileCache tenantProfileCache;
@ -108,6 +108,7 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
public void onApplicationEvent(ApplicationReadyEvent event) {
log.info("Subscribing to notifications: {}", nfConsumer.getTopic());
this.nfConsumer.subscribe();
this.isReady = true;
launchNotificationsConsumer();
launchMainConsumers();
}

15
common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java
View File

@ -15,6 +15,7 @@
*/
package org.thingsboard.server.common.msg;
import lombok.Getter;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
@ -28,7 +29,7 @@ public enum MsgType {
*
* See {@link PartitionChangeMsg}
*/
PARTITION_CHANGE_MSG,
PARTITION_CHANGE_MSG(true),
APP_INIT_MSG,
@ -108,7 +109,7 @@ public enum MsgType {
* Message that is sent from the Device Actor to Rule Engine. Requires acknowledgement
*/
SESSION_TIMEOUT_MSG,
SESSION_TIMEOUT_MSG(true),
STATS_PERSIST_TICK_MSG,
@ -130,4 +131,14 @@ public enum MsgType {
EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG,
EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG;
@Getter
private final boolean ignoreOnStart;
MsgType() {
this.ignoreOnStart = false;
}
MsgType(boolean ignoreOnStart) {
this.ignoreOnStart = ignoreOnStart;
}
}

14
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java
View File

@ -76,6 +76,10 @@ public class TbDeviceProfileNode implements TbNode {
this.ctx = ctx;
scheduleAlarmHarvesting(ctx, null);
ctx.addDeviceProfileListeners(this::onProfileUpdate, this::onDeviceUpdate);
initAlarmRuleState(false);
}
private void initAlarmRuleState(boolean printNewlyAddedDeviceStates) {
if (config.isFetchAlarmRulesStateOnStart()) {
log.info("[{}] Fetching alarm rule state", ctx.getSelfId());
int fetchCount = 0;
@ -86,7 +90,7 @@ public class TbDeviceProfileNode implements TbNode {
for (RuleNodeState rns : states.getData()) {
fetchCount++;
if (rns.getEntityId().getEntityType().equals(EntityType.DEVICE) && ctx.isLocalEntity(rns.getEntityId())) {
getOrCreateDeviceState(ctx, new DeviceId(rns.getEntityId().getId()), rns);
getOrCreateDeviceState(ctx, new DeviceId(rns.getEntityId().getId()), rns, printNewlyAddedDeviceStates);
}
}
}
@ -130,7 +134,7 @@ public class TbDeviceProfileNode implements TbNode {
removeDeviceState(deviceId);
ctx.tellSuccess(msg);
} else {
DeviceState deviceState = getOrCreateDeviceState(ctx, deviceId, null);
DeviceState deviceState = getOrCreateDeviceState(ctx, deviceId, null, false);
if (deviceState != null) {
deviceState.process(ctx, msg);
} else {
@ -148,6 +152,7 @@ public class TbDeviceProfileNode implements TbNode {
public void onPartitionChangeMsg(TbContext ctx, PartitionChangeMsg msg) {
// Cleanup the cache for all entities that are no longer assigned to current server partitions
deviceStates.entrySet().removeIf(entry -> !ctx.isLocalEntity(entry.getKey()));
initAlarmRuleState(true);
}
@Override
@ -156,13 +161,16 @@ public class TbDeviceProfileNode implements TbNode {
deviceStates.clear();
}
protected DeviceState getOrCreateDeviceState(TbContext ctx, DeviceId deviceId, RuleNodeState rns) {
protected DeviceState getOrCreateDeviceState(TbContext ctx, DeviceId deviceId, RuleNodeState rns, boolean printNewlyAddedDeviceStates) {
DeviceState deviceState = deviceStates.get(deviceId);
if (deviceState == null) {
DeviceProfile deviceProfile = cache.get(ctx.getTenantId(), deviceId);
if (deviceProfile != null) {
deviceState = new DeviceState(ctx, config, deviceId, new ProfileState(deviceProfile), rns);
deviceStates.put(deviceId, deviceState);
if (printNewlyAddedDeviceStates) {
log.info("[{}][{}] Device [{}] was added during PartitionChangeMsg", ctx.getTenantId(), ctx.getSelfId(), deviceId);
}
}
}
return deviceState;