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 9c0e3bc80f..9a72b0792a 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 @@ -21,6 +21,9 @@ import org.thingsboard.server.actors.calculatedField.CalculatedFieldStateRestore import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.exception.CalculatedFieldStateException; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.common.consumer.PartitionedQueueConsumerManager; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; @@ -32,6 +35,13 @@ public abstract class AbstractCalculatedFieldStateService implements CalculatedF @Autowired private ActorSystemContext actorSystemContext; + protected PartitionedQueueConsumerManager> eventConsumer; + + @Override + public void init(PartitionedQueueConsumerManager> eventConsumer) { + this.eventConsumer = eventConsumer; + } + @Override public final void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) { if (state.isSizeExceedsLimit()) { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldStateService.java index df2b04c20a..18d5518da3 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldStateService.java @@ -18,6 +18,9 @@ package org.thingsboard.server.service.cf; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.exception.CalculatedFieldStateException; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.common.consumer.PartitionedQueueConsumerManager; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; @@ -25,10 +28,14 @@ import java.util.Set; public interface CalculatedFieldStateService { + void init(PartitionedQueueConsumerManager> eventConsumer); + void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) throws CalculatedFieldStateException; void removeState(CalculatedFieldEntityCtxId stateId, TbCallback callback); void restore(Set partitions); + void stop(); + } 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 959522ca63..a9cee454f2 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 @@ -15,15 +15,11 @@ */ package org.thingsboard.server.service.cf.ctx.state; -import jakarta.annotation.PostConstruct; -import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Service; -import org.thingsboard.common.util.ThingsBoardExecutors; -import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityIdFactory; @@ -31,25 +27,22 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueMsgHeaders; import org.thingsboard.server.queue.TbQueueMsgMetadata; import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.common.consumer.PartitionedQueueConsumerManager; +import org.thingsboard.server.queue.common.consumer.QueueStateService; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.QueueKey; import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate; import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory; import org.thingsboard.server.service.cf.AbstractCalculatedFieldStateService; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; -import org.thingsboard.server.service.queue.DefaultTbCalculatedFieldConsumerService.CalculatedFieldQueueConfig; -import org.thingsboard.server.service.queue.consumer.MainQueueConsumerManager; import java.util.Set; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.atomic.AtomicInteger; -import java.util.stream.Collectors; import static org.thingsboard.server.queue.common.AbstractTbQueueTemplate.bytesToString; import static org.thingsboard.server.queue.common.AbstractTbQueueTemplate.bytesToUuid; @@ -67,27 +60,20 @@ public class KafkaCalculatedFieldStateService extends AbstractCalculatedFieldSta @Value("${queue.calculated_fields.poll_interval:25}") private long pollInterval; - @Value("${queue.calculated_fields.consumer_per_partition:true}") - private boolean consumerPerPartition; - private MainQueueConsumerManager, CalculatedFieldQueueConfig> stateConsumer; + private PartitionedQueueConsumerManager> stateConsumer; private TbKafkaProducerTemplate> stateProducer; - - protected ExecutorService consumersExecutor; - protected ExecutorService mgmtExecutor; - protected ScheduledExecutorService scheduler; + private QueueStateService, TbProtoQueueMsg> queueStateService; private final AtomicInteger counter = new AtomicInteger(); - @PostConstruct - private void init() { - this.consumersExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("cf-state-consumer")); - this.mgmtExecutor = ThingsBoardExecutors.newWorkStealingPool(Math.max(Runtime.getRuntime().availableProcessors(), 4), "cf-state-mgmt"); - this.scheduler = ThingsBoardExecutors.newSingleThreadScheduledExecutor("cf-state-consumer-scheduler"); - - this.stateConsumer = MainQueueConsumerManager., CalculatedFieldQueueConfig>builder() + @Override + public void init(PartitionedQueueConsumerManager> eventConsumer) { + super.init(eventConsumer); + this.stateConsumer = PartitionedQueueConsumerManager.>create() .queueKey(QueueKey.CF_STATES) - .config(CalculatedFieldQueueConfig.of(consumerPerPartition, (int) pollInterval)) + .topic(partitionService.getTopic(QueueKey.CF_STATES)) + .pollInterval(pollInterval) .msgPackProcessor((msgs, consumer, config) -> { for (TbProtoQueueMsg msg : msgs) { try { @@ -107,11 +93,13 @@ public class KafkaCalculatedFieldStateService extends AbstractCalculatedFieldSta } }) .consumerCreator((config, partitionId) -> queueFactory.createCalculatedFieldStateConsumer()) - .consumerExecutor(consumersExecutor) - .scheduler(scheduler) - .taskExecutor(mgmtExecutor) + .consumerExecutor(eventConsumer.getConsumerExecutor()) + .scheduler(eventConsumer.getScheduler()) + .taskExecutor(eventConsumer.getTaskExecutor()) .build(); this.stateProducer = (TbKafkaProducerTemplate>) queueFactory.createCalculatedFieldStateProducer(); + this.queueStateService = new QueueStateService<>(); + this.queueStateService.init(stateConsumer, super.eventConsumer); } @Override @@ -145,15 +133,7 @@ public class KafkaCalculatedFieldStateService extends AbstractCalculatedFieldSta @Override public void restore(Set partitions) { - partitions = partitions.stream().map(tpi -> tpi.newByTopic(partitionService.getTopic(QueueKey.CF_STATES))).collect(Collectors.toSet()); - log.info("Restoring calculated field states for partitions: {}", partitions.stream().map(TopicPartitionInfo::getFullTopicName).toList()); - long startTs = System.currentTimeMillis(); - counter.set(0); - - stateConsumer.doUpdate(partitions); // calling blocking doUpdate instead of update - stateConsumer.awaitStop(0);// consumers should stop on their own because stopWhenRead is true, we just need to wait - - log.info("Restored {} calculated field states in {} ms", counter.get(), System.currentTimeMillis() - startTs); + queueStateService.update(partitions); } private void putStateId(TbQueueMsgHeaders headers, CalculatedFieldEntityCtxId stateId) { @@ -170,15 +150,11 @@ public class KafkaCalculatedFieldStateService extends AbstractCalculatedFieldSta return new CalculatedFieldEntityCtxId(tenantId, cfId, entityId); } - @PreDestroy - private void preDestroy() { + @Override + public void stop() { stateConsumer.stop(); stateConsumer.awaitStop(); stateProducer.stop(); - - consumersExecutor.shutdownNow(); - mgmtExecutor.shutdownNow(); - scheduler.shutdownNow(); } } 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 61e752b9d1..f5f2051fb4 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 @@ -54,18 +54,21 @@ public class RocksDBCalculatedFieldStateService extends AbstractCalculatedFieldS @Override public void restore(Set partitions) { if (this.partitions == null) { - this.partitions = partitions; - } else { - return; + cfRocksDb.forEach((key, value) -> { + try { + processRestoredState(CalculatedFieldStateProto.parseFrom(value)); + } catch (InvalidProtocolBufferException e) { + log.error("[{}] Failed to process restored state", key, e); + } + }); } - cfRocksDb.forEach((key, value) -> { - try { - processRestoredState(CalculatedFieldStateProto.parseFrom(value)); - } catch (InvalidProtocolBufferException e) { - log.error("[{}] Failed to process restored state", key, e); - } - }); + eventConsumer.update(partitions); + this.partitions = partitions; + } + + @Override + public void stop() { } } 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 82ee4de7dc..d579d1e538 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 @@ -19,13 +19,11 @@ import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; -import lombok.Data; 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.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.calculatedField.CalculatedFieldLinkedTelemetryMsg; import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg; @@ -45,6 +43,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.common.consumer.PartitionedQueueConsumerManager; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.QueueKey; import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; @@ -55,7 +54,6 @@ import org.thingsboard.server.service.cf.CalculatedFieldCache; import org.thingsboard.server.service.cf.CalculatedFieldStateService; import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; -import org.thingsboard.server.service.queue.consumer.MainQueueConsumerManager; import org.thingsboard.server.service.queue.processing.AbstractConsumerService; import org.thingsboard.server.service.queue.processing.IdMsgPair; import org.thingsboard.server.service.security.auth.jwt.settings.JwtSettingsService; @@ -66,8 +64,6 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -83,18 +79,15 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer private long pollInterval; @Value("${queue.calculated_fields.pack_processing_timeout:60000}") private long packProcessingTimeout; - @Value("${queue.calculated_fields.consumer_per_partition:true}") - private boolean consumerPerPartition; @Value("${queue.calculated_fields.pool_size:8}") private int poolSize; private final TbRuleEngineQueueFactory queueFactory; private final CalculatedFieldStateService stateService; - private MainQueueConsumerManager, CalculatedFieldQueueConfig> mainConsumer; + private PartitionedQueueConsumerManager> eventConsumer; private ListeningExecutorService calculatedFieldsExecutor; - private ExecutorService repartitionExecutor; public DefaultTbCalculatedFieldConsumerService(TbRuleEngineQueueFactory tbQueueFactory, ActorSystemContext actorContext, @@ -117,17 +110,18 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer public void init() { super.init("tb-cf"); this.calculatedFieldsExecutor = MoreExecutors.listeningDecorator(ThingsBoardExecutors.newWorkStealingPool(poolSize, "tb-cf-executor")); // TODO: multiple threads. - this.repartitionExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-cf-repartition")); - this.mainConsumer = MainQueueConsumerManager., CalculatedFieldQueueConfig>builder() + this.eventConsumer = PartitionedQueueConsumerManager.>create() .queueKey(QueueKey.CF) - .config(CalculatedFieldQueueConfig.of(consumerPerPartition, (int) pollInterval)) + .topic(partitionService.getTopic(QueueKey.CF)) + .pollInterval(pollInterval) .msgPackProcessor(this::processMsgs) .consumerCreator((config, partitionId) -> queueFactory.createToCalculatedFieldMsgConsumer()) .consumerExecutor(consumersExecutor) .scheduler(scheduler) .taskExecutor(mgmtExecutor) .build(); + stateService.init(eventConsumer); } @PreDestroy @@ -146,17 +140,16 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer @Override protected void onTbApplicationEvent(PartitionChangeEvent event) { var partitions = event.getCfPartitions(); - repartitionExecutor.submit(() -> { - try { - stateService.restore(partitions); - mainConsumer.update(partitions); - // Cleanup old entities after corresponding consumers are stopped. - // Any periodic tasks need to check that the entity is still managed by the current server before processing. - actorContext.tell(new CalculatedFieldPartitionChangeMsg(partitionsToBooleanIndexArray(partitions))); - } catch (Throwable t) { - log.error("Failed to process partition change event: {}", event, t); - } - }); + try { + stateService.restore(partitions); + // eventConsumer's partitions will be updated by stateService + + // Cleanup old entities after corresponding consumers are stopped. + // Any periodic tasks need to check that the entity is still managed by the current server before processing. + actorContext.tell(new CalculatedFieldPartitionChangeMsg(partitionsToBooleanIndexArray(partitions))); + } catch (Throwable t) { + log.error("Failed to process partition change event: {}", event, t); + } } private boolean[] partitionsToBooleanIndexArray(Set partitions) { @@ -167,7 +160,7 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer return myPartitions; } - private void processMsgs(List> msgs, TbQueueConsumer> consumer, CalculatedFieldQueueConfig config) throws Exception { + private void processMsgs(List> msgs, TbQueueConsumer> consumer, QueueConfig config) throws Exception { List> orderedMsgList = msgs.stream().map(msg -> new IdMsgPair<>(UUID.randomUUID(), msg)).toList(); ConcurrentMap> pendingMap = orderedMsgList.stream().collect( Collectors.toConcurrentMap(IdMsgPair::getUuid, IdMsgPair::getMsg)); @@ -275,14 +268,9 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer @Override protected void stopConsumers() { super.stopConsumers(); - mainConsumer.stop(); - mainConsumer.awaitStop(); - } - - @Data(staticConstructor = "of") - public static class CalculatedFieldQueueConfig implements QueueConfig { - private final boolean consumerPerPartition; - private final int pollInterval; + eventConsumer.stop(); + eventConsumer.awaitStop(); + stateService.stop(); } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index f47cff5b6c..7d4e776d5d 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -21,8 +21,6 @@ import com.google.common.util.concurrent.MoreExecutors; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.Data; -import lombok.Getter; -import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Value; @@ -90,7 +88,7 @@ import org.thingsboard.server.service.notification.NotificationSchedulerService; import org.thingsboard.server.service.ota.OtaPackageStateService; import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; -import org.thingsboard.server.service.queue.consumer.MainQueueConsumerManager; +import org.thingsboard.server.queue.common.consumer.MainQueueConsumerManager; import org.thingsboard.server.service.queue.processing.AbstractConsumerService; import org.thingsboard.server.service.queue.processing.IdMsgPair; import org.thingsboard.server.service.resource.TbImageService; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java index 6a6924b827..f276202ad0 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java @@ -20,8 +20,6 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import lombok.Data; -import lombok.Getter; -import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.checkerframework.checker.nullness.qual.Nullable; import org.jetbrains.annotations.NotNull; @@ -52,7 +50,7 @@ import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.edge.EdgeContextComponent; -import org.thingsboard.server.service.queue.consumer.MainQueueConsumerManager; +import org.thingsboard.server.queue.common.consumer.MainQueueConsumerManager; import org.thingsboard.server.service.queue.processing.AbstractConsumerService; import org.thingsboard.server.service.queue.processing.IdMsgPair; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerManagerTask.java b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerManagerTask.java deleted file mode 100644 index e5821df68d..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerManagerTask.java +++ /dev/null @@ -1,48 +0,0 @@ -/** - * 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.queue.ruleengine; - -import lombok.AllArgsConstructor; -import lombok.Getter; -import lombok.ToString; -import org.thingsboard.server.common.data.queue.QueueConfig; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; - -import java.util.Set; - -@Getter -@ToString -@AllArgsConstructor -public class TbQueueConsumerManagerTask { - - private final QueueEvent event; - private QueueConfig config; - private Set partitions; - private boolean drainQueue; - - public static TbQueueConsumerManagerTask delete(boolean drainQueue) { - return new TbQueueConsumerManagerTask(QueueEvent.DELETE, null, null, drainQueue); - } - - public static TbQueueConsumerManagerTask configUpdate(QueueConfig config) { - return new TbQueueConsumerManagerTask(QueueEvent.CONFIG_UPDATE, config, null, false); - } - - public static TbQueueConsumerManagerTask partitionChange(Set partitions) { - return new TbQueueConsumerManagerTask(QueueEvent.PARTITION_CHANGE, null, partitions, false); - } - -} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java index c2823d3c00..aebd012e2b 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java @@ -33,11 +33,14 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.common.consumer.MainQueueConsumerManager; +import org.thingsboard.server.queue.common.consumer.TbQueueConsumerManagerTask; +import org.thingsboard.server.queue.common.consumer.TbQueueConsumerManagerTask.DeleteQueueTask; +import org.thingsboard.server.queue.common.consumer.TbQueueConsumerTask; import org.thingsboard.server.queue.discovery.QueueKey; import org.thingsboard.server.service.queue.TbMsgPackCallback; import org.thingsboard.server.service.queue.TbMsgPackProcessingContext; import org.thingsboard.server.service.queue.TbRuleEngineConsumerStats; -import org.thingsboard.server.service.queue.consumer.MainQueueConsumerManager; import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingDecision; import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingResult; import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategy; @@ -75,13 +78,13 @@ public class TbRuleEngineQueueConsumerManager extends MainQueueConsumerManager i @Override public void commit() { if (consumerLock.isLocked()) { + if (stopped) { + return; + } log.error("commit. consumerLock is locked. will wait with no timeout. it looks like a race conditions or deadlock topic " + topic, new RuntimeException("stacktrace")); } consumerLock.lock(); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/consumer/MainQueueConsumerManager.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/MainQueueConsumerManager.java similarity index 83% rename from application/src/main/java/org/thingsboard/server/service/queue/consumer/MainQueueConsumerManager.java rename to common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/MainQueueConsumerManager.java index 5fa69695d2..f9ade10b1d 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/consumer/MainQueueConsumerManager.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/MainQueueConsumerManager.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.service.queue.consumer; +package org.thingsboard.server.queue.common.consumer; import lombok.Builder; import lombok.Getter; @@ -23,10 +23,9 @@ import org.thingsboard.server.common.data.queue.QueueConfig; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueMsg; +import org.thingsboard.server.queue.common.consumer.TbQueueConsumerManagerTask.UpdateConfigTask; +import org.thingsboard.server.queue.common.consumer.TbQueueConsumerManagerTask.UpdatePartitionsTask; import org.thingsboard.server.queue.discovery.QueueKey; -import org.thingsboard.server.service.queue.ruleengine.QueueEvent; -import org.thingsboard.server.service.queue.ruleengine.TbQueueConsumerManagerTask; -import org.thingsboard.server.service.queue.ruleengine.TbQueueConsumerTask; import java.util.Collection; import java.util.Collections; @@ -34,6 +33,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ExecutorService; @@ -42,6 +42,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.ReentrantLock; import java.util.function.BiFunction; +import java.util.function.Consumer; import java.util.stream.Collectors; @Slf4j @@ -52,8 +53,11 @@ public class MainQueueConsumerManager msgPackProcessor; protected final BiFunction> consumerCreator; + @Getter protected final ExecutorService consumerExecutor; + @Getter protected final ScheduledExecutorService scheduler; + @Getter protected final ExecutorService taskExecutor; private final java.util.Queue tasks = new ConcurrentLinkedQueue<>(); @@ -85,20 +89,24 @@ public class MainQueueConsumerManager createConsumerWrapper(C config) { if (config.isConsumerPerPartition()) { - this.consumerWrapper = new ConsumerPerPartitionWrapper(); + return new ConsumerPerPartitionWrapper(); } else { - this.consumerWrapper = new SingleConsumerWrapper(); + return new SingleConsumerWrapper(); } - log.debug("[{}] Initialized consumer for queue: {}", queueKey, config); } public void update(C config) { - addTask(TbQueueConsumerManagerTask.configUpdate(config)); + addTask(new UpdateConfigTask(config)); } public void update(Set partitions) { - addTask(TbQueueConsumerManagerTask.partitionChange(partitions)); + addTask(new UpdatePartitionsTask(partitions)); } protected void addTask(TbQueueConsumerManagerTask todo) { @@ -123,10 +131,10 @@ public class MainQueueConsumerManager partitions) { + private void doUpdate(Set partitions) { this.partitions = partitions; consumerWrapper.updatePartitions(partitions); } @@ -197,6 +205,15 @@ public class MainQueueConsumerManager consumerTask.awaitCompletion(timeoutSec)); log.debug("[{}] Unsubscribed and stopped consumers", queueKey); } - private static String partitionsToString(Collection partitions) { + static String partitionsToString(Collection partitions) { return partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.joining(", ", "[", "]")); } @@ -273,15 +290,24 @@ public class MainQueueConsumerManager removedPartitions = new HashSet<>(consumers.keySet()); removedPartitions.removeAll(partitions); + log.info("[{}] Added partitions: {}, removed partitions: {}", queueKey, partitionsToString(addedPartitions), partitionsToString(removedPartitions)); + removePartitions(removedPartitions); + addPartitions(addedPartitions, null); + } - removedPartitions.forEach((tpi) -> consumers.get(tpi).initiateStop()); - removedPartitions.forEach((tpi) -> consumers.remove(tpi).awaitCompletion()); + protected void removePartitions(Set removedPartitions) { + removedPartitions.forEach((tpi) -> Optional.ofNullable(consumers.get(tpi)).ifPresent(TbQueueConsumerTask::initiateStop)); + removedPartitions.forEach((tpi) -> Optional.ofNullable(consumers.remove(tpi)).ifPresent(TbQueueConsumerTask::awaitCompletion)); + } - addedPartitions.forEach((tpi) -> { + protected void addPartitions(Set partitions, Consumer onStop) { + partitions.forEach(tpi -> { Integer partitionId = tpi.getPartition().orElse(-1); String key = queueKey + "-" + partitionId; - TbQueueConsumerTask consumer = new TbQueueConsumerTask<>(key, () -> consumerCreator.apply(config, partitionId)); + Runnable callback = onStop != null ? () -> onStop.accept(tpi) : null; + + TbQueueConsumerTask consumer = new TbQueueConsumerTask<>(key, () -> consumerCreator.apply(config, partitionId), callback); consumers.put(tpi, consumer); consumer.subscribe(Set.of(tpi)); launchConsumer(consumer); @@ -310,7 +336,7 @@ public class MainQueueConsumerManager(queueKey, () -> consumerCreator.apply(config, null)); // no partitionId passed + consumer = new TbQueueConsumerTask<>(queueKey, () -> consumerCreator.apply(config, null), null); // no partitionId passed } consumer.subscribe(partitions); if (!consumer.isRunning()) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/PartitionedQueueConsumerManager.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/PartitionedQueueConsumerManager.java new file mode 100644 index 0000000000..f8ba1671ae --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/PartitionedQueueConsumerManager.java @@ -0,0 +1,75 @@ +/** + * 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.queue.common.consumer; + +import lombok.Builder; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.common.data.queue.QueueConfig; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueMsg; +import org.thingsboard.server.queue.common.consumer.TbQueueConsumerManagerTask.AddPartitionsTask; +import org.thingsboard.server.queue.common.consumer.TbQueueConsumerManagerTask.RemovePartitionsTask; +import org.thingsboard.server.queue.discovery.QueueKey; + +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; +import java.util.function.BiFunction; +import java.util.function.Consumer; + +@Slf4j +public class PartitionedQueueConsumerManager extends MainQueueConsumerManager { + + private final ConsumerPerPartitionWrapper consumerWrapper; + @Getter + private final String topic; + + @Builder(builderMethodName = "create") // not to conflict with super.builder() + public PartitionedQueueConsumerManager(QueueKey queueKey, String topic, long pollInterval, MsgPackProcessor msgPackProcessor, + BiFunction> consumerCreator, + ExecutorService consumerExecutor, ScheduledExecutorService scheduler, + ExecutorService taskExecutor) { + super(queueKey, QueueConfig.of(true, pollInterval), msgPackProcessor, consumerCreator, consumerExecutor, scheduler, taskExecutor); + this.topic = topic; + this.consumerWrapper = (ConsumerPerPartitionWrapper) super.consumerWrapper; + } + + @Override + protected void processTask(TbQueueConsumerManagerTask task) { + if (task instanceof AddPartitionsTask addPartitionsTask) { + log.info("[{}] Added partitions: {}", queueKey, partitionsToString(addPartitionsTask.partitions())); + consumerWrapper.addPartitions(addPartitionsTask.partitions(), addPartitionsTask.onStop()); + } else if (task instanceof RemovePartitionsTask removePartitionsTask) { + log.info("[{}] Removed partitions: {}", queueKey, partitionsToString(removePartitionsTask.partitions())); + consumerWrapper.removePartitions(removePartitionsTask.partitions()); + } + } + + public void addPartitions(Set partitions) { + addPartitions(partitions, null); + } + + public void addPartitions(Set partitions, Consumer onStop) { + addTask(new AddPartitionsTask(partitions, onStop)); + } + + public void removePartitions(Set partitions) { + addTask(new RemovePartitionsTask(partitions)); + } + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueStateService.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueStateService.java new file mode 100644 index 0000000000..0931b9ed5b --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueStateService.java @@ -0,0 +1,82 @@ +/** + * 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.queue.common.consumer; + +import lombok.Getter; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.queue.TbQueueMsg; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; +import java.util.stream.Collectors; + +public class QueueStateService { + + private PartitionedQueueConsumerManager stateConsumer; + private PartitionedQueueConsumerManager eventConsumer; + + @Getter + private Set partitions; + private final Lock lock = new ReentrantLock(); + + public void init(PartitionedQueueConsumerManager stateConsumer, PartitionedQueueConsumerManager eventConsumer) { + this.stateConsumer = stateConsumer; + this.eventConsumer = eventConsumer; + } + + public void update(Set newPartitions) { + newPartitions = withTopic(newPartitions, stateConsumer.getTopic()); + lock.lock(); + Set oldPartitions = this.partitions != null ? this.partitions : Collections.emptySet(); + Set addedPartitions; + Set removedPartitions; + try { + addedPartitions = new HashSet<>(newPartitions); + addedPartitions.removeAll(oldPartitions); + removedPartitions = new HashSet<>(oldPartitions); + removedPartitions.removeAll(newPartitions); + this.partitions = newPartitions; + } finally { + lock.unlock(); + } + + if (!removedPartitions.isEmpty()) { + stateConsumer.removePartitions(removedPartitions); + eventConsumer.removePartitions(withTopic(removedPartitions, eventConsumer.getTopic())); + } + + if (!addedPartitions.isEmpty()) { + stateConsumer.addPartitions(addedPartitions, partition -> { + lock.lock(); + try { + if (this.partitions.contains(partition)) { + eventConsumer.addPartitions(Set.of(partition.newByTopic(eventConsumer.getTopic()))); + } + } finally { + lock.unlock(); + } + }); + } + } + + private Set withTopic(Set partitions, String topic) { + return partitions.stream().map(tpi -> tpi.newByTopic(topic)).collect(Collectors.toSet()); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueTaskType.java similarity index 77% rename from application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java rename to common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueTaskType.java index 9e5766b374..4e218cd268 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueTaskType.java @@ -13,12 +13,13 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.service.queue.ruleengine; +package org.thingsboard.server.queue.common.consumer; import java.io.Serializable; -public enum QueueEvent implements Serializable { +public enum QueueTaskType implements Serializable { - PARTITION_CHANGE, CONFIG_UPDATE, DELETE + UPDATE_PARTITIONS, UPDATE_CONFIG, DELETE, + ADD_PARTITIONS, REMOVE_PARTITIONS } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/TbQueueConsumerManagerTask.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/TbQueueConsumerManagerTask.java new file mode 100644 index 0000000000..7e2c848a66 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/TbQueueConsumerManagerTask.java @@ -0,0 +1,63 @@ +/** + * 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.queue.common.consumer; + +import org.thingsboard.server.common.data.queue.QueueConfig; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; + +import java.util.Set; +import java.util.function.Consumer; + +public interface TbQueueConsumerManagerTask { + + QueueTaskType getType(); + + record DeleteQueueTask(boolean drainQueue) implements TbQueueConsumerManagerTask { + @Override + public QueueTaskType getType() { + return QueueTaskType.DELETE; + } + } + + record UpdateConfigTask(QueueConfig config) implements TbQueueConsumerManagerTask { + @Override + public QueueTaskType getType() { + return QueueTaskType.UPDATE_CONFIG; + } + } + + record UpdatePartitionsTask(Set partitions) implements TbQueueConsumerManagerTask { + @Override + public QueueTaskType getType() { + return QueueTaskType.UPDATE_PARTITIONS; + } + } + + record AddPartitionsTask(Set partitions, Consumer onStop) implements TbQueueConsumerManagerTask { + @Override + public QueueTaskType getType() { + return QueueTaskType.ADD_PARTITIONS; + } + } + + record RemovePartitionsTask(Set partitions) implements TbQueueConsumerManagerTask { + @Override + public QueueTaskType getType() { + return QueueTaskType.REMOVE_PARTITIONS; + } + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/TbQueueConsumerTask.java similarity index 94% rename from application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java rename to common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/TbQueueConsumerTask.java index 0b4e7c02d7..c3edc48285 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/TbQueueConsumerTask.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.service.queue.ruleengine; +package org.thingsboard.server.queue.common.consumer; import lombok.Getter; import lombok.Setter; @@ -35,14 +35,17 @@ public class TbQueueConsumerTask { private final Object key; private volatile TbQueueConsumer consumer; private volatile Supplier> consumerSupplier; + @Getter + private final Runnable callback; @Setter private Future task; - public TbQueueConsumerTask(Object key, Supplier> consumerSupplier) { + public TbQueueConsumerTask(Object key, Supplier> consumerSupplier, Runnable callback) { this.key = key; this.consumer = null; this.consumerSupplier = consumerSupplier; + this.callback = callback; } public TbQueueConsumer getConsumer() {