diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java index 5ac230096b..020fa7f61e 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java @@ -132,6 +132,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } else { removeState(cfId); } + msg.getCallback().onSuccess(); } public void process(CalculatedFieldStatePartitionRestoreMsg msg) { 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 b2fe6d2fd9..5d3965feda 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 @@ -167,10 +167,13 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware if (ctx != null) { msg.setCtx(ctx); - log.debug("Pushing CF state restore msg to specific actor [{}]", msg.getId().entityId()); + log.debug("[{}] Pushing CF state restore msg to specific actor [{}]", tenantId, msg.getId().entityId()); getOrCreateActor(msg.getId().entityId()).tellWithHighPriority(msg); - } else { + } else if (msg.getState() != null) { + log.debug("[{}] Received CF state restore msg for non-existing CF [{}]. Removing state", tenantId, cfId); cfStateService.deleteState(msg.getId(), msg.getCallback()); + } else { + msg.getCallback().onSuccess(); } } diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldStateRestoreMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldStateRestoreMsg.java index d1c2f11aeb..3969afbabf 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldStateRestoreMsg.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldStateRestoreMsg.java @@ -19,6 +19,7 @@ import lombok.Data; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.MsgType; import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; @@ -30,6 +31,7 @@ public class CalculatedFieldStateRestoreMsg implements ToCalculatedFieldSystemMs private final CalculatedFieldEntityCtxId id; private final CalculatedFieldState state; private final TopicPartitionInfo partition; + private final TbCallback callback; private CalculatedFieldCtx ctx; @Override @@ -41,4 +43,5 @@ public class CalculatedFieldStateRestoreMsg implements ToCalculatedFieldSystemMs public TenantId getTenantId() { return id.tenantId(); } + } 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 11a8651026..be7826df2b 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 @@ -43,7 +43,6 @@ 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.data.rule.RuleChainType; -import org.thingsboard.server.common.msg.MsgType; import org.thingsboard.server.common.msg.TbActorMsg; import org.thingsboard.server.common.msg.TbActorStopReason; import org.thingsboard.server.common.msg.TbMsg; @@ -139,13 +138,22 @@ public class TenantActor extends RuleChainManagerActor { @Override protected boolean doProcess(TbActorMsg msg) { if (cantFindTenant) { - log.info("[{}] Processing missing Tenant msg: {}", tenantId, msg); - if (msg.getMsgType().equals(MsgType.QUEUE_TO_RULE_ENGINE_MSG)) { - QueueToRuleEngineMsg queueMsg = (QueueToRuleEngineMsg) msg; - queueMsg.getMsg().getCallback().onSuccess(); - } else if (msg.getMsgType().equals(MsgType.TRANSPORT_TO_DEVICE_ACTOR_MSG)) { - TransportToDeviceActorMsgWrapper transportMsg = (TransportToDeviceActorMsgWrapper) msg; - transportMsg.getCallback().onSuccess(); + log.debug("[{}] Processing message for non-existing tenant: {}", tenantId, msg); + switch (msg.getMsgType()) { + case QUEUE_TO_RULE_ENGINE_MSG -> { + ((QueueToRuleEngineMsg) msg).getMsg().getCallback().onSuccess(); + } + case TRANSPORT_TO_DEVICE_ACTOR_MSG -> { + ((TransportToDeviceActorMsgWrapper) msg).getCallback().onSuccess(); + } + case CF_STATE_RESTORE_MSG -> { + ((CalculatedFieldStateRestoreMsg) msg).getCallback().onSuccess(); + } + default -> { + if (!log.isDebugEnabled()) { + log.info("[{}] Processing message for non-existing tenant: {}", tenantId, msg); + } + } } return true; } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java index dd0bf45eb9..81cc9c4087 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java @@ -68,7 +68,7 @@ public abstract class AbstractCalculatedFieldStateService implements CalculatedF protected abstract void doRemove(CalculatedFieldEntityCtxId stateId, TbCallback callback); - protected void processRestoredState(CalculatedFieldStateProto stateMsg, TopicPartitionInfo partition) { + protected void processRestoredState(CalculatedFieldStateProto stateMsg, TopicPartitionInfo partition, TbCallback callback) { var id = fromProto(stateMsg.getId()); if (partition == null) { try { @@ -79,12 +79,12 @@ public abstract class AbstractCalculatedFieldStateService implements CalculatedF } } var state = fromProto(id, stateMsg); - processRestoredState(id, state, partition); + processRestoredState(id, state, partition, callback); } - protected void processRestoredState(CalculatedFieldEntityCtxId id, CalculatedFieldState state, TopicPartitionInfo partition) { + protected void processRestoredState(CalculatedFieldEntityCtxId id, CalculatedFieldState state, TopicPartitionInfo partition, TbCallback callback) { partition = partition.withTopic(DataConstants.CF_STATES_QUEUE_NAME); - actorSystemContext.tellWithHighPriority(new CalculatedFieldStateRestoreMsg(id, state, partition)); + actorSystemContext.tellWithHighPriority(new CalculatedFieldStateRestoreMsg(id, state, partition, callback)); } @Override diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java index ffc3c1584b..016db006dc 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java @@ -43,6 +43,8 @@ import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory; import org.thingsboard.server.service.cf.AbstractCalculatedFieldStateService; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import static org.thingsboard.server.queue.common.AbstractTbQueueTemplate.bytesToString; @@ -61,6 +63,8 @@ public class KafkaCalculatedFieldStateService extends AbstractCalculatedFieldSta @Value("${queue.calculated_fields.poll_interval:25}") private long pollInterval; + @Value("${queue.calculated_fields.pack_processing_timeout:60000}") + private long packProcessingTimeout; private TbKafkaProducerTemplate> stateProducer; @@ -74,21 +78,39 @@ public class KafkaCalculatedFieldStateService extends AbstractCalculatedFieldSta .topic(partitionService.getTopic(queueKey)) .pollInterval(pollInterval) .msgPackProcessor((msgs, consumer, consumerKey, config) -> { + CountDownLatch completionLatch = new CountDownLatch(msgs.size()); for (TbProtoQueueMsg msg : msgs) { + TbCallback callback = new TbCallback() { + @Override + public void onSuccess() { + int processedMsgCount = counter.incrementAndGet(); + if (processedMsgCount % 10000 == 0) { + log.info("Processed {} CF state messages", processedMsgCount); + } + completionLatch.countDown(); + } + + @Override + public void onFailure(Throwable t) { + log.error("Failed to process CF state message: {}", msg, t); + completionLatch.countDown(); + } + }; + try { if (msg.getValue() != null) { - processRestoredState(msg.getValue(), consumerKey.partition()); + processRestoredState(msg.getValue(), consumerKey.partition(), callback); } else { - processRestoredState(getStateId(msg.getHeaders()), null, consumerKey.partition()); + processRestoredState(getStateId(msg.getHeaders()), null, consumerKey.partition(), callback); } } catch (Throwable t) { - log.error("Failed to process state message: {}", msg, t); + callback.onFailure(t); } + } - int processedMsgCount = counter.incrementAndGet(); - if (processedMsgCount % 10000 == 0) { - log.info("Processed {} calculated field state msgs", processedMsgCount); - } + boolean success = completionLatch.await(packProcessingTimeout, TimeUnit.MILLISECONDS); + if (!success) { + log.error("Timeout to process CF state messages pack of size {}", msgs.size()); } }) .consumerCreator((queueConfig, tpi) -> queueFactory.createCalculatedFieldStateConsumer()) diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java index 05bfb8b717..26fffe6d5f 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.service.cf.ctx.state; +import com.google.protobuf.InvalidProtocolBufferException; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; @@ -62,11 +63,22 @@ public class RocksDBCalculatedFieldStateService extends AbstractCalculatedFieldS public void restore(QueueKey queueKey, Set partitions) { if (stateService.getPartitions().isEmpty()) { cfRocksDb.forEach((key, value) -> { + CalculatedFieldStateProto stateMsg; try { - processRestoredState(CalculatedFieldStateProto.parseFrom(value), null); - } catch (Exception e) { - log.error("[{}] Failed to process restored state", key, e); + stateMsg = CalculatedFieldStateProto.parseFrom(value); + } catch (InvalidProtocolBufferException e) { + log.error("Failed to parse CalculatedFieldStateProto for key {}", key, e); + return; } + processRestoredState(stateMsg, null, new TbCallback() { + @Override + public void onSuccess() {} + + @Override + public void onFailure(Throwable t) { + log.error("Failed to process CF state message: {}", stateMsg, t); + } + }); }); } super.restore(queueKey, partitions);