From 5dc6991907b9043f52214222a012cb724c0d284d Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Wed, 19 Feb 2025 15:44:05 +0200 Subject: [PATCH] Kafka states restore improvements --- .../KafkaCalculatedFieldStateService.java | 2 +- ...faultTbCalculatedFieldConsumerService.java | 2 +- .../queue/DefaultTbCoreConsumerService.java | 4 +- .../queue/DefaultTbEdgeConsumerService.java | 4 +- .../TbQueueConsumerManagerTask.java | 48 ----------- .../TbRuleEngineQueueConsumerManager.java | 11 ++- .../server/common/data/queue/QueueConfig.java | 12 +++ .../AbstractTbQueueConsumerTemplate.java | 3 + .../consumer/MainQueueConsumerManager.java | 65 ++++++++++----- .../PartitionedQueueConsumerManager.java | 80 +++++++++++++++++++ .../common/consumer/QueueStateService.java | 76 ++++++++++++++++++ .../queue/common/consumer/QueueTaskType.java | 7 +- .../consumer/TbQueueConsumerManagerTask.java | 63 +++++++++++++++ .../common/consumer}/TbQueueConsumerTask.java | 7 +- 14 files changed, 298 insertions(+), 86 deletions(-) delete mode 100644 application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerManagerTask.java rename {application/src/main/java/org/thingsboard/server/service/queue => common/queue/src/main/java/org/thingsboard/server/queue/common}/consumer/MainQueueConsumerManager.java (83%) create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/PartitionedQueueConsumerManager.java create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueStateService.java rename application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java => common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueTaskType.java (77%) create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/TbQueueConsumerManagerTask.java rename {application/src/main/java/org/thingsboard/server/service/queue/ruleengine => common/queue/src/main/java/org/thingsboard/server/queue/common/consumer}/TbQueueConsumerTask.java (94%) 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..c05517acba 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 @@ -42,7 +42,7 @@ 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 org.thingsboard.server.queue.common.consumer.MainQueueConsumerManager; import java.util.Set; import java.util.concurrent.ExecutorService; 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..5e741c5064 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 @@ -55,7 +55,7 @@ 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.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.security.auth.jwt.settings.JwtSettingsService; 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..bf04491050 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 @@ -85,20 +86,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 +128,10 @@ public class MainQueueConsumerManager partitions) { + private void doUpdate(Set partitions) { this.partitions = partitions; consumerWrapper.updatePartitions(partitions); } @@ -197,6 +202,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 +287,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 +333,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..1d51563bee --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/PartitionedQueueConsumerManager.java @@ -0,0 +1,80 @@ +/** + * 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 + public void update(Set partitions) { + throw new UnsupportedOperationException("Use manual addPartitions and removePartitions"); + } + + @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..3c444b8495 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueStateService.java @@ -0,0 +1,76 @@ +/** + * 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) { + 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(removedPartitions.stream().map(tpi -> tpi.newByTopic(eventConsumer.getTopic())).collect(Collectors.toSet())); + } + + 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(); + } + }); + } + } + +} 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() {