tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Refactoring of the consumer service

pull/9453/head
Andrii Shvaika 3 years ago
committed by ViacheslavKlimov
parent
45fd970898
commit
3a0afd8a5a
10 changed files with 659 additions and 655 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 426
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java
  2. 37
      application/src/main/java/org/thingsboard/server/service/queue/TbQueueConsumerLauncher.java
  3. 17
      application/src/main/java/org/thingsboard/server/service/queue/TbQueueConsumerManagerTask.java
  4. 205
      application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineQueueConsumerManager.java
  5. 9
      application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java
  6. 32
      application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerManagerTask.java
  7. 70
      application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java
  8. 75
      application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineConsumerContext.java
  9. 431
      application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java
  10. 12
      common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java

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

@ -15,44 +15,25 @@
*/
package org.thingsboard.server.service.queue;
import com.google.protobuf.ProtocolStringList;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityIdFactory;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.data.rpc.RpcError;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.gen.MsgProtos;
import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
import org.thingsboard.server.common.msg.queue.RuleEngineException;
import org.thingsboard.server.common.msg.queue.RuleNodeInfo;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.common.msg.queue.TbMsgCallback;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse;
import org.thingsboard.server.common.stats.StatsFactory;
import org.thingsboard.server.dao.queue.QueueService;
import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg;
import org.thingsboard.server.queue.TbQueueAdmin;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.queue.discovery.QueueKey;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory;
import org.thingsboard.server.queue.util.DataDecodingEncodingService;
import org.thingsboard.server.queue.util.TbRuleEngineComponent;
@ -60,102 +41,48 @@ import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
import org.thingsboard.server.service.profile.TbAssetProfileCache;
import org.thingsboard.server.service.profile.TbDeviceProfileCache;
import org.thingsboard.server.service.queue.processing.AbstractConsumerService;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingDecision;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingResult;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategy;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategyFactory;
import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy;
import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategyFactory;
import org.thingsboard.server.service.queue.ruleengine.TbRuleEngineConsumerContext;
import org.thingsboard.server.service.queue.ruleengine.TbRuleEngineQueueConsumerManager;
import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService;
import org.thingsboard.server.service.stats.RuleEngineStatisticsService;
import org.threadly.concurrent.wrapper.KeyDistributedExecutor;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
@Service
@TbRuleEngineComponent
@Slf4j
public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<ToRuleEngineNotificationMsg> implements TbRuleEngineConsumerService {
public static final String SUCCESSFUL_STATUS = "successful";
public static final String FAILED_STATUS = "failed";
public static final String THREAD_TOPIC_SEPARATOR = " | ";
@Value("${queue.rule-engine.poll-interval}")
private long pollDuration;
@Value("${queue.rule-engine.pack-processing-timeout}")
private long packProcessingTimeout;
@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 int topicDeletionDelayInSec;
private final StatsFactory statsFactory;
private final TbRuleEngineSubmitStrategyFactory submitStrategyFactory;
private final TbRuleEngineProcessingStrategyFactory processingStrategyFactory;
private final TbRuleEngineQueueFactory tbRuleEngineQueueFactory;
private final RuleEngineStatisticsService statisticsService;
private final TbRuleEngineConsumerContext ctx;
private final TbRuleEngineDeviceRpcService tbDeviceRpcService;
private final TbServiceInfoProvider serviceInfoProvider;
private final QueueService queueService;
private final TbQueueProducerProvider producerProvider;
private final TbQueueAdmin queueAdmin;
private final ConcurrentMap<QueueKey, TbRuleEngineQueueConsumerManager> consumerMap = new ConcurrentHashMap<>();
private final ExecutorService consumersExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer"));
private final ExecutorService submitExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-submit"));
private final ScheduledExecutorService repartitionExecutor = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-repartition"));
private final ConcurrentMap<QueueKey, TbRuleEngineQueueConsumerManager> consumers = new ConcurrentHashMap<>();
public DefaultTbRuleEngineConsumerService(TbRuleEngineProcessingStrategyFactory processingStrategyFactory,
TbRuleEngineSubmitStrategyFactory submitStrategyFactory,
public DefaultTbRuleEngineConsumerService(TbRuleEngineConsumerContext ctx,
TbRuleEngineQueueFactory tbRuleEngineQueueFactory,
RuleEngineStatisticsService statisticsService,
ActorSystemContext actorContext,
DataDecodingEncodingService encodingService,
TbRuleEngineDeviceRpcService tbDeviceRpcService,
StatsFactory statsFactory,
TbDeviceProfileCache deviceProfileCache,
TbAssetProfileCache assetProfileCache,
TbTenantProfileCache tenantProfileCache,
TbApiUsageStateService apiUsageStateService,
PartitionService partitionService, ApplicationEventPublisher eventPublisher,
TbServiceInfoProvider serviceInfoProvider, QueueService queueService,
TbQueueProducerProvider producerProvider, TbQueueAdmin queueAdmin) {
super(actorContext, encodingService, tenantProfileCache, deviceProfileCache, assetProfileCache, apiUsageStateService, partitionService, eventPublisher, tbRuleEngineQueueFactory.createToRuleEngineNotificationsMsgConsumer(), Optional.empty());
this.statisticsService = statisticsService;
this.tbRuleEngineQueueFactory = tbRuleEngineQueueFactory;
this.submitStrategyFactory = submitStrategyFactory;
this.processingStrategyFactory = processingStrategyFactory;
PartitionService partitionService, ApplicationEventPublisher eventPublisher) {
super(actorContext, encodingService, tenantProfileCache, deviceProfileCache, assetProfileCache, apiUsageStateService, partitionService,
eventPublisher, tbRuleEngineQueueFactory.createToRuleEngineNotificationsMsgConsumer(), Optional.empty());
this.ctx = ctx;
this.tbDeviceRpcService = tbDeviceRpcService;
this.statsFactory = statsFactory;
this.serviceInfoProvider = serviceInfoProvider;
this.queueService = queueService;
this.producerProvider = producerProvider;
this.queueAdmin = queueAdmin;
}
@PostConstruct
public void init() {
super.init("tb-rule-engine-notifications-consumer"); // TODO: restore init of the main consumer?
List<Queue> queues = queueService.findAllQueues();
List<Queue> queues = ctx.findAllQueues();
for (Queue configuration : queues) {
if (partitionService.isManagedByCurrentService(configuration.getTenantId())) {
initConsumer(configuration);
@ -164,22 +91,14 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
}
private void initConsumer(Queue configuration) {
consumerMap.computeIfAbsent(new QueueKey(ServiceType.TB_RULE_ENGINE, configuration),
key -> new TbRuleEngineQueueConsumerManager(repartitionExecutor, consumersExecutor, statsFactory, tbRuleEngineQueueFactory, key)).init(configuration);
}
@PreDestroy
public void stop() {
super.destroy();
consumersExecutor.shutdownNow(); // TODO: shutdown or shutdownNow?
submitExecutor.shutdownNow();
repartitionExecutor.shutdownNow();
consumers.computeIfAbsent(new QueueKey(ServiceType.TB_RULE_ENGINE, configuration),
key -> new TbRuleEngineQueueConsumerManager(ctx, key)).init(configuration);
}
@Override
protected void onTbApplicationEvent(PartitionChangeEvent event) {
if (event.getServiceType().equals(getServiceType())) {
var consumer = consumerMap.get(event.getQueueKey());
var consumer = consumers.get(event.getQueueKey());
if (consumer != null) {
consumer.subscribe(event);
} else {
@ -188,207 +107,14 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
}
}
// void subscribeConsumerPerPartition(QueueKey queue, Set<TopicPartitionInfo> partitions) {
// topicsConsumerPerPartition.get(queue).getSubscribeQueue().add(partitions);
// scheduleTopicRepartition(queue);
// }
//
// private void scheduleTopicRepartition(QueueKey queue) {
// repartitionExecutor.schedule(() -> repartitionTopicWithConsumerPerPartition(queue), 1, TimeUnit.SECONDS);
// }
//
// void repartitionTopicWithConsumerPerPartition(final QueueKey queueKey) {
// if (stopped) {
// return;
// }
// TbTopicWithConsumerPerPartition tbTopicWithConsumerPerPartition = topicsConsumerPerPartition.get(queueKey);
// java.util.Queue<Set<TopicPartitionInfo>> subscribeQueue = tbTopicWithConsumerPerPartition.getSubscribeQueue();
// if (subscribeQueue.isEmpty()) {
// return;
// }
// if (tbTopicWithConsumerPerPartition.getLock().tryLock()) {
// try {
// Set<TopicPartitionInfo> partitions = null;
// while (!subscribeQueue.isEmpty()) {
// partitions = subscribeQueue.poll();
// }
// if (partitions == null) {
// return;
// }
//
// Set<TopicPartitionInfo> addedPartitions = new HashSet<>(partitions);
// ConcurrentMap<TopicPartitionInfo, TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>>> consumers = tbTopicWithConsumerPerPartition.getConsumers();
// addedPartitions.removeAll(consumers.keySet());
// log.info("calculated addedPartitions {}", addedPartitions);
//
// Set<TopicPartitionInfo> removedPartitions = new HashSet<>(consumers.keySet());
// removedPartitions.removeAll(partitions);
// log.info("calculated removedPartitions {}", removedPartitions);
//
// removedPartitions.forEach((tpi) -> {
// removeConsumerForTopicByTpi(queueKey.getQueueName(), consumers, tpi);
// });
//
// addedPartitions.forEach((tpi) -> {
// log.info("[{}] Adding consumer for topic: {}", queueKey, tpi);
// Queue configuration = consumerConfigurations.get(queueKey);
// TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer = tbRuleEngineQueueFactory.createToRuleEngineMsgConsumer(configuration);
// consumers.put(tpi, consumer);
// launchConsumer(consumer, queueKey, tpi.getFullTopicName(), queueKey + "-" + tpi.getPartition().orElse(-999999));
// consumer.subscribe(Collections.singleton(tpi));
// });
// } finally {
// tbTopicWithConsumerPerPartition.getLock().unlock();
// }
// } else {
// scheduleTopicRepartition(queueKey); //reschedule later
// }
// }
void removeConsumerForTopicByTpi(String queue, ConcurrentMap<TopicPartitionInfo, TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>>> consumers, TopicPartitionInfo tpi) {
log.info("[{}] Removing consumer for topic: {}", queue, tpi);
consumers.remove(tpi).stop();
}
@Override
protected void launchMainConsumers() {
consumers.forEach((queue, consumer) -> launchConsumer(consumer, queue, queue, queue.getQueueName()));
consumers.values().forEach(TbRuleEngineQueueConsumerManager::launchMainConsumer);
}
@Override
protected void stopMainConsumers() {
consumerMap.values().forEach(TbRuleEngineQueueConsumerManager::stop);
}
void launchConsumer(TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer, QueueKey queueKey, Object consumerKey, String threadSuffix) {
if (isReady) {
log.info("[{}] Launching consumer", consumerKey);
consumersExecutor.execute(consumerKey, () -> consumerLoop(consumer, queueKey, threadSuffix));
} else {
scheduleLaunchConsumer(consumer, queueKey, consumerKey, threadSuffix);
}
}
private void scheduleLaunchConsumer(TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer, QueueKey queueKey, Object consumerKey, String threadSuffix) {
repartitionExecutor.schedule(() -> {
launchConsumer(consumer, queueKey, consumerKey, threadSuffix);
}, 10, TimeUnit.SECONDS);
}
void consumerLoop(TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer, QueueKey queueKey, String threadSuffix) {
Queue configuration = consumerConfigurations.get(queueKey);
TbRuleEngineConsumerStats stats = consumerStats.get(queueKey);
updateCurrentThreadName(threadSuffix);
while (!stopped && !consumer.isStopped() && !consumer.isQueueDeleted()) {
try {
List<TbProtoQueueMsg<ToRuleEngineMsg>> msgs = consumer.poll(configuration.getPollInterval());
if (msgs.isEmpty()) {
continue;
}
final TbRuleEngineSubmitStrategy submitStrategy = getSubmitStrategy(configuration);
final TbRuleEngineProcessingStrategy ackStrategy = getAckStrategy(configuration);
submitStrategy.init(msgs);
while (!stopped && !consumer.isStopped()) {
TbMsgPackProcessingContext ctx = new TbMsgPackProcessingContext(configuration.getName(), submitStrategy, ackStrategy.isSkipTimeoutMsgs());
submitStrategy.submitAttempt((id, msg) -> submitExecutor.submit(() -> submitMessage(configuration, stats, ctx, id, msg)));
final boolean timeout = !ctx.await(configuration.getPackProcessingTimeout(), TimeUnit.MILLISECONDS);
TbRuleEngineProcessingResult result = new TbRuleEngineProcessingResult(configuration.getName(), timeout, ctx);
if (timeout) {
printFirstOrAll(configuration, ctx, ctx.getPendingMap(), "Timeout");
}
if (!ctx.getFailedMap().isEmpty()) {
printFirstOrAll(configuration, ctx, ctx.getFailedMap(), "Failed");
}
ctx.printProfilerStats();
TbRuleEngineProcessingDecision decision = ackStrategy.analyze(result);
if (statsEnabled) {
stats.log(result, decision.isCommit());
}
ctx.cleanup();
if (decision.isCommit()) {
submitStrategy.stop();
break;
} else {
submitStrategy.update(decision.getReprocessMap());
}
}
consumer.commit();
} catch (Exception e) {
if (!stopped) {
log.warn("Failed to process messages from queue.", e);
try {
Thread.sleep(pollDuration);
} catch (InterruptedException e2) {
log.trace("Failed to wait until the server has capacity to handle new requests", e2);
}
}
}
}
if (consumer.isStopped()) {
consumer.unsubscribe();
} else if (consumer.isQueueDeleted()) {
processQueueDeletion(configuration, consumer);
}
log.info("TB Rule Engine Consumer stopped.");
}
void updateCurrentThreadName(String threadSuffix) {
String name = Thread.currentThread().getName();
int spliteratorIndex = name.indexOf(THREAD_TOPIC_SEPARATOR);
if (spliteratorIndex > 0) {
name = name.substring(0, spliteratorIndex);
}
name = name + THREAD_TOPIC_SEPARATOR + threadSuffix;
Thread.currentThread().setName(name);
}
TbRuleEngineProcessingStrategy getAckStrategy(Queue configuration) {
return processingStrategyFactory.newInstance(configuration.getName(), configuration.getProcessingStrategy());
}
TbRuleEngineSubmitStrategy getSubmitStrategy(Queue configuration) {
return submitStrategyFactory.newInstance(configuration.getName(), configuration.getSubmitStrategy());
}
void submitMessage(Queue configuration, TbRuleEngineConsumerStats stats, TbMsgPackProcessingContext ctx, UUID id, TbProtoQueueMsg<ToRuleEngineMsg> msg) {
log.trace("[{}] Creating callback for topic {} message: {}", id, configuration.getName(), msg.getValue());
ToRuleEngineMsg toRuleEngineMsg = msg.getValue();
TenantId tenantId = TenantId.fromUUID(new UUID(toRuleEngineMsg.getTenantIdMSB(), toRuleEngineMsg.getTenantIdLSB()));
TbMsgCallback callback = prometheusStatsEnabled ?
new TbMsgPackCallback(id, tenantId, ctx, stats.getTimer(tenantId, SUCCESSFUL_STATUS), stats.getTimer(tenantId, FAILED_STATUS)) :
new TbMsgPackCallback(id, tenantId, ctx);
try {
if (toRuleEngineMsg.getTbMsg() != null && !toRuleEngineMsg.getTbMsg().isEmpty()) {
forwardToRuleEngineActor(configuration.getName(), tenantId, toRuleEngineMsg, callback);
} else {
callback.onSuccess();
}
} catch (Exception e) {
callback.onFailure(new RuleEngineException(e.getMessage(), e));
}
}
private void printFirstOrAll(Queue configuration, TbMsgPackProcessingContext ctx, Map<UUID, TbProtoQueueMsg<ToRuleEngineMsg>> map, String prefix) {
boolean printAll = log.isTraceEnabled();
log.info("{} to process [{}] messages", prefix, map.size());
for (Map.Entry<UUID, TbProtoQueueMsg<ToRuleEngineMsg>> pending : map.entrySet()) {
ToRuleEngineMsg tmp = pending.getValue().getValue();
TbMsg tmpMsg = TbMsg.fromBytes(configuration.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);
} else {
log.info("[{}] {} to process message: {}, Last Rule Node: {}", TenantId.fromUUID(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo);
break;
}
}
consumers.values().forEach(TbRuleEngineQueueConsumerManager::stop);
}
@Override
@ -398,18 +124,18 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
@Override
protected long getNotificationPollDuration() {
return pollDuration;
return ctx.getPollDuration();
}
@Override
protected long getNotificationPackProcessingTimeout() {
return packProcessingTimeout;
return ctx.getPackProcessingTimeout();
}
@Override
protected void handleNotification(UUID id, TbProtoQueueMsg<ToRuleEngineNotificationMsg> msg, TbCallback callback) throws Exception {
ToRuleEngineNotificationMsg nfMsg = msg.getValue();
if (nfMsg.getComponentLifecycleMsg() != null && !nfMsg.getComponentLifecycleMsg().isEmpty()) {
if (!nfMsg.getComponentLifecycleMsg().isEmpty()) {
handleComponentLifecycleMsg(id, nfMsg.getComponentLifecycleMsg());
callback.onSuccess();
} else if (nfMsg.hasFromDeviceRpcResponse()) {
@ -420,10 +146,10 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
tbDeviceRpcService.processRpcResponseFromDevice(response);
callback.onSuccess();
} else if (nfMsg.hasQueueUpdateMsg()) {
repartitionExecutor.execute(() -> updateQueue(nfMsg.getQueueUpdateMsg()));
ctx.getScheduler().execute(() -> updateQueue(nfMsg.getQueueUpdateMsg()));
callback.onSuccess();
} else if (nfMsg.hasQueueDeleteMsg()) {
repartitionExecutor.execute(() -> deleteQueue(nfMsg.getQueueDeleteMsg()));
ctx.getScheduler().execute(() -> deleteQueue(nfMsg.getQueueDeleteMsg()));
callback.onSuccess();
} else {
log.trace("Received notification with missing handler");
@ -438,33 +164,13 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
QueueId queueId = new QueueId(new UUID(queueUpdateMsg.getQueueIdMSB(), queueUpdateMsg.getQueueIdLSB()));
String queueName = queueUpdateMsg.getQueueName();
QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queueName, tenantId);
Queue queue = queueService.findQueueById(tenantId, queueId);
Queue oldQueue = consumerConfigurations.remove(queueKey);
if (oldQueue != null) {
if (oldQueue.isConsumerPerPartition()) {
TbTopicWithConsumerPerPartition consumerPerPartition = topicsConsumerPerPartition.remove(queueKey);
ReentrantLock lock = consumerPerPartition.getLock();
try {
lock.lock();
consumerPerPartition.getConsumers().values().forEach(TbQueueConsumer::stop);
} finally {
lock.unlock();
}
} else {
TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer = consumers.remove(queueKey);
consumer.stop();
}
}
initConsumer(queue);
if (!queue.isConsumerPerPartition()) {
launchConsumer(consumers.get(queueKey), queueKey, queueKey, queueName);
}
Queue queue = ctx.getQueueService().findQueueById(tenantId, queueId);
consumers.computeIfAbsent(queueKey, key -> new TbRuleEngineQueueConsumerManager(ctx, key)).update(queue);
}
partitionService.updateQueue(queueUpdateMsg);
partitionService.recalculatePartitions(serviceInfoProvider.getServiceInfo(), new ArrayList<>(partitionService.getOtherServices(ServiceType.TB_RULE_ENGINE)));
partitionService.recalculatePartitions(ctx.getServiceInfoProvider().getServiceInfo(),
new ArrayList<>(partitionService.getOtherServices(ServiceType.TB_RULE_ENGINE)));
}
private void deleteQueue(TransportProtos.QueueDeleteMsg queueDeleteMsg) {
@ -473,90 +179,18 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queueDeleteMsg.getQueueName(), tenantId);
partitionService.removeQueue(queueDeleteMsg);
Queue queue = consumerConfigurations.remove(queueKey);
if (queue != null) {
if (queue.isConsumerPerPartition()) {
TbTopicWithConsumerPerPartition tbTopicWithConsumerPerPartition = topicsConsumerPerPartition.remove(queueKey);
if (tbTopicWithConsumerPerPartition != null) {
tbTopicWithConsumerPerPartition.getConsumers().values().forEach(TbQueueConsumer::onQueueDelete);
tbTopicWithConsumerPerPartition.getConsumers().clear();
}
} else {
TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer = consumers.remove(queueKey);
if (consumer != null) {
consumer.onQueueDelete();
}
}
}
partitionService.recalculatePartitions(serviceInfoProvider.getServiceInfo(), new ArrayList<>(partitionService.getOtherServices(ServiceType.TB_RULE_ENGINE)));
}
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();
Set<String> relationTypes = null;
if (relationTypesList != null) {
if (relationTypesList.size() == 1) {
relationTypes = Collections.singleton(relationTypesList.get(0));
} else {
relationTypes = new HashSet<>(relationTypesList);
}
}
msg = new QueueToRuleEngineMsg(tenantId, tbMsg, relationTypes, toRuleEngineMsg.getFailureMessage());
actorContext.tell(msg);
}
private void processQueueDeletion(Queue queue, TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer) {
long finishTs = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(topicDeletionDelayInSec);
try {
int n = 0;
while (System.currentTimeMillis() <= finishTs) {
List<TbProtoQueueMsg<ToRuleEngineMsg>> msgs = consumer.poll(queue.getPollInterval());
if (msgs.isEmpty()) {
continue;
}
for (TbProtoQueueMsg<ToRuleEngineMsg> 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);
var manager = consumers.remove(queueKey);
if (manager != null) {
manager.delete();
}
partitionService.recalculatePartitions(ctx.getServiceInfoProvider().getServiceInfo(), new ArrayList<>(partitionService.getOtherServices(ServiceType.TB_RULE_ENGINE)));
}
@Scheduled(fixedDelayString = "${queue.rule-engine.stats.print-interval-ms}")
public void printStats() {
if (statsEnabled) {
if (ctx.isStatsEnabled()) {
long ts = System.currentTimeMillis();
consumerStats.forEach((queue, stats) -> {
stats.printStats();
statisticsService.reportQueueStats(ts, stats);
stats.reset();
});
consumers.values().forEach(manager -> manager.printStats(ts));
}
}

37
application/src/main/java/org/thingsboard/server/service/queue/TbQueueConsumerLauncher.java
View File

@ -1,37 +0,0 @@
package org.thingsboard.server.service.queue;
import lombok.Data;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.data.queue.Queue;
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
public class TbQueueConsumerLauncher {
private final TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> consumer;
private volatile Future<?> task;
public void stop() {
this.consumer.stop();
}
public void awaitStopped() throws ExecutionException, InterruptedException, TimeoutException {
if (task != null) {
this.task.get(3, TimeUnit.MINUTES);
}
}
public void subscribe(Set<TopicPartitionInfo> partitions) {
this.consumer.subscribe(partitions);
}
}

17
application/src/main/java/org/thingsboard/server/service/queue/TbQueueConsumerManagerTask.java
View File

@ -1,17 +0,0 @@
package org.thingsboard.server.service.queue;
import lombok.Data;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import java.util.Set;
@Data
public class TbQueueConsumerManagerTask {
private final ComponentLifecycleEvent event;
private final Queue queue;
private final Set<TopicPartitionInfo> partitions;
}

205
application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineQueueConsumerManager.java
View File

@ -1,205 +0,0 @@
package org.thingsboard.server.service.queue;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.stats.StatsFactory;
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 org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
@Data
@Slf4j
public class TbRuleEngineQueueConsumerManager {
private final ScheduledExecutorService scheduler;
private final ExecutorService consumerExecutor;
private final StatsFactory statsFactory;
private final TbRuleEngineQueueFactory queueFactory;
private final QueueKey key;
private final ReentrantLock lock = new ReentrantLock(); //NonfairSync
private final ConcurrentMap<TopicPartitionInfo, TbQueueConsumerLauncher> consumers = new ConcurrentHashMap<>();
private final TbRuleEngineConsumerStats stats;
private volatile Set<TopicPartitionInfo> partitions = Collections.emptySet();
private volatile Queue queue;
private volatile TbQueueConsumerLauncher mainConsumer;
private final java.util.Queue<TbQueueConsumerManagerTask> tasks = new ConcurrentLinkedQueue<>();
public TbRuleEngineQueueConsumerManager(ScheduledExecutorService scheduler, ExecutorService consumerExecutor, StatsFactory statsFactory, TbRuleEngineQueueFactory queueFactory, QueueKey key) {
this.scheduler = scheduler;
this.consumerExecutor = consumerExecutor;
this.statsFactory = statsFactory;
this.queueFactory = queueFactory;
this.key = key;
this.stats = new TbRuleEngineConsumerStats(key, statsFactory);
}
public void init(Queue queue) {
processTask(new TbQueueConsumerManagerTask(ComponentLifecycleEvent.CREATED, queue, null));
}
private void processTask(TbQueueConsumerManagerTask todo) {
tasks.add(todo);
log.info("[{}] Adding task: {}", key, todo);
tryProcessTasks();
}
private void tryProcessTasks() {
consumerExecutor.submit(() -> {
if (lock.tryLock()) {
try {
TbQueueConsumerManagerTask lastUpdateTask = null;
while (!tasks.isEmpty()) {
TbQueueConsumerManagerTask task = tasks.poll();
switch (task.getEvent()) {
case CREATED:
doInit(task.getQueue());
break;
case UPDATED:
lastUpdateTask = task;
break;
case DELETED:
lastUpdateTask = null;
doDelete();
break;
}
}
if (lastUpdateTask != null) {
doUpdate(lastUpdateTask.getQueue(), lastUpdateTask.getPartitions());
}
} finally {
lock.unlock();
}
} else {
log.debug("[{}] Failed to acquire lock.", key);
scheduler.schedule(this::tryProcessTasks, 1, TimeUnit.SECONDS);
}
});
}
public void doInit(Queue queue) {
log.info("[{}] Init consumer with queue: {}", key, queue);
this.queue = queue;
if (!queue.isConsumerPerPartition()) {
mainConsumer = new TbQueueConsumerLauncher(queueFactory.createToRuleEngineMsgConsumer(queue));
}
}
private void doUpdate(Queue newQueue, Set<TopicPartitionInfo> partitions) {
if (newQueue.isConsumerPerPartition()) {
} else {
for (var oldConsumer : consumers.values()) {
oldConsumer.stop();
}
for (var oldConsumer : consumers.entrySet()) {
try {
oldConsumer.getValue().awaitStopped();
} catch (Exception e) {
log.info("[{}][{}] Failed to stop the consumer during update", key, oldConsumer.getKey().getPartition().orElse(-1), e);
}
}
if (mainConsumer == null) {
mainConsumer = new TbQueueConsumerLauncher(queueFactory.createToRuleEngineMsgConsumer(queue));
//TODO: launch
}
mainConsumer.subscribe(partitions);
}
}
private void doDelete() {
}
public void subscribe(PartitionChangeEvent event) {
log.info("[{}] Subscribing to partitions: {}", key, event.getPartitions());
if (!queue.isConsumerPerPartition()) {
mainConsumer.subscribe(event.getPartitions());
} else {
log.info("[{}] Subscribing consumer per partition: {}", key, event.getPartitions());
subscribeConsumerPerPartition(event.getQueueKey(), event.getPartitions());
}
}
void subscribeConsumerPerPartition(QueueKey queue, Set<TopicPartitionInfo> partitions) {
topicsConsumerPerPartition.get(queue).getSubscribeQueue().add(partitions);
scheduleTopicRepartition(queue);
}
private void scheduleTopicRepartition(QueueKey queue) {
repartitionExecutor.schedule(() -> repartitionTopicWithConsumerPerPartition(queue), 1, TimeUnit.SECONDS);
}
void repartitionTopicWithConsumerPerPartition(final QueueKey queueKey) {
if (stopped) {
return;
}
TbTopicWithConsumerPerPartition tbTopicWithConsumerPerPartition = topicsConsumerPerPartition.get(queueKey);
java.util.Queue<Set<TopicPartitionInfo>> subscribeQueue = tbTopicWithConsumerPerPartition.getSubscribeQueue();
if (subscribeQueue.isEmpty()) {
return;
}
if (tbTopicWithConsumerPerPartition.getLock().tryLock()) {
try {
Set<TopicPartitionInfo> partitions = null;
while (!subscribeQueue.isEmpty()) {
partitions = subscribeQueue.poll();
}
if (partitions == null) {
return;
}
Set<TopicPartitionInfo> addedPartitions = new HashSet<>(partitions);
ConcurrentMap<TopicPartitionInfo, TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>>> consumers = tbTopicWithConsumerPerPartition.getConsumers();
addedPartitions.removeAll(consumers.keySet());
log.info("calculated addedPartitions {}", addedPartitions);
Set<TopicPartitionInfo> removedPartitions = new HashSet<>(consumers.keySet());
removedPartitions.removeAll(partitions);
log.info("calculated removedPartitions {}", removedPartitions);
removedPartitions.forEach((tpi) -> {
removeConsumerForTopicByTpi(queueKey.getQueueName(), consumers, tpi);
});
addedPartitions.forEach((tpi) -> {
log.info("[{}] Adding consumer for topic: {}", queueKey, tpi);
Queue configuration = consumerConfigurations.get(queueKey);
TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> consumer = tbRuleEngineQueueFactory.createToRuleEngineMsgConsumer(configuration);
consumers.put(tpi, consumer);
launchConsumer(consumer, queueKey, tpi.getFullTopicName(), queueKey + "-" + tpi.getPartition().orElse(-999999));
consumer.subscribe(Collections.singleton(tpi));
});
} finally {
tbTopicWithConsumerPerPartition.getLock().unlock();
}
} else {
scheduleTopicRepartition(queueKey); //reschedule later
}
}
public void stop() {
// consumers.values().forEach(TbQueueConsumer::stop);
// topicsConsumerPerPartition.values().forEach(tbTopicWithConsumerPerPartition -> tbTopicWithConsumerPerPartition.getConsumers().keySet()
// .forEach((tpi) -> removeConsumerForTopicByTpi(tbTopicWithConsumerPerPartition.getTopic(), tbTopicWithConsumerPerPartition.getConsumers(), tpi)));
}
}

9
application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java
View File

@ -0,0 +1,9 @@
package org.thingsboard.server.service.queue.ruleengine;
import java.io.Serializable;
public enum QueueEvent implements Serializable {
CREATED, LAUNCHED, UPDATED, PARTITION_CHANGE, STOP, DELETED
}

32
application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerManagerTask.java
View File

@ -0,0 +1,32 @@
/**
* 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;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import java.util.Set;
@Data
public class TbQueueConsumerManagerTask {
private final QueueEvent event;
private final Queue queue;
private final Set<TopicPartitionInfo> partitions;
}

70
application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java
View File

@ -0,0 +1,70 @@
/**
* 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;
import lombok.extern.slf4j.Slf4j;
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 TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> consumer;
private volatile Future<?> task;
public void stop() {
this.consumer.stop();
}
public boolean stopAndAwait() {
this.consumer.stop();
return await();
}
public boolean await() {
if (task != null) {
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;
}
}
return true;
}
public void subscribe(Set<TopicPartitionInfo> partitions) {
this.consumer.subscribe(partitions);
}
public void unsubscribe() {
this.consumer.unsubscribe();
}
}

75
application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineConsumerContext.java
View File

@ -0,0 +1,75 @@
package org.thingsboard.server.service.queue.ruleengine;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.stats.StatsFactory;
import org.thingsboard.server.dao.queue.QueueService;
import org.thingsboard.server.queue.TbQueueAdmin;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory;
import org.thingsboard.server.queue.util.TbRuleEngineComponent;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategyFactory;
import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategyFactory;
import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService;
import org.thingsboard.server.service.stats.RuleEngineStatisticsService;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@Component
@TbRuleEngineComponent
@Slf4j
@Data
public class TbRuleEngineConsumerContext {
@Value("${queue.rule-engine.poll-interval}")
private long pollDuration;
@Value("${queue.rule-engine.pack-processing-timeout}")
private long packProcessingTimeout;
@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 int topicDeletionDelayInSec;
protected volatile boolean stopped = false;
protected volatile boolean isReady = false;
private final ActorSystemContext actorContext;
private final StatsFactory statsFactory;
private final TbRuleEngineSubmitStrategyFactory submitStrategyFactory;
private final TbRuleEngineProcessingStrategyFactory processingStrategyFactory;
private final TbRuleEngineQueueFactory queueFactory;
private final RuleEngineStatisticsService statisticsService;
private final TbServiceInfoProvider serviceInfoProvider;
private final QueueService queueService;
private final TbQueueProducerProvider producerProvider;
private final TbQueueAdmin queueAdmin;
//TODO: add reasonable limit for mgmt pool.
private final ExecutorService mgmtExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("tb-rule-engine-mgmt"));
private final ExecutorService consumersExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer"));
private final ExecutorService submitExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-submit"));
private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-scheduler"));
public List<Queue> findAllQueues() {
return queueService.findAllQueues();
}
@PreDestroy
public void stop() {
consumersExecutor.shutdownNow(); // TODO: shutdown or shutdownNow?
submitExecutor.shutdownNow();
scheduler.shutdownNow();
}
}

431
application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java
View File

@ -0,0 +1,431 @@
/**
* 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 com.google.protobuf.ProtocolStringList;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityIdFactory;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.gen.MsgProtos;
import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
import org.thingsboard.server.common.msg.queue.RuleEngineException;
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.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;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingDecision;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingResult;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategy;
import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy;
import java.util.Collections;
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.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
@Data
@Slf4j
public class TbRuleEngineQueueConsumerManager {
public static final String SUCCESSFUL_STATUS = "successful";
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<TopicPartitionInfo, TbQueueConsumerTask> consumersPerPartition = new ConcurrentHashMap<>();
private final TbRuleEngineConsumerStats stats;
private volatile Set<TopicPartitionInfo> partitions = Collections.emptySet();
private volatile Queue queue;
private volatile TbQueueConsumerTask mainConsumer;
private final java.util.Queue<TbQueueConsumerManagerTask> tasks = new ConcurrentLinkedQueue<>();
public TbRuleEngineQueueConsumerManager(TbRuleEngineConsumerContext ctx, QueueKey key) {
this.ctx = ctx;
this.key = key;
this.stats = new TbRuleEngineConsumerStats(key, ctx.getStatsFactory());
}
public void init(Queue queue) {
processTask(new TbQueueConsumerManagerTask(QueueEvent.CREATED, queue, null));
}
public void update(Queue queue) {
processTask(new TbQueueConsumerManagerTask(QueueEvent.UPDATED, queue, null));
}
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 stop() {
processTask(new TbQueueConsumerManagerTask(QueueEvent.STOP, null, null));
}
public void delete() {
processTask(new TbQueueConsumerManagerTask(QueueEvent.DELETED, null, null));
}
private void processTask(TbQueueConsumerManagerTask todo) {
tasks.add(todo);
log.info("[{}] Adding task: {}", key, todo);
tryProcessTasks();
}
private void tryProcessTasks() {
ctx.getMgmtExecutor().submit(() -> {
if (lock.tryLock()) {
try {
Queue newConfiguration = null;
Set<TopicPartitionInfo> newPartitions = null;
while (!tasks.isEmpty()) {
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 (newConfiguration != null) {
doUpdate(newConfiguration);
}
if (newPartitions != null) {
doUpdate(newPartitions);
}
} finally {
lock.unlock();
}
} else {
log.debug("[{}] Failed to acquire lock.", key);
ctx.getScheduler().schedule(this::tryProcessTasks, 1, TimeUnit.SECONDS);
}
});
}
public void doInit(Queue queue) {
log.info("[{}] Init consumer with queue: {}", key, queue);
this.queue = queue;
if (queue.isConsumerPerPartition()) {
log.debug("[{}] Ignore init event since isConsumerPerPartition is enabled.", key);
} 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);
}
}
private void doUpdate(Queue newQueue) {
log.info("[{}] Processing queue update: {}", key, newQueue);
var oldQueue = queue;
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();
}
}
private void doUpdate(Set<TopicPartitionInfo> partitions) {
log.info("[{}] Subscribing to partitions: {}", key, partitions);
if (queue.isConsumerPerPartition()) {
log.debug("[{}] Subscribing consumers per partition separately: {}", key, partitions);
Set<TopicPartitionInfo> addedPartitions = new HashSet<>(partitions);
addedPartitions.removeAll(consumersPerPartition.keySet());
log.info("calculated addedPartitions {}", addedPartitions);
Set<TopicPartitionInfo> 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();
});
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));
});
} else {
mainConsumer.subscribe(partitions);
}
}
private void doStop() {
doStop(queue);
}
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 doDelete() {
doStop();
//TODO: repack messages
}
@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)));
}
void consumerLoop(TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> consumer, Queue configuration, String threadSuffix) {
ThingsBoardThreadFactory.updateCurrentThreadName(threadSuffix);
while (!ctx.stopped && !consumer.isStopped()
//TODO: remove this.
&& !consumer.isQueueDeleted()) {
try {
List<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> 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();
} catch (Exception e) {
if (!ctx.stopped) {
log.warn("Failed to process messages from queue.", e);
try {
Thread.sleep(ctx.getPollDuration());
} catch (InterruptedException e2) {
log.trace("Failed to wait until the server has capacity to handle new requests", e2);
}
}
}
}
//TODO: refactor and move to the "doDelete" method. Use separate consumer if needed (it is still synchronous).
if (consumer.isQueueDeleted()) {
processQueueDeletion(configuration, consumer);
}
log.info("TB Rule Engine Consumer stopped.");
}
private void processQueueDeletion(Queue queue, TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> consumer) {
// long finishTs = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(topicDeletionDelayInSec);
// try {
// int n = 0;
// while (System.currentTimeMillis() <= finishTs) {
// List<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> msgs = consumer.poll(queue.getPollInterval());
// if (msgs.isEmpty()) {
// continue;
// }
// for (TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg> 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);
// }
}
TbRuleEngineSubmitStrategy getSubmitStrategy(Queue configuration) {
return ctx.getSubmitStrategyFactory().newInstance(configuration.getName(), configuration.getSubmitStrategy());
}
TbRuleEngineProcessingStrategy getProcessingStrategy(Queue configuration) {
return ctx.getProcessingStrategyFactory().newInstance(configuration.getName(), configuration.getProcessingStrategy());
}
void submitMessage(Queue configuration, TbRuleEngineConsumerStats stats, TbMsgPackProcessingContext packCtx, UUID id, TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg> msg) {
log.trace("[{}] Creating callback for topic {} message: {}", id, configuration.getName(), msg.getValue());
TransportProtos.ToRuleEngineMsg toRuleEngineMsg = msg.getValue();
TenantId tenantId = TenantId.fromUUID(new UUID(toRuleEngineMsg.getTenantIdMSB(), toRuleEngineMsg.getTenantIdLSB()));
TbMsgCallback callback = ctx.prometheusStatsEnabled ?
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);
} else {
callback.onSuccess();
}
} catch (Exception e) {
callback.onFailure(new RuleEngineException(e.getMessage(), e));
}
}
private void forwardToRuleEngineActor(String queueName, TenantId tenantId, TransportProtos.ToRuleEngineMsg toRuleEngineMsg, TbMsgCallback callback) {
TbMsg tbMsg = TbMsg.fromBytes(queueName, toRuleEngineMsg.getTbMsg().toByteArray(), callback);
QueueToRuleEngineMsg msg;
ProtocolStringList relationTypesList = toRuleEngineMsg.getRelationTypesList();
Set<String> relationTypes;
if (relationTypesList.size() == 1) {
relationTypes = Collections.singleton(relationTypesList.get(0));
} else {
relationTypes = new HashSet<>(relationTypesList);
}
msg = new QueueToRuleEngineMsg(tenantId, tbMsg, relationTypes, toRuleEngineMsg.getFailureMessage());
ctx.getActorContext().tell(msg);
}
private void printFirstOrAll(Queue configuration, TbMsgPackProcessingContext ctx, Map<UUID, TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> map, String prefix) {
boolean printAll = log.isTraceEnabled();
log.info("{} to process [{}] messages", prefix, map.size());
for (Map.Entry<UUID, TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> pending : map.entrySet()) {
TransportProtos.ToRuleEngineMsg tmp = pending.getValue().getValue();
TbMsg tmpMsg = TbMsg.fromBytes(configuration.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);
} else {
log.info("[{}] {} to process message: {}, Last Rule Node: {}", TenantId.fromUUID(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo);
break;
}
}
}
public void printStats(long ts) {
stats.printStats();
ctx.getStatisticsService().reportQueueStats(ts, stats);
stats.reset();
}
}

12
common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java
View File

@ -22,6 +22,7 @@ import java.util.concurrent.atomic.AtomicInteger;
* Copy of Executors.DefaultThreadFactory but with ability to set name of the pool
*/
public class ThingsBoardThreadFactory implements ThreadFactory {
public static final String THREAD_TOPIC_SEPARATOR = " | ";
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
@ -40,6 +41,17 @@ public class ThingsBoardThreadFactory implements ThreadFactory {
"-thread-";
}
public static void updateCurrentThreadName(String threadSuffix) {
String name = Thread.currentThread().getName();
int spliteratorIndex = name.indexOf(THREAD_TOPIC_SEPARATOR);
if (spliteratorIndex > 0) {
name = name.substring(0, spliteratorIndex);
}
name = name + THREAD_TOPIC_SEPARATOR + threadSuffix;
Thread.currentThread().setName(name);
}
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,

Write Preview
Loading…
Cancel
Save