diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index 164b48c25f..64139530b1 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -121,9 +121,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< } @Override - protected void launchMainConsumers() { - consumers.values().forEach(TbRuleEngineQueueConsumerManager::launchMainConsumer); - } + protected void launchMainConsumers() {} @Override protected void stopConsumers() { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java index 4d01d15d9a..a4a3bcc59a 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java @@ -1,9 +1,24 @@ +/** + * Copyright © 2016-2023 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 java.io.Serializable; public enum QueueEvent implements Serializable { - CREATED, LAUNCHED, UPDATED, PARTITION_CHANGE, STOP, DELETED + PARTITION_CHANGE, CONFIG_UPDATE, STOP, DELETE } 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 index 5b4f164a10..6f1adbdcf2 100644 --- 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 @@ -15,18 +15,33 @@ */ package org.thingsboard.server.service.queue.ruleengine; -import lombok.Data; -import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; +import lombok.Getter; +import lombok.ToString; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import java.util.Set; -@Data +@Getter +@ToString public class TbQueueConsumerManagerTask { private final QueueEvent event; - private final Queue queue; - private final Set partitions; + private Queue queue; + private Set partitions; + + public TbQueueConsumerManagerTask(QueueEvent event) { + this.event = event; + } + + public TbQueueConsumerManagerTask(QueueEvent event, Queue queue) { + this.event = event; + this.queue = queue; + } + + public TbQueueConsumerManagerTask(QueueEvent event, Set partitions) { + this.event = event; + this.partitions = partitions; + } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java index ed1f36074f..a0a5579951 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -21,51 +21,44 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.QueueKey; import java.util.Set; -import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; @Data @Slf4j public class TbQueueConsumerTask { - private final QueueKey key; - private final Object id; + private final Object key; private final TbQueueConsumer> consumer; private volatile Future task; - public void stop() { - this.consumer.stop(); + public void subscribe(Set partitions) { + log.trace("[{}] Subscribing to partitions: {}", key, partitions); + consumer.subscribe(partitions); } - public boolean stopAndAwait() { - this.consumer.stop(); - return await(); + public void initiateStop() { + log.debug("[{}] Initiating stop", key); + consumer.stop(); } - public boolean await() { - if (task != null) { - //TODO: maybe task.cancel() to interrupt the consumer? + public void awaitFinish() { + log.trace("[{}] Awaiting finish", key); + if (isRunning()) { try { - this.task.get(3, TimeUnit.MINUTES); - } catch (ExecutionException | InterruptedException | TimeoutException e) { - log.warn("[{}][{}] Failed to await for consumer to stop", key, id, e); - return false; + task.get(60, TimeUnit.SECONDS); + task = null; + } catch (Exception e) { + log.warn("[{}] Failed to await for consumer to stop", key, e); } } - return true; + log.trace("[{}] Awaited finish", key); } - public void subscribe(Set partitions) { - this.consumer.subscribe(partitions); + public boolean isRunning() { + return task != null && !task.isDone(); } - - public void unsubscribe() { - this.consumer.unsubscribe(); - } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineConsumerContext.java b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineConsumerContext.java index c1d17746e1..325a685972 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineConsumerContext.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineConsumerContext.java @@ -1,3 +1,18 @@ +/** + * Copyright © 2016-2023 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.Data; @@ -36,8 +51,8 @@ public class TbRuleEngineConsumerContext { @Value("${queue.rule-engine.stats.enabled:true}") private boolean statsEnabled; @Value("${queue.rule-engine.prometheus-stats.enabled:false}") - boolean prometheusStatsEnabled; - @Value("${queue.rule-engine.topic-deletion-delay:30}") + private boolean prometheusStatsEnabled; + @Value("${queue.rule-engine.topic-deletion-delay:15}") private int topicDeletionDelayInSec; @Value("${queue.rule-engine.management-thread-pool-size:12}") private int mgmtThreadPoolSize; 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 2599a214f0..dc3966d0d2 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 @@ -5,7 +5,7 @@ * 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 + * 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, @@ -16,8 +16,7 @@ package org.thingsboard.server.service.queue.ruleengine; import com.google.protobuf.ProtocolStringList; -import lombok.Data; -import lombok.SneakyThrows; +import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.data.id.EntityId; @@ -32,11 +31,10 @@ import org.thingsboard.server.common.msg.queue.RuleNodeInfo; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbMsgCallback; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; -import org.thingsboard.server.gen.transport.TransportProtos; +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.discovery.QueueKey; -import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; import org.thingsboard.server.service.queue.TbMsgPackCallback; import org.thingsboard.server.service.queue.TbMsgPackProcessingContext; import org.thingsboard.server.service.queue.TbRuleEngineConsumerStats; @@ -45,19 +43,20 @@ import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingRes import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategy; import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy; +import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.ReentrantLock; +import java.util.stream.Collectors; -@Data @Slf4j public class TbRuleEngineQueueConsumerManager { @@ -65,87 +64,83 @@ public class TbRuleEngineQueueConsumerManager { public static final String FAILED_STATUS = "failed"; private final TbRuleEngineConsumerContext ctx; - private final QueueKey key; - - private final ReentrantLock lock = new ReentrantLock(); //NonfairSync - private final ConcurrentMap consumersPerPartition = new ConcurrentHashMap<>(); + private final QueueKey queueKey; private final TbRuleEngineConsumerStats stats; + private final ReentrantLock lock = new ReentrantLock(); //NonfairSync - private volatile Set partitions = Collections.emptySet(); + @Getter private volatile Queue queue; - private volatile TbQueueConsumerTask mainConsumer; + @Getter + private volatile Set partitions; + private volatile ConsumerWrapper consumerWrapper; + + private volatile boolean stopped; private final java.util.Queue tasks = new ConcurrentLinkedQueue<>(); - public TbRuleEngineQueueConsumerManager(TbRuleEngineConsumerContext ctx, QueueKey key) { + public TbRuleEngineQueueConsumerManager(TbRuleEngineConsumerContext ctx, QueueKey queueKey) { this.ctx = ctx; - this.key = key; - this.stats = new TbRuleEngineConsumerStats(key, ctx.getStatsFactory()); + this.queueKey = queueKey; + this.stats = new TbRuleEngineConsumerStats(queueKey, ctx.getStatsFactory()); } public void init(Queue queue) { - processTask(new TbQueueConsumerManagerTask(QueueEvent.CREATED, queue, null)); + doInit(queue); } public void update(Queue queue) { - processTask(new TbQueueConsumerManagerTask(QueueEvent.UPDATED, queue, null)); + addTask(new TbQueueConsumerManagerTask(QueueEvent.CONFIG_UPDATE, queue)); } - public void subscribe(PartitionChangeEvent event) { - processTask(new TbQueueConsumerManagerTask(QueueEvent.PARTITION_CHANGE, queue, event.getPartitions())); - } - - public void launchMainConsumer() { - processTask(new TbQueueConsumerManagerTask(QueueEvent.LAUNCHED, null, null)); + public void update(Set partitions) { + addTask(new TbQueueConsumerManagerTask(QueueEvent.PARTITION_CHANGE, partitions)); } public void stop() { - processTask(new TbQueueConsumerManagerTask(QueueEvent.STOP, null, null)); + addTask(new TbQueueConsumerManagerTask(QueueEvent.STOP)); } public void delete() { - processTask(new TbQueueConsumerManagerTask(QueueEvent.DELETED, null, null)); + addTask(new TbQueueConsumerManagerTask(QueueEvent.DELETE)); } - private void processTask(TbQueueConsumerManagerTask todo) { + private void addTask(TbQueueConsumerManagerTask todo) { + if (stopped) { + return; + } tasks.add(todo); - log.info("[{}] Adding task: {}", key, todo); + log.trace("[{}] Added task: {}", queueKey, todo); tryProcessTasks(); } private void tryProcessTasks() { + if (!ctx.isReady()) { + log.debug("[{}] TbRuleEngineConsumerContext is not ready yet, will process tasks later", queueKey); + ctx.getScheduler().schedule(this::tryProcessTasks, 1, TimeUnit.SECONDS); + return; + } ctx.getMgmtExecutor().submit(() -> { if (lock.tryLock()) { try { Queue newConfiguration = null; Set newPartitions = null; - while (!tasks.isEmpty()) { + while (!stopped) { TbQueueConsumerManagerTask task = tasks.poll(); - switch (task.getEvent()) { - case CREATED: - doInit(task.getQueue()); - break; - case LAUNCHED: - if (!queue.isConsumerPerPartition()) { - doLaunchMainConsumer(); - } - break; - case UPDATED: - newConfiguration = task.getQueue(); - break; - case PARTITION_CHANGE: - newPartitions = task.getPartitions(); - break; - case STOP: - newConfiguration = null; - newPartitions = null; - doStop(); - break; - case DELETED: - newConfiguration = null; - newPartitions = null; - doDelete(); - break; + if (task == null) { + break; + } + log.trace("[{}] Processing task: {}", queueKey, task); + + if (task.getEvent() == QueueEvent.PARTITION_CHANGE) { + newPartitions = task.getPartitions(); + } else if (task.getEvent() == QueueEvent.CONFIG_UPDATE) { + newConfiguration = task.getQueue(); + } else if (task.getEvent() == QueueEvent.STOP) { + doStop(); + return; + } else if (task.getEvent() == QueueEvent.DELETE) { + doDelete(); + return; } } if (newConfiguration != null) { @@ -158,158 +153,108 @@ public class TbRuleEngineQueueConsumerManager { lock.unlock(); } } else { - log.debug("[{}] Failed to acquire lock.", key); + log.trace("[{}] Failed to acquire lock", queueKey); ctx.getScheduler().schedule(this::tryProcessTasks, 1, TimeUnit.SECONDS); } }); } - public void doInit(Queue queue) { - log.info("[{}] Init consumer with queue: {}", key, queue); + private void doInit(Queue queue) { this.queue = queue; if (queue.isConsumerPerPartition()) { - log.debug("[{}] Ignore init event since isConsumerPerPartition is enabled.", key); + consumerWrapper = new ConsumerPerPartitionWrapper(); } else { - mainConsumer = new TbQueueConsumerTask(key, "main", ctx.getQueueFactory().createToRuleEngineMsgConsumer(queue)); - } - } - - private void doLaunchMainConsumer() { - if (mainConsumer != null) { - launchConsumer(mainConsumer, queue, mainConsumer.getId(), queue.getName()); - } else { - log.warn("[{}] Can't launch main consumer since it is empty!", key); + consumerWrapper = new SingleConsumerWrapper(); } + log.debug("[{}] Initialized consumer for queue: {}", queueKey, queue); } private void doUpdate(Queue newQueue) { - log.info("[{}] Processing queue update: {}", key, newQueue); - var oldQueue = queue; + log.info("[{}] Processing queue update: {}", queueKey, newQueue); + var oldQueue = this.queue; + this.queue = newQueue; if (log.isTraceEnabled()) { - log.trace("[{}] Old queue configuration: {}", key, oldQueue); - log.trace("[{}] New queue configuration: {}", key, newQueue); - } - if (oldQueue != null) { - doStop(oldQueue); - } - doInit(newQueue); - if (!newQueue.isConsumerPerPartition()) { - doLaunchMainConsumer(); + log.trace("[{}] Old queue configuration: {}", queueKey, oldQueue); + log.trace("[{}] New queue configuration: {}", queueKey, newQueue); } - } - - private void doUpdate(Set partitions) { - log.info("[{}] Subscribing to partitions: {}", key, partitions); - if (queue.isConsumerPerPartition()) { - log.debug("[{}] Subscribing consumers per partition separately: {}", key, partitions); - Set addedPartitions = new HashSet<>(partitions); - addedPartitions.removeAll(consumersPerPartition.keySet()); - log.info("calculated addedPartitions {}", addedPartitions); - - Set removedPartitions = new HashSet<>(consumersPerPartition.keySet()); - removedPartitions.removeAll(partitions); - log.info("calculated removedPartitions {}", removedPartitions); - removedPartitions.forEach((tpi) -> { - log.info("[{}] Unsubscribing from topic: {}", queue, tpi); - consumersPerPartition.get(tpi).unsubscribe(); - }); + if (oldQueue == null) { + doInit(queue); + } else if (newQueue.isConsumerPerPartition() != oldQueue.isConsumerPerPartition()) { + consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::initiateStop); + consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitFinish); - removedPartitions.forEach((tpi) -> { - log.info("[{}] Removing consumer for topic: {}", queue, tpi); - consumersPerPartition.get(tpi).stopAndAwait(); - consumersPerPartition.remove(tpi); - }); - - addedPartitions.forEach((tpi) -> { - log.info("[{}] Adding consumer for topic: {}", key, tpi); - TbQueueConsumerTask consumerTask = new TbQueueConsumerTask(key, tpi, ctx.getQueueFactory().createToRuleEngineMsgConsumer(queue)); - consumersPerPartition.put(tpi, consumerTask); - //TODO: Is it ok to subscribe first? - consumerTask.subscribe(Collections.singleton(tpi)); - launchConsumer(consumerTask, queue, mainConsumer.getId(), key + "-" + tpi.getPartition().orElse(-999999)); - }); + doInit(queue); + if (partitions != null) { + doUpdate(partitions); // even if partitions number was changed, there can be no partition change event + } } else { - mainConsumer.subscribe(partitions); + // do nothing, because partitions change (if they changed) will be handled on PartitionChangeEvent, + // and changes to pollInterval/packProcessingTimeout/submitStrategy/processingStrategy will be picked up by consumer on the fly, + // and queue topic and name are immutable } } - private void doStop() { - doStop(queue); + private void doUpdate(Set partitions) { + this.partitions = partitions; + consumerWrapper.updatePartitions(partitions); } - private void doStop(Queue queue) { - if (queue.isConsumerPerPartition()) { - consumersPerPartition.values().forEach(TbQueueConsumerTask::unsubscribe); - consumersPerPartition.values().forEach(TbQueueConsumerTask::stopAndAwait); - } else if (mainConsumer != null) { - mainConsumer.unsubscribe(); - mainConsumer.stopAndAwait(); - } + private void doStop() { + consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::initiateStop); + consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitFinish); + log.debug("[{}] Unsubscribed and stopped consumers", queueKey); + stopped = true; } private void doDelete() { - doStop(); - //TODO: repack messages + stopped = true; + log.info("[{}] Handling queue deletion", queueKey); + consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitFinish); + + List>> queueConsumers = consumerWrapper.getConsumers().stream() + .map(TbQueueConsumerTask::getConsumer).collect(Collectors.toList()); + ctx.getConsumersExecutor().submit(() -> { + drainQueue(queueConsumers); + + queueConsumers.forEach(consumer -> { + for (String topic : consumer.getFullTopicNames()) { + try { + ctx.getQueueAdmin().deleteTopic(topic); + log.info("Deleted topic {}", topic); + } catch (Exception e) { + log.error("Failed to delete topic {}", topic, e); + } + } + try { + consumer.unsubscribe(); + } catch (Exception e) { + log.error("[{}] Failed to unsubscribe consumer", queueKey, e); + } + }); + }); } - @SneakyThrows - void launchConsumer(TbQueueConsumerTask consumerTask, Queue configuration, Object consumerKey, String threadSuffix) { - log.info("[{}] Launching consumer: [{}]", key, consumerKey); - while (!ctx.isReady) { - //TODO: Remember this task. Cancel previous task if needed. - log.debug("[{}][{}] Waiting for consumer to get ready..", key, consumerKey); - Thread.sleep(1000); - } - consumerTask.setTask(ctx.getConsumersExecutor().submit(() -> consumerLoop(consumerTask.getConsumer(), configuration, threadSuffix))); + private void launchConsumer(TbQueueConsumerTask consumerTask) { + log.info("[{}] Launching consumer", consumerTask.getKey()); + Future consumerLoop = ctx.getConsumersExecutor().submit(() -> { + ThingsBoardThreadFactory.updateCurrentThreadName(consumerTask.getKey().toString()); + consumerLoop(consumerTask.getConsumer()); + }); + consumerTask.setTask(consumerLoop); } - void consumerLoop(TbQueueConsumer> consumer, Queue configuration, String threadSuffix) { - ThingsBoardThreadFactory.updateCurrentThreadName(threadSuffix); - while (!ctx.stopped && !consumer.isStopped() - //TODO: remove this. - && !consumer.isQueueDeleted()) { + private void consumerLoop(TbQueueConsumer> consumer) { + while (!stopped && !consumer.isStopped()) { try { - List> msgs = consumer.poll(queue.getPollInterval()); + List> msgs = consumer.poll(queue.getPollInterval()); if (msgs.isEmpty()) { continue; } - final TbRuleEngineSubmitStrategy submitStrategy = getSubmitStrategy(queue); - final TbRuleEngineProcessingStrategy ackStrategy = getProcessingStrategy(queue); - submitStrategy.init(msgs); - while (!ctx.isStopped() && !consumer.isStopped()) { - TbMsgPackProcessingContext packCtx = new TbMsgPackProcessingContext(queue.getName(), submitStrategy, ackStrategy.isSkipTimeoutMsgs()); - submitStrategy.submitAttempt((id, msg) -> ctx.getSubmitExecutor().submit(() -> submitMessage(configuration, stats, packCtx, id, msg))); - - final boolean timeout = !packCtx.await(configuration.getPackProcessingTimeout(), TimeUnit.MILLISECONDS); - - TbRuleEngineProcessingResult result = new TbRuleEngineProcessingResult(configuration.getName(), timeout, packCtx); - if (timeout) { - printFirstOrAll(configuration, packCtx, packCtx.getPendingMap(), "Timeout"); - } - if (!packCtx.getFailedMap().isEmpty()) { - printFirstOrAll(configuration, packCtx, packCtx.getFailedMap(), "Failed"); - } - packCtx.printProfilerStats(); - - TbRuleEngineProcessingDecision decision = ackStrategy.analyze(result); - if (ctx.isStatsEnabled()) { - stats.log(result, decision.isCommit()); - } - - packCtx.cleanup(); - - if (decision.isCommit()) { - submitStrategy.stop(); - break; - } else { - submitStrategy.update(decision.getReprocessMap()); - } - } - consumer.commit(); + processMsgs(msgs, consumer, queue); } catch (Exception e) { - if (!ctx.stopped) { - log.warn("Failed to process messages from queue.", e); + if (!stopped) { + log.warn("Failed to process messages from queue", e); try { Thread.sleep(ctx.getPollDuration()); } catch (InterruptedException e2) { @@ -318,72 +263,68 @@ public class TbRuleEngineQueueConsumerManager { } } } - //TODO: refactor and move to the "doDelete" method. Use separate consumer if needed (it is still synchronous). - if (consumer.isQueueDeleted()) { - processQueueDeletion(configuration, consumer); + if (consumer.isStopped()) { + consumer.unsubscribe(); } - log.info("TB Rule Engine Consumer stopped."); + log.info("Rule Engine consumer stopped"); } - private void processQueueDeletion(Queue queue, TbQueueConsumer> consumer) { -// long finishTs = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(topicDeletionDelayInSec); -// try { -// int n = 0; -// while (System.currentTimeMillis() <= finishTs) { -// List> msgs = consumer.poll(queue.getPollInterval()); -// if (msgs.isEmpty()) { -// continue; -// } -// for (TbProtoQueueMsg msg : msgs) { -// try { -// MsgProtos.TbMsgProto tbMsgProto = MsgProtos.TbMsgProto.parseFrom(msg.getValue().getTbMsg().toByteArray()); -// EntityId originator = EntityIdFactory.getByTypeAndUuid(tbMsgProto.getEntityType(), new UUID(tbMsgProto.getEntityIdMSB(), tbMsgProto.getEntityIdLSB())); -// -// TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, queue.getName(), TenantId.SYS_TENANT_ID, originator); -// producerProvider.getRuleEngineMsgProducer().send(tpi, msg, null); -// n++; -// } catch (Throwable e) { -// log.debug("Failed to move message to system {}: {}", consumer.getTopic(), msg, e); -// } -// } -// consumer.commit(); -// } -// if (n > 0) { -// log.info("Moved {} messages from {} to system {}", n, consumer.getFullTopicNames(), consumer.getTopic()); -// } -// -// consumer.unsubscribe(); -// for (String topic : consumer.getFullTopicNames()) { -// try { -// queueAdmin.deleteTopic(topic); -// log.info("Deleted topic {}", topic); -// } catch (Exception e) { -// log.error("Failed to delete topic {} after unsubscribing", topic, e); -// } -// } -// } catch (Exception e) { -// log.error("Failed to process deletion of {} ({})", consumer.getTopic(), queue.getTenantId(), e); -// } + private void processMsgs(List> msgs, + TbQueueConsumer> consumer, + Queue queue) throws InterruptedException { + TbRuleEngineSubmitStrategy submitStrategy = getSubmitStrategy(queue); + TbRuleEngineProcessingStrategy ackStrategy = getProcessingStrategy(queue); + submitStrategy.init(msgs); + while (!stopped && !consumer.isStopped()) { + TbMsgPackProcessingContext packCtx = new TbMsgPackProcessingContext(queue.getName(), submitStrategy, ackStrategy.isSkipTimeoutMsgs()); + submitStrategy.submitAttempt((id, msg) -> submitMessage(packCtx, id, msg)); + + final boolean timeout = !packCtx.await(queue.getPackProcessingTimeout(), TimeUnit.MILLISECONDS); + + TbRuleEngineProcessingResult result = new TbRuleEngineProcessingResult(queue.getName(), timeout, packCtx); + if (timeout) { + printFirstOrAll(packCtx, packCtx.getPendingMap(), "Timeout"); + } + if (!packCtx.getFailedMap().isEmpty()) { + printFirstOrAll(packCtx, packCtx.getFailedMap(), "Failed"); + } + packCtx.printProfilerStats(); + + TbRuleEngineProcessingDecision decision = ackStrategy.analyze(result); + if (ctx.isStatsEnabled()) { + stats.log(result, decision.isCommit()); + } + + packCtx.cleanup(); + + if (decision.isCommit()) { + submitStrategy.stop(); + consumer.commit(); + break; + } else { + submitStrategy.update(decision.getReprocessMap()); + } + } } - TbRuleEngineSubmitStrategy getSubmitStrategy(Queue configuration) { - return ctx.getSubmitStrategyFactory().newInstance(configuration.getName(), configuration.getSubmitStrategy()); + private TbRuleEngineSubmitStrategy getSubmitStrategy(Queue queue) { + return ctx.getSubmitStrategyFactory().newInstance(queue.getName(), queue.getSubmitStrategy()); } - TbRuleEngineProcessingStrategy getProcessingStrategy(Queue configuration) { - return ctx.getProcessingStrategyFactory().newInstance(configuration.getName(), configuration.getProcessingStrategy()); + private TbRuleEngineProcessingStrategy getProcessingStrategy(Queue queue) { + return ctx.getProcessingStrategyFactory().newInstance(queue.getName(), queue.getProcessingStrategy()); } - void submitMessage(Queue configuration, TbRuleEngineConsumerStats stats, TbMsgPackProcessingContext packCtx, UUID id, TbProtoQueueMsg msg) { - log.trace("[{}] Creating callback for topic {} message: {}", id, configuration.getName(), msg.getValue()); - TransportProtos.ToRuleEngineMsg toRuleEngineMsg = msg.getValue(); + private void submitMessage(TbMsgPackProcessingContext packCtx, UUID id, TbProtoQueueMsg msg) { + log.trace("[{}] Creating callback for topic {} message: {}", id, queue.getName(), msg.getValue()); + ToRuleEngineMsg toRuleEngineMsg = msg.getValue(); TenantId tenantId = TenantId.fromUUID(new UUID(toRuleEngineMsg.getTenantIdMSB(), toRuleEngineMsg.getTenantIdLSB())); - TbMsgCallback callback = ctx.prometheusStatsEnabled ? + TbMsgCallback callback = ctx.isPrometheusStatsEnabled() ? new TbMsgPackCallback(id, tenantId, packCtx, stats.getTimer(tenantId, SUCCESSFUL_STATUS), stats.getTimer(tenantId, FAILED_STATUS)) : new TbMsgPackCallback(id, tenantId, packCtx); try { if (!toRuleEngineMsg.getTbMsg().isEmpty()) { - forwardToRuleEngineActor(configuration.getName(), tenantId, toRuleEngineMsg, callback); + forwardToRuleEngineActor(queue.getName(), tenantId, toRuleEngineMsg, callback); } else { callback.onSuccess(); } @@ -392,7 +333,7 @@ public class TbRuleEngineQueueConsumerManager { } } - private void forwardToRuleEngineActor(String queueName, TenantId tenantId, TransportProtos.ToRuleEngineMsg toRuleEngineMsg, TbMsgCallback callback) { + private void forwardToRuleEngineActor(String queueName, TenantId tenantId, ToRuleEngineMsg toRuleEngineMsg, TbMsgCallback callback) { TbMsg tbMsg = TbMsg.fromBytes(queueName, toRuleEngineMsg.getTbMsg().toByteArray(), callback); QueueToRuleEngineMsg msg; ProtocolStringList relationTypesList = toRuleEngineMsg.getRelationTypesList(); @@ -406,16 +347,15 @@ public class TbRuleEngineQueueConsumerManager { ctx.getActorContext().tell(msg); } - - private void printFirstOrAll(Queue configuration, TbMsgPackProcessingContext ctx, Map> map, String prefix) { + private void printFirstOrAll(TbMsgPackProcessingContext ctx, Map> map, String prefix) { boolean printAll = log.isTraceEnabled(); - log.info("{} to process [{}] messages", prefix, map.size()); - for (Map.Entry> pending : map.entrySet()) { - TransportProtos.ToRuleEngineMsg tmp = pending.getValue().getValue(); - TbMsg tmpMsg = TbMsg.fromBytes(configuration.getName(), tmp.getTbMsg().toByteArray(), TbMsgCallback.EMPTY); + log.info("[{}] {} to process [{}] messages", queueKey, prefix, map.size()); + for (Map.Entry> pending : map.entrySet()) { + ToRuleEngineMsg tmp = pending.getValue().getValue(); + TbMsg tmpMsg = TbMsg.fromBytes(queue.getName(), tmp.getTbMsg().toByteArray(), TbMsgCallback.EMPTY); RuleNodeInfo ruleNodeInfo = ctx.getLastVisitedRuleNode(pending.getKey()); if (printAll) { - log.trace("[{}] {} to process message: {}, Last Rule Node: {}", TenantId.fromUUID(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo); + log.trace("[{}][{}] {} to process message: {}, Last Rule Node: {}", queueKey, TenantId.fromUUID(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo); } else { log.info("[{}] {} to process message: {}, Last Rule Node: {}", TenantId.fromUUID(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo); break; @@ -428,4 +368,117 @@ public class TbRuleEngineQueueConsumerManager { ctx.getStatisticsService().reportQueueStats(ts, stats); stats.reset(); } + + private void drainQueue(List>> consumers) { + long finishTs = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(ctx.getTopicDeletionDelayInSec()); + try { + int n = 0; + while (System.currentTimeMillis() <= finishTs) { + for (TbQueueConsumer> consumer : consumers) { + List> msgs = consumer.poll(queue.getPollInterval()); + if (msgs.isEmpty()) { + continue; + } + for (TbProtoQueueMsg msg : msgs) { + try { + MsgProtos.TbMsgProto tbMsgProto = MsgProtos.TbMsgProto.parseFrom(msg.getValue().getTbMsg().toByteArray()); + EntityId originator = EntityIdFactory.getByTypeAndUuid(tbMsgProto.getEntityType(), new UUID(tbMsgProto.getEntityIdMSB(), tbMsgProto.getEntityIdLSB())); + + TopicPartitionInfo tpi = ctx.getPartitionService().resolve(ServiceType.TB_RULE_ENGINE, queue.getName(), TenantId.SYS_TENANT_ID, originator); + ctx.getProducerProvider().getRuleEngineMsgProducer().send(tpi, msg, null); + n++; + } catch (Throwable e) { + log.warn("Failed to move message to system {}: {}", consumer.getTopic(), msg, e); + } + } + consumer.commit(); + } + } + if (n > 0) { + log.info("Moved {} messages from {} to system {}", n, queueKey, queue.getName()); + } + } catch (Exception e) { + log.error("[{}] Failed to drain queue", queueKey, e); + } + } + + private static String partitionsToString(Collection partitions) { + return partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.joining(", ", "[", "]")); + } + + interface ConsumerWrapper { + + void updatePartitions(Set partitions); + + Collection getConsumers(); + + } + + class ConsumerPerPartitionWrapper implements ConsumerWrapper { + private final Map consumers = new HashMap<>(); + + @Override + public void updatePartitions(Set partitions) { + Set addedPartitions = new HashSet<>(partitions); + addedPartitions.removeAll(consumers.keySet()); + + Set removedPartitions = new HashSet<>(consumers.keySet()); + removedPartitions.removeAll(partitions); + log.info("[{}] Added partitions: {}, removed partitions: {}", queueKey, partitionsToString(addedPartitions), partitionsToString(removedPartitions)); + + removedPartitions.forEach((tpi) -> { + consumers.get(tpi).initiateStop(); + }); + removedPartitions.forEach((tpi) -> { + consumers.remove(tpi).awaitFinish(); + }); + + addedPartitions.forEach((tpi) -> { + String key = queueKey + "-" + tpi.getPartition().orElse(-999999); + TbQueueConsumerTask consumer = new TbQueueConsumerTask(key, ctx.getQueueFactory().createToRuleEngineMsgConsumer(queue)); + consumers.put(tpi, consumer); + consumer.subscribe(Set.of(tpi)); + launchConsumer(consumer); + }); + } + + @Override + public Collection getConsumers() { + return consumers.values(); + } + } + + class SingleConsumerWrapper implements ConsumerWrapper { + private TbQueueConsumerTask consumer; + + @Override + public void updatePartitions(Set partitions) { + log.info("[{}] New partitions: {}", queueKey, partitionsToString(partitions)); + if (partitions.isEmpty()) { + if (consumer != null && consumer.isRunning()) { + consumer.initiateStop(); + consumer.awaitFinish(); + } + consumer = null; + return; + } + + if (consumer == null) { + consumer = new TbQueueConsumerTask(queueKey, ctx.getQueueFactory().createToRuleEngineMsgConsumer(queue)); + } + consumer.subscribe(partitions); + if (!consumer.isRunning()) { + launchConsumer(consumer); + } + } + + @Override + public Collection getConsumers() { + if (consumer == null) { + return Collections.emptyList(); + } + return List.of(consumer); + } + } + } diff --git a/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueConsumer.java b/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueConsumer.java index 9c41f9d342..21216e164e 100644 --- a/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueConsumer.java +++ b/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueConsumer.java @@ -28,6 +28,8 @@ public interface TbQueueConsumer { void subscribe(Set partitions); + void stop(); + void unsubscribe(); List poll(long durationInMillis); @@ -36,10 +38,6 @@ public interface TbQueueConsumer { boolean isStopped(); - void onQueueDelete(); - - boolean isQueueDeleted(); - List getFullTopicNames(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java index 2ebe41850d..073371ebb5 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java @@ -44,7 +44,6 @@ public abstract class AbstractTbQueueConsumerTemplate i protected volatile Set partitions; protected final ReentrantLock consumerLock = new ReentrantLock(); //NonfairSync final Queue> subscribeQueue = new ConcurrentLinkedQueue<>(); - protected volatile boolean queueDeleted = false; @Getter private final String topic; @@ -55,7 +54,7 @@ public abstract class AbstractTbQueueConsumerTemplate i @Override public void subscribe() { - log.info("enqueue topic subscribe {} ", topic); + log.debug("enqueue topic subscribe {} ", topic); if (stopped) { log.error("trying subscribe, but consumer stopped for topic {}", topic); return; @@ -65,7 +64,7 @@ public abstract class AbstractTbQueueConsumerTemplate i @Override public void subscribe(Set partitions) { - log.info("enqueue topics subscribe {} ", partitions); + log.debug("enqueue topics subscribe {} ", partitions); if (stopped) { log.error("trying subscribe, but consumer stopped for topic {}", topic); return; @@ -78,7 +77,8 @@ public abstract class AbstractTbQueueConsumerTemplate i List records; long startNanos = System.nanoTime(); if (stopped) { - return errorAndReturnEmpty(); + log.error("poll invoked but consumer stopped for topic " + topic, new RuntimeException("stacktrace")); + return emptyList(); } if (!subscribed && partitions == null && subscribeQueue.isEmpty()) { return sleepAndReturnEmpty(startNanos, durationInMillis); @@ -96,6 +96,7 @@ public abstract class AbstractTbQueueConsumerTemplate i } if (!subscribed) { List topicNames = getFullTopicNames(); + log.info("Subscribing to topics {}", topicNames); doSubscribe(topicNames); subscribed = true; } @@ -127,11 +128,6 @@ public abstract class AbstractTbQueueConsumerTemplate i return result; } - List errorAndReturnEmpty() { - log.error("poll invoked but consumer stopped for topic" + topic, new RuntimeException("stacktrace")); - return emptyList(); - } - List sleepAndReturnEmpty(final long startNanos, final long durationInMillis) { long durationNanos = TimeUnit.MILLISECONDS.toNanos(durationInMillis); long spentNanos = System.nanoTime() - startNanos; @@ -163,15 +159,20 @@ public abstract class AbstractTbQueueConsumerTemplate i } } + @Override + public void stop() { + stopped = true; + } + @Override public void unsubscribe() { - log.info("Unsubscribing from topics and stopping consumer for topics {}", partitions.stream() - .map(TopicPartitionInfo::getFullTopicName) - .collect(Collectors.joining(", "))); + log.info("Unsubscribing and stopping consumer for topics {}", getFullTopicNames()); stopped = true; consumerLock.lock(); try { - doUnsubscribe(); + if (subscribed) { + doUnsubscribe(); + } } finally { consumerLock.unlock(); } @@ -192,17 +193,11 @@ public abstract class AbstractTbQueueConsumerTemplate i abstract protected void doUnsubscribe(); - @Override - public void onQueueDelete() { - queueDeleted = true; - } - - public boolean isQueueDeleted() { - return queueDeleted; - } - @Override public List getFullTopicNames() { + if (partitions == null) { + return Collections.emptyList(); + } return partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toList()); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java index 9f58446966..ff3f0cc9b3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java @@ -73,7 +73,6 @@ public class TbKafkaConsumerTemplate extends AbstractTbQue protected void doSubscribe(List topicNames) { if (!topicNames.isEmpty()) { topicNames.forEach(admin::createTopicIfNotExists); - log.info("subscribe topics {}", topicNames); consumer.subscribe(topicNames); } else { log.info("unsubscribe due to empty topic list"); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueConsumer.java b/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueConsumer.java index 8711cbbcf1..a7f8cadd0d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueConsumer.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueConsumer.java @@ -31,7 +31,6 @@ public class InMemoryTbQueueConsumer implements TbQueueCon private volatile Set partitions; private volatile boolean stopped; private volatile boolean subscribed; - private volatile boolean queueDeleted; public InMemoryTbQueueConsumer(InMemoryStorage storage, String topic) { this.storage = storage; @@ -58,9 +57,15 @@ public class InMemoryTbQueueConsumer implements TbQueueCon subscribed = true; } + @Override + public void stop() { + stopped = true; + } + @Override public void unsubscribe() { stopped = true; + subscribed = false; } @Override @@ -104,16 +109,6 @@ public class InMemoryTbQueueConsumer implements TbQueueCon return stopped; } - @Override - public void onQueueDelete() { - queueDeleted = true; - } - - @Override - public boolean isQueueDeleted() { - return queueDeleted; - } - @Override public List getFullTopicNames() { return partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toList());