From 8636ed3b0d68af3160385c1ff302c4ebe281a824 Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Tue, 19 Mar 2024 12:57:37 +0200 Subject: [PATCH 1/6] TB Core consumer - parallel processing; refactoring --- .../queue/DefaultTbCoreConsumerService.java | 432 ++++++++---------- .../DefaultTbRuleEngineConsumerService.java | 39 +- .../consumer/BasicQueueConsumerManager.java | 109 +++++ .../queue/consumer/QueueConsumerManager.java | 321 +++++++++++++ .../processing/AbstractConsumerService.java | 128 +++--- .../TbQueueConsumerManagerTask.java | 8 +- .../queue/ruleengine/TbQueueConsumerTask.java | 7 +- .../TbRuleEngineConsumerContext.java | 25 - .../TbRuleEngineQueueConsumerManager.java | 290 ++---------- .../TbRuleEngineQueueConsumerManagerTest.java | 60 ++- .../server/common/data/queue/Queue.java | 2 +- .../server/common/data/queue/QueueConfig.java | 24 + .../provider/KafkaMonolithQueueFactory.java | 4 +- .../provider/KafkaTbCoreQueueFactory.java | 7 +- .../common/util/ThingsBoardThreadFactory.java | 5 + 15 files changed, 809 insertions(+), 652 deletions(-) create mode 100644 application/src/main/java/org/thingsboard/server/service/queue/consumer/BasicQueueConsumerManager.java create mode 100644 application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/queue/QueueConfig.java diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index 836c35aa85..c0ccaea9d0 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -18,12 +18,12 @@ package org.thingsboard.server.service.queue; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; +import lombok.Data; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.context.event.ApplicationReadyEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; @@ -40,6 +40,7 @@ import org.thingsboard.server.common.data.id.NotificationRequestId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTrigger; +import org.thingsboard.server.common.data.queue.QueueConfig; import org.thingsboard.server.common.data.rpc.RpcError; import org.thingsboard.server.common.msg.MsgType; import org.thingsboard.server.common.msg.TbActorMsg; @@ -75,9 +76,9 @@ import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceAct 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.event.PartitionChangeEvent; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; -import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.DataDecodingEncodingService; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; @@ -86,6 +87,8 @@ import org.thingsboard.server.service.notification.NotificationSchedulerService; import org.thingsboard.server.service.ota.OtaPackageStateService; import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; +import org.thingsboard.server.service.queue.consumer.BasicQueueConsumerManager; +import org.thingsboard.server.service.queue.consumer.QueueConsumerManager; import org.thingsboard.server.service.queue.processing.AbstractConsumerService; import org.thingsboard.server.service.queue.processing.IdMsgPair; import org.thingsboard.server.service.resource.TbImageService; @@ -108,7 +111,6 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -121,7 +123,7 @@ import java.util.stream.Collectors; public class DefaultTbCoreConsumerService extends AbstractConsumerService implements TbCoreConsumerService { @Value("${queue.core.poll-interval}") - private long pollDuration; + private long pollInterval; @Value("${queue.core.pack-processing-timeout}") private long packProcessingTimeout; @Value("${queue.core.stats.enabled:false}") @@ -132,7 +134,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService> mainConsumer; private final DeviceStateService stateService; private final TbApiUsageStateService statsService; private final TbLocalSubscriptionService localSubscriptionService; @@ -143,14 +144,14 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService> usageStatsConsumer; - private final TbQueueConsumer> firmwareStatesConsumer; + private final TbCoreQueueFactory queueFactory; private final TbImageService imageService; + private final TbCoreConsumerStats stats; + + private QueueConsumerManager, CoreQueueConfig> mainConsumer; + private BasicQueueConsumerManager> usageStatsConsumer; + private BasicQueueConsumerManager> firmwareStatesConsumer; - protected volatile ExecutorService consumersExecutor; - protected volatile ExecutorService usageStatsExecutor; - private volatile ExecutorService firmwareStatesExecutor; private volatile ListeningExecutorService deviceActivityEventsExecutor; public DefaultTbCoreConsumerService(TbCoreQueueFactory tbCoreQueueFactory, @@ -175,10 +176,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService, CoreQueueConfig>builder() + .queueKey(new QueueKey(ServiceType.TB_CORE)) + .config(CoreQueueConfig.of(true, (int) pollInterval)) + .msgPackProcessor(this::processMsgs) + .consumerCreator(config -> queueFactory.createToCoreMsgConsumer()) + .consumerExecutor(consumersExecutor) + .scheduler(scheduler) + .taskExecutor(mgmtExecutor) + .build(); + this.usageStatsConsumer = BasicQueueConsumerManager.>builder() + .key("usage-stats") + .name("TB Usage Stats") + .pollInterval(pollInterval) + .msgPackProcessor(this::processUsageStatsMsg) + .consumerCreator(queueFactory::createToUsageStatsServiceMsgConsumer) + .consumerExecutor(consumersExecutor) + .build(); + this.firmwareStatesConsumer = BasicQueueConsumerManager.>builder() + .key("firmware") + .name("TB Ota Package States") + .pollInterval(pollInterval) + .msgPackProcessor(this::processFirmwareMsgs) + .consumerCreator(queueFactory::createToOtaPackageStateServiceMsgConsumer) + .consumerExecutor(consumersExecutor) + .build(); } @PreDestroy public void destroy() { super.destroy(); - if (consumersExecutor != null) { - consumersExecutor.shutdownNow(); - } - if (usageStatsExecutor != null) { - usageStatsExecutor.shutdownNow(); - } - if (firmwareStatesExecutor != null) { - firmwareStatesExecutor.shutdownNow(); - } if (deviceActivityEventsExecutor != null) { deviceActivityEventsExecutor.shutdownNow(); } } - @AfterStartUp(order = AfterStartUp.REGULAR_SERVICE) - public void onApplicationEvent(ApplicationReadyEvent event) { - super.onApplicationEvent(event); - launchUsageStatsConsumer(); - launchOtaPackageUpdateNotificationConsumer(); + @Override + protected void startConsumers() { + super.startConsumers(); + firmwareStatesConsumer.subscribe(); + firmwareStatesConsumer.launch(); + usageStatsConsumer.launch(); } @Override protected void onTbApplicationEvent(PartitionChangeEvent event) { log.info("Subscribing to partitions: {}", event.getPartitions()); - this.mainConsumer.subscribe(event.getPartitions()); - this.usageStatsConsumer.subscribe( - event - .getPartitions() - .stream() - .map(tpi -> tpi.newByTopic(usageStatsConsumer.getTopic())) - .collect(Collectors.toSet())); - this.firmwareStatesConsumer.subscribe(); - } - - @Override - protected void launchMainConsumers() { - consumersExecutor.submit(() -> { - while (!stopped) { + mainConsumer.update(event.getPartitions()); + usageStatsConsumer.subscribe(event.getPartitions() + .stream() + .map(tpi -> tpi.newByTopic(usageStatsConsumer.getConsumer().getTopic())) + .collect(Collectors.toSet())); + } + + private void processMsgs(List> msgs, TbQueueConsumer> consumer, CoreQueueConfig config) throws Exception { + List> orderedMsgList = msgs.stream().map(msg -> new IdMsgPair<>(UUID.randomUUID(), msg)).collect(Collectors.toList()); + ConcurrentMap> pendingMap = orderedMsgList.stream().collect( + Collectors.toConcurrentMap(IdMsgPair::getUuid, IdMsgPair::getMsg)); + CountDownLatch processingTimeoutLatch = new CountDownLatch(1); + TbPackProcessingContext> ctx = new TbPackProcessingContext<>( + processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>()); + PendingMsgHolder pendingMsgHolder = new PendingMsgHolder(); + Future packSubmitFuture = consumersExecutor.submit(() -> { + orderedMsgList.forEach((element) -> { + UUID id = element.getUuid(); + TbProtoQueueMsg msg = element.getMsg(); + log.trace("[{}] Creating main callback for message: {}", id, msg.getValue()); + TbCallback callback = new TbPackCallback<>(id, ctx); try { - List> msgs = mainConsumer.poll(pollDuration); - if (msgs.isEmpty()) { - continue; - } - List> orderedMsgList = msgs.stream().map(msg -> new IdMsgPair<>(UUID.randomUUID(), msg)).collect(Collectors.toList()); - ConcurrentMap> pendingMap = orderedMsgList.stream().collect( - Collectors.toConcurrentMap(IdMsgPair::getUuid, IdMsgPair::getMsg)); - CountDownLatch processingTimeoutLatch = new CountDownLatch(1); - TbPackProcessingContext> ctx = new TbPackProcessingContext<>( - processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>()); - PendingMsgHolder pendingMsgHolder = new PendingMsgHolder(); - Future packSubmitFuture = consumersExecutor.submit(() -> { - orderedMsgList.forEach((element) -> { - UUID id = element.getUuid(); - TbProtoQueueMsg msg = element.getMsg(); - log.trace("[{}] Creating main callback for message: {}", id, msg.getValue()); - TbCallback callback = new TbPackCallback<>(id, ctx); - try { - ToCoreMsg toCoreMsg = msg.getValue(); - pendingMsgHolder.setToCoreMsg(toCoreMsg); - if (toCoreMsg.hasToSubscriptionMgrMsg()) { - log.trace("[{}] Forwarding message to subscription manager service {}", id, toCoreMsg.getToSubscriptionMgrMsg()); - forwardToSubMgrService(toCoreMsg.getToSubscriptionMgrMsg(), callback); - } else if (toCoreMsg.hasToDeviceActorMsg()) { - log.trace("[{}] Forwarding message to device actor {}", id, toCoreMsg.getToDeviceActorMsg()); - forwardToDeviceActor(toCoreMsg.getToDeviceActorMsg(), callback); - } else if (toCoreMsg.hasDeviceStateServiceMsg()) { - log.trace("[{}] Forwarding message to device state service {}", id, toCoreMsg.getDeviceStateServiceMsg()); - forwardToStateService(toCoreMsg.getDeviceStateServiceMsg(), callback); - } else if (toCoreMsg.hasEdgeNotificationMsg()) { - log.trace("[{}] Forwarding message to edge service {}", id, toCoreMsg.getEdgeNotificationMsg()); - forwardToEdgeNotificationService(toCoreMsg.getEdgeNotificationMsg(), callback); - } else if (toCoreMsg.hasDeviceConnectMsg()) { - log.trace("[{}] Forwarding message to device state service {}", id, toCoreMsg.getDeviceConnectMsg()); - forwardToStateService(toCoreMsg.getDeviceConnectMsg(), callback); - } else if (toCoreMsg.hasDeviceActivityMsg()) { - log.trace("[{}] Forwarding message to device state service {}", id, toCoreMsg.getDeviceActivityMsg()); - forwardToStateService(toCoreMsg.getDeviceActivityMsg(), callback); - } else if (toCoreMsg.hasDeviceDisconnectMsg()) { - log.trace("[{}] Forwarding message to device state service {}", id, toCoreMsg.getDeviceDisconnectMsg()); - forwardToStateService(toCoreMsg.getDeviceDisconnectMsg(), callback); - } else if (toCoreMsg.hasDeviceInactivityMsg()) { - log.trace("[{}] Forwarding message to device state service {}", id, toCoreMsg.getDeviceInactivityMsg()); - forwardToStateService(toCoreMsg.getDeviceInactivityMsg(), callback); - } else if (toCoreMsg.hasToDeviceActorNotification()) { - TbActorMsg actorMsg = ProtoUtils.fromProto(toCoreMsg.getToDeviceActorNotification()); - if (actorMsg != null) { - if (actorMsg.getMsgType().equals(MsgType.DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG)) { - tbCoreDeviceRpcService.forwardRpcRequestToDeviceActor((ToDeviceRpcRequestActorMsg) actorMsg); - } else { - log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg); - actorContext.tell(actorMsg); - } - } - callback.onSuccess(); - } else if (!toCoreMsg.getToDeviceActorNotificationMsg().isEmpty()) { - // will be removed in 3.6.1 in favour of hasToDeviceActorNotification() - Optional actorMsg = encodingService.decode(toCoreMsg.getToDeviceActorNotificationMsg().toByteArray()); - if (actorMsg.isPresent()) { - TbActorMsg tbActorMsg = actorMsg.get(); - if (tbActorMsg.getMsgType().equals(MsgType.DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG)) { - tbCoreDeviceRpcService.forwardRpcRequestToDeviceActor((ToDeviceRpcRequestActorMsg) tbActorMsg); - } else { - log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get()); - actorContext.tell(actorMsg.get()); - } - } - callback.onSuccess(); - } else if (toCoreMsg.hasNotificationSchedulerServiceMsg()) { - TransportProtos.NotificationSchedulerServiceMsg notificationSchedulerServiceMsg = toCoreMsg.getNotificationSchedulerServiceMsg(); - log.trace("[{}] Forwarding message to notification scheduler service {}", id, toCoreMsg.getNotificationSchedulerServiceMsg()); - forwardToNotificationSchedulerService(notificationSchedulerServiceMsg, callback); - } else if (toCoreMsg.hasErrorEventMsg()) { - forwardToEventService(toCoreMsg.getErrorEventMsg(), callback); - } else if (toCoreMsg.hasLifecycleEventMsg()) { - forwardToEventService(toCoreMsg.getLifecycleEventMsg(), callback); - } - } catch (Throwable e) { - log.warn("[{}] Failed to process message: {}", id, msg, e); - callback.onFailure(e); + ToCoreMsg toCoreMsg = msg.getValue(); + pendingMsgHolder.setToCoreMsg(toCoreMsg); + if (toCoreMsg.hasToSubscriptionMgrMsg()) { + log.trace("[{}] Forwarding message to subscription manager service {}", id, toCoreMsg.getToSubscriptionMgrMsg()); + forwardToSubMgrService(toCoreMsg.getToSubscriptionMgrMsg(), callback); + } else if (toCoreMsg.hasToDeviceActorMsg()) { + log.trace("[{}] Forwarding message to device actor {}", id, toCoreMsg.getToDeviceActorMsg()); + forwardToDeviceActor(toCoreMsg.getToDeviceActorMsg(), callback); + } else if (toCoreMsg.hasDeviceStateServiceMsg()) { + log.trace("[{}] Forwarding message to device state service {}", id, toCoreMsg.getDeviceStateServiceMsg()); + forwardToStateService(toCoreMsg.getDeviceStateServiceMsg(), callback); + } else if (toCoreMsg.hasEdgeNotificationMsg()) { + log.trace("[{}] Forwarding message to edge service {}", id, toCoreMsg.getEdgeNotificationMsg()); + forwardToEdgeNotificationService(toCoreMsg.getEdgeNotificationMsg(), callback); + } else if (toCoreMsg.hasDeviceConnectMsg()) { + log.trace("[{}] Forwarding message to device state service {}", id, toCoreMsg.getDeviceConnectMsg()); + forwardToStateService(toCoreMsg.getDeviceConnectMsg(), callback); + } else if (toCoreMsg.hasDeviceActivityMsg()) { + log.trace("[{}] Forwarding message to device state service {}", id, toCoreMsg.getDeviceActivityMsg()); + forwardToStateService(toCoreMsg.getDeviceActivityMsg(), callback); + } else if (toCoreMsg.hasDeviceDisconnectMsg()) { + log.trace("[{}] Forwarding message to device state service {}", id, toCoreMsg.getDeviceDisconnectMsg()); + forwardToStateService(toCoreMsg.getDeviceDisconnectMsg(), callback); + } else if (toCoreMsg.hasDeviceInactivityMsg()) { + log.trace("[{}] Forwarding message to device state service {}", id, toCoreMsg.getDeviceInactivityMsg()); + forwardToStateService(toCoreMsg.getDeviceInactivityMsg(), callback); + } else if (toCoreMsg.hasToDeviceActorNotification()) { + TbActorMsg actorMsg = ProtoUtils.fromProto(toCoreMsg.getToDeviceActorNotification()); + if (actorMsg != null) { + if (actorMsg.getMsgType().equals(MsgType.DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG)) { + tbCoreDeviceRpcService.forwardRpcRequestToDeviceActor((ToDeviceRpcRequestActorMsg) actorMsg); + } else { + log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg); + actorContext.tell(actorMsg); } - }); - }); - if (!processingTimeoutLatch.await(packProcessingTimeout, TimeUnit.MILLISECONDS)) { - if (!packSubmitFuture.isDone()) { - packSubmitFuture.cancel(true); - ToCoreMsg lastSubmitMsg = pendingMsgHolder.getToCoreMsg(); - log.info("Timeout to process message: {}", lastSubmitMsg); } - ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue())); - ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue())); - } - mainConsumer.commit(); - } catch (Exception e) { - if (!stopped) { - log.warn("Failed to obtain 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); + callback.onSuccess(); + } else if (!toCoreMsg.getToDeviceActorNotificationMsg().isEmpty()) { + // will be removed in 3.6.1 in favour of hasToDeviceActorNotification() + Optional actorMsg = encodingService.decode(toCoreMsg.getToDeviceActorNotificationMsg().toByteArray()); + if (actorMsg.isPresent()) { + TbActorMsg tbActorMsg = actorMsg.get(); + if (tbActorMsg.getMsgType().equals(MsgType.DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG)) { + tbCoreDeviceRpcService.forwardRpcRequestToDeviceActor((ToDeviceRpcRequestActorMsg) tbActorMsg); + } else { + log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get()); + actorContext.tell(actorMsg.get()); + } } + callback.onSuccess(); + } else if (toCoreMsg.hasNotificationSchedulerServiceMsg()) { + TransportProtos.NotificationSchedulerServiceMsg notificationSchedulerServiceMsg = toCoreMsg.getNotificationSchedulerServiceMsg(); + log.trace("[{}] Forwarding message to notification scheduler service {}", id, toCoreMsg.getNotificationSchedulerServiceMsg()); + forwardToNotificationSchedulerService(notificationSchedulerServiceMsg, callback); + } else if (toCoreMsg.hasErrorEventMsg()) { + forwardToEventService(toCoreMsg.getErrorEventMsg(), callback); + } else if (toCoreMsg.hasLifecycleEventMsg()) { + forwardToEventService(toCoreMsg.getLifecycleEventMsg(), callback); } + } catch (Throwable e) { + log.warn("[{}] Failed to process message: {}", id, msg, e); + callback.onFailure(e); } - } - log.info("TB Core Consumer stopped."); + }); }); + if (!processingTimeoutLatch.await(packProcessingTimeout, TimeUnit.MILLISECONDS)) { + if (!packSubmitFuture.isDone()) { + packSubmitFuture.cancel(true); + ToCoreMsg lastSubmitMsg = pendingMsgHolder.getToCoreMsg(); + log.info("Timeout to process message: {}", lastSubmitMsg); + } + ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue())); + ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue())); + } + consumer.commit(); } private static class PendingMsgHolder { @@ -365,7 +355,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService> createNotificationsConsumer() { + return queueFactory.createToCoreNotificationsMsgConsumer(); + } + @Override protected void handleNotification(UUID id, TbProtoQueueMsg msg, TbCallback callback) { ToCoreNotificationMsg toCoreNotification = msg.getValue(); @@ -430,92 +430,53 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService { - while (!stopped) { - try { - List> msgs = usageStatsConsumer.poll(getNotificationPollDuration()); - if (msgs.isEmpty()) { - continue; - } - ConcurrentMap> pendingMap = msgs.stream().collect( - Collectors.toConcurrentMap(s -> UUID.randomUUID(), Function.identity())); - CountDownLatch processingTimeoutLatch = new CountDownLatch(1); - TbPackProcessingContext> ctx = new TbPackProcessingContext<>( - processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>()); - pendingMap.forEach((id, msg) -> { - log.trace("[{}] Creating usage stats callback for message: {}", id, msg.getValue()); - TbCallback callback = new TbPackCallback<>(id, ctx); - try { - handleUsageStats(msg, callback); - } catch (Throwable e) { - log.warn("[{}] Failed to process usage stats: {}", id, msg, e); - callback.onFailure(e); - } - }); - if (!processingTimeoutLatch.await(getNotificationPackProcessingTimeout(), TimeUnit.MILLISECONDS)) { - ctx.getAckMap().forEach((id, msg) -> log.warn("[{}] Timeout to process usage stats: {}", id, msg.getValue())); - ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process usage stats: {}", id, msg.getValue())); - } - usageStatsConsumer.commit(); - } catch (Exception e) { - if (!stopped) { - log.warn("Failed to obtain usage stats from queue.", e); - try { - Thread.sleep(getNotificationPollDuration()); - } catch (InterruptedException e2) { - log.trace("Failed to wait until the server has capacity to handle new usage stats", e2); - } - } - } + private void processUsageStatsMsg(List> msgs, TbQueueConsumer> consumer) throws Exception { + ConcurrentMap> pendingMap = msgs.stream().collect( + Collectors.toConcurrentMap(s -> UUID.randomUUID(), Function.identity())); + CountDownLatch processingTimeoutLatch = new CountDownLatch(1); + TbPackProcessingContext> ctx = new TbPackProcessingContext<>( + processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>()); + pendingMap.forEach((id, msg) -> { + log.trace("[{}] Creating usage stats callback for message: {}", id, msg.getValue()); + TbCallback callback = new TbPackCallback<>(id, ctx); + try { + handleUsageStats(msg, callback); + } catch (Throwable e) { + log.warn("[{}] Failed to process usage stats: {}", id, msg, e); + callback.onFailure(e); } - log.info("TB Usage Stats Consumer stopped."); }); + if (!processingTimeoutLatch.await(getNotificationPackProcessingTimeout(), TimeUnit.MILLISECONDS)) { + ctx.getAckMap().forEach((id, msg) -> log.warn("[{}] Timeout to process usage stats: {}", id, msg.getValue())); + ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process usage stats: {}", id, msg.getValue())); + } + consumer.commit(); + } - private void launchOtaPackageUpdateNotificationConsumer() { + private void processFirmwareMsgs(List> msgs, TbQueueConsumer> consumer) { long maxProcessingTimeoutPerRecord = firmwarePackInterval / firmwarePackSize; - firmwareStatesExecutor.submit(() -> { - while (!stopped) { - try { - List> msgs = firmwareStatesConsumer.poll(getNotificationPollDuration()); - if (msgs.isEmpty()) { - continue; - } - long timeToSleep = maxProcessingTimeoutPerRecord; - for (TbProtoQueueMsg msg : msgs) { - try { - long startTime = System.currentTimeMillis(); - boolean isSuccessUpdate = handleOtaPackageUpdates(msg); - long endTime = System.currentTimeMillis(); - long spentTime = endTime - startTime; - timeToSleep = timeToSleep - spentTime; - if (isSuccessUpdate) { - if (timeToSleep > 0) { - log.debug("Spent time per record is: [{}]!", spentTime); - Thread.sleep(timeToSleep); - timeToSleep = 0; - } - timeToSleep += maxProcessingTimeoutPerRecord; - } - } catch (Throwable e) { - log.warn("Failed to process firmware update msg: {}", msg, e); - } - } - firmwareStatesConsumer.commit(); - } catch (Exception e) { - if (!stopped) { - log.warn("Failed to obtain usage stats from queue.", e); - try { - Thread.sleep(getNotificationPollDuration()); - } catch (InterruptedException e2) { - log.trace("Failed to wait until the server has capacity to handle new firmware updates", e2); - } + long timeToSleep = maxProcessingTimeoutPerRecord; + for (TbProtoQueueMsg msg : msgs) { + try { + long startTime = System.currentTimeMillis(); + boolean isSuccessUpdate = handleOtaPackageUpdates(msg); + long endTime = System.currentTimeMillis(); + long spentTime = endTime - startTime; + timeToSleep = timeToSleep - spentTime; + if (isSuccessUpdate) { + if (timeToSleep > 0) { + log.debug("Spent time per record is: [{}]!", spentTime); + Thread.sleep(timeToSleep); + timeToSleep = 0; } + timeToSleep += maxProcessingTimeoutPerRecord; } + } catch (Throwable e) { + log.warn("Failed to process firmware update msg: {}", msg, e); } - log.info("TB Ota Package States Consumer stopped."); - }); + } + consumer.commit(); } private void handleUsageStats(TbProtoQueueMsg msg, TbCallback callback) { @@ -803,15 +764,16 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService consumers = new ConcurrentHashMap<>(); public DefaultTbRuleEngineConsumerService(TbRuleEngineConsumerContext ctx, - TbRuleEngineQueueFactory tbRuleEngineQueueFactory, ActorSystemContext actorContext, DataDecodingEncodingService encodingService, TbRuleEngineDeviceRpcService tbDeviceRpcService, @@ -89,8 +86,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< PartitionService partitionService, ApplicationEventPublisher eventPublisher, JwtSettingsService jwtSettingsService) { - super(actorContext, encodingService, tenantProfileCache, deviceProfileCache, assetProfileCache, apiUsageStateService, partitionService, - eventPublisher, tbRuleEngineQueueFactory.createToRuleEngineNotificationsMsgConsumer(), jwtSettingsService); + super(actorContext, encodingService, tenantProfileCache, deviceProfileCache, assetProfileCache, apiUsageStateService, partitionService, eventPublisher, jwtSettingsService); this.ctx = ctx; this.tbDeviceRpcService = tbDeviceRpcService; this.queueService = queueService; @@ -98,7 +94,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< @PostConstruct public void init() { - super.init("tb-rule-engine-notifications-consumer"); + super.init("tb-rule-engine"); List queues = queueService.findAllQueues(); for (Queue configuration : queues) { if (partitionService.isManagedByCurrentService(configuration.getTenantId())) { @@ -130,20 +126,11 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< }); } - @AfterStartUp(order = AfterStartUp.REGULAR_SERVICE) - public void onApplicationEvent(ApplicationReadyEvent event) { - super.onApplicationEvent(event); - ctx.setReady(true); - } - - @Override - protected void launchMainConsumers() {} - @Override protected void stopConsumers() { + super.stopConsumers(); consumers.values().forEach(TbRuleEngineQueueConsumerManager::stop); consumers.values().forEach(TbRuleEngineQueueConsumerManager::awaitStop); - ctx.stop(); } @Override @@ -161,6 +148,16 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< return ctx.getPackProcessingTimeout(); } + @Override + protected int getMgmtThreadPoolSize() { + return ctx.getMgmtThreadPoolSize(); + } + + @Override + protected TbQueueConsumer> createNotificationsConsumer() { + return ctx.getQueueFactory().createToRuleEngineNotificationsMsgConsumer(); + } + @Override protected void handleNotification(UUID id, TbProtoQueueMsg msg, TbCallback callback) throws Exception { ToRuleEngineNotificationMsg nfMsg = msg.getValue(); @@ -237,7 +234,13 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< } private TbRuleEngineQueueConsumerManager createConsumer(QueueKey queueKey, Queue queue) { - var consumer = new TbRuleEngineQueueConsumerManager(ctx, queueKey); + var consumer = TbRuleEngineQueueConsumerManager.create() + .ctx(ctx) + .queueKey(queueKey) + .consumerExecutor(consumersExecutor) + .scheduler(scheduler) + .taskExecutor(mgmtExecutor) + .build(); consumers.put(queueKey, consumer); consumer.init(queue); return consumer; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/consumer/BasicQueueConsumerManager.java b/application/src/main/java/org/thingsboard/server/service/queue/consumer/BasicQueueConsumerManager.java new file mode 100644 index 0000000000..1d3773e3a1 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/queue/consumer/BasicQueueConsumerManager.java @@ -0,0 +1,109 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.queue.consumer; + +import lombok.Builder; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueMsg; + +import java.util.List; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.function.Supplier; + +@Slf4j +public class BasicQueueConsumerManager { + + private final String key; + private final String name; + private final long pollInterval; + private final MsgPackProcessor msgPackProcessor; + private final ExecutorService consumerExecutor; + + @Getter + private final TbQueueConsumer consumer; + private volatile boolean stopped; + + @Builder + public BasicQueueConsumerManager(String key, String name, + long pollInterval, + MsgPackProcessor msgPackProcessor, + Supplier> consumerCreator, + ExecutorService consumerExecutor) { + this.key = key; + this.name = name; + this.pollInterval = pollInterval; + this.msgPackProcessor = msgPackProcessor; + this.consumerExecutor = consumerExecutor; + this.consumer = consumerCreator.get(); + } + + public void subscribe() { + consumer.subscribe(); + } + + public void subscribe(Set partitions) { + consumer.subscribe(partitions); + } + + public void launch() { + log.info("[{}] Launching consumer", name); + consumerExecutor.submit(() -> { + ThingsBoardThreadFactory.addThreadNamePrefix(key); + try { + consumerLoop(consumer); + } catch (Throwable e) { + log.error("Failure in consumer loop", e); + } + }); + } + + private void consumerLoop(TbQueueConsumer consumer) { + while (!stopped && !consumer.isStopped()) { + try { + List msgs = consumer.poll(pollInterval); + if (msgs.isEmpty()) { + continue; + } + msgPackProcessor.process(msgs, consumer); + } catch (Exception e) { + if (!consumer.isStopped()) { + log.warn("Failed to process messages from queue", e); + try { + Thread.sleep(pollInterval); + } catch (InterruptedException interruptedException) { + log.trace("Failed to wait until the server has capacity to handle new requests", interruptedException); + } + } + } + } + log.info("{} Consumer stopped", name); + } + + public void stop() { + log.debug("[{}] Stopping consumer", name); + stopped = true; + consumer.unsubscribe(); + } + + public interface MsgPackProcessor { + void process(List msgs, TbQueueConsumer consumer) throws Exception; + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java b/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java new file mode 100644 index 0000000000..04538289ea --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java @@ -0,0 +1,321 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.queue.consumer; + +import lombok.Builder; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.common.data.queue.QueueConfig; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueMsg; +import org.thingsboard.server.queue.discovery.QueueKey; +import org.thingsboard.server.service.queue.ruleengine.QueueEvent; +import org.thingsboard.server.service.queue.ruleengine.TbQueueConsumerManagerTask; +import org.thingsboard.server.service.queue.ruleengine.TbQueueConsumerTask; + +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Function; +import java.util.stream.Collectors; + +@Slf4j +public class QueueConsumerManager { + + protected final QueueKey queueKey; + @Getter + protected C config; + protected final MsgPackProcessor msgPackProcessor; + protected final Function> consumerCreator; + protected final ExecutorService consumerExecutor; + protected final ScheduledExecutorService scheduler; + protected final ExecutorService taskExecutor; + + private final java.util.Queue tasks = new ConcurrentLinkedQueue<>(); + private final ReentrantLock lock = new ReentrantLock(); + + @Getter + private volatile Set partitions; + protected volatile ConsumerWrapper consumerWrapper; + protected volatile boolean stopped; + + @Builder + public QueueConsumerManager(QueueKey queueKey, + C config, + MsgPackProcessor msgPackProcessor, + Function> consumerCreator, + ExecutorService consumerExecutor, + ScheduledExecutorService scheduler, + ExecutorService taskExecutor) { + this.queueKey = queueKey; + this.config = config; + this.msgPackProcessor = msgPackProcessor; + this.consumerCreator = consumerCreator; + this.consumerExecutor = consumerExecutor; + this.scheduler = scheduler; + this.taskExecutor = taskExecutor; + if (config != null) { + init(config); + } + } + + public void init(C config) { + this.config = config; + if (config.isConsumerPerPartition()) { + this.consumerWrapper = new ConsumerPerPartitionWrapper(); + } else { + this.consumerWrapper = new SingleConsumerWrapper(); + } + log.debug("[{}] Initialized consumer for queue: {}", queueKey, config); + } + + public void update(C config) { + addTask(TbQueueConsumerManagerTask.configUpdate(config)); + } + + public void update(Set partitions) { + addTask(TbQueueConsumerManagerTask.partitionChange(partitions)); + } + + protected void addTask(TbQueueConsumerManagerTask todo) { + if (stopped) { + return; + } + tasks.add(todo); + log.trace("[{}] Added task: {}", queueKey, todo); + tryProcessTasks(); + } + + private void tryProcessTasks() { + taskExecutor.submit(() -> { + if (lock.tryLock()) { + try { + C newConfig = null; + Set newPartitions = null; + while (!stopped) { + TbQueueConsumerManagerTask task = tasks.poll(); + 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) { + newConfig = (C) task.getConfig(); + } else { + processTask(task); + } + } + if (stopped) { + return; + } + if (newConfig != null) { + doUpdate(newConfig); + } + if (newPartitions != null) { + doUpdate(newPartitions); + } + } catch (Exception e) { + log.error("[{}] Failed to process tasks", queueKey, e); + } finally { + lock.unlock(); + } + } else { + log.trace("[{}] Failed to acquire lock", queueKey); + scheduler.schedule(this::tryProcessTasks, 1, TimeUnit.SECONDS); + } + }); + } + + protected void processTask(TbQueueConsumerManagerTask task) { + } + + private void doUpdate(C newConfig) { + log.info("[{}] Processing queue update: {}", queueKey, newConfig); + var oldConfig = this.config; + this.config = newConfig; + if (log.isTraceEnabled()) { + log.trace("[{}] Old queue configuration: {}", queueKey, oldConfig); + log.trace("[{}] New queue configuration: {}", queueKey, newConfig); + } + + if (oldConfig == null) { + init(config); + } else if (newConfig.isConsumerPerPartition() != oldConfig.isConsumerPerPartition()) { + consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::initiateStop); + consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitCompletion); + + init(config); + if (partitions != null) { + doUpdate(partitions); // even if partitions number was changed, there can be no partition change event + } + } else { + // 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 doUpdate(Set partitions) { + this.partitions = partitions; + consumerWrapper.updatePartitions(partitions); + } + + private void launchConsumer(TbQueueConsumerTask consumerTask) { + log.info("[{}] Launching consumer", consumerTask.getKey()); + Future consumerLoop = consumerExecutor.submit(() -> { + ThingsBoardThreadFactory.updateCurrentThreadName(consumerTask.getKey().toString()); + try { + consumerLoop(consumerTask.getConsumer()); + } catch (Throwable e) { + log.error("Failure in consumer loop", e); + } + }); + consumerTask.setTask(consumerLoop); + } + + private void consumerLoop(TbQueueConsumer consumer) { + while (!stopped && !consumer.isStopped()) { + try { + List msgs = consumer.poll(config.getPollInterval()); + if (msgs.isEmpty()) { + continue; + } + processMsgs(msgs, consumer, config); + } catch (Exception e) { + if (!consumer.isStopped()) { + log.warn("Failed to process messages from queue", e); + try { + Thread.sleep(config.getPollInterval()); + } catch (InterruptedException e2) { + log.trace("Failed to wait until the server has capacity to handle new requests", e2); + } + } + } + } + if (consumer.isStopped()) { + consumer.unsubscribe(); + } + log.info("{} Consumer stopped", queueKey); + } + + protected void processMsgs(List msgs, TbQueueConsumer consumer, C config) throws Exception { + msgPackProcessor.process(msgs, consumer, config); + } + + public void stop() { + log.debug("[{}] Stopping consumers", queueKey); + consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::initiateStop); + stopped = true; + } + + public void awaitStop() { + consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitCompletion); + log.debug("[{}] Unsubscribed and stopped consumers", queueKey); + } + + private static String partitionsToString(Collection partitions) { + return partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.joining(", ", "[", "]")); + } + + public interface MsgPackProcessor { + void process(List msgs, TbQueueConsumer consumer, C config) throws Exception; + } + + public 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).awaitCompletion()); + + addedPartitions.forEach((tpi) -> { + String key = queueKey + "-" + tpi.getPartition().orElse(-999999); + TbQueueConsumerTask consumer = new TbQueueConsumerTask<>(key, consumerCreator.apply(config)); + 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.awaitCompletion(); + } + consumer = null; + return; + } + + if (consumer == null) { + consumer = new TbQueueConsumerTask<>(queueKey, consumerCreator.apply(config)); + } + 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/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index 26e23549e3..e2a6a3f117 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.queue.processing; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.context.event.ApplicationReadyEvent; import org.springframework.context.ApplicationEventPublisher; +import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.common.data.EntityType; @@ -45,6 +46,7 @@ import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.service.queue.TbPackCallback; import org.thingsboard.server.service.queue.TbPackProcessingContext; +import org.thingsboard.server.service.queue.consumer.BasicQueueConsumerManager; import org.thingsboard.server.service.security.auth.jwt.settings.JwtSettingsService; import javax.annotation.PreDestroy; @@ -55,15 +57,13 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @Slf4j public abstract class AbstractConsumerService extends TbApplicationEventListener { - protected volatile ExecutorService notificationsConsumerExecutor; - protected volatile boolean stopped = false; - protected volatile boolean isReady = false; protected final ActorSystemContext actorContext; protected final DataDecodingEncodingService encodingService; protected final TbTenantProfileCache tenantProfileCache; @@ -72,16 +72,19 @@ public abstract class AbstractConsumerService> nfConsumer; protected final JwtSettingsService jwtSettingsService; + protected BasicQueueConsumerManager> nfConsumer; + + protected ExecutorService consumersExecutor; + protected ExecutorService mgmtExecutor; + protected ScheduledExecutorService scheduler; public AbstractConsumerService(ActorSystemContext actorContext, DataDecodingEncodingService encodingService, TbTenantProfileCache tenantProfileCache, TbDeviceProfileCache deviceProfileCache, TbAssetProfileCache assetProfileCache, TbApiUsageStateService apiUsageStateService, PartitionService partitionService, ApplicationEventPublisher eventPublisher, - TbQueueConsumer> nfConsumer, JwtSettingsService jwtSettingsService) { + JwtSettingsService jwtSettingsService) { this.actorContext = actorContext; this.encodingService = encodingService; this.tenantProfileCache = tenantProfileCache; @@ -90,21 +93,32 @@ public abstract class AbstractConsumerService>builder() + .key("notifications") + .name("TB Notifications") + .pollInterval(getNotificationPollDuration()) + .msgPackProcessor(this::processNotifications) + .consumerCreator(this::createNotificationsConsumer) + .consumerExecutor(consumersExecutor) + .build(); } @AfterStartUp(order = AfterStartUp.REGULAR_SERVICE) public void onApplicationEvent(ApplicationReadyEvent event) { - log.info("Subscribing to notifications: {}", nfConsumer.getTopic()); - this.nfConsumer.subscribe(); - this.isReady = true; - launchNotificationsConsumer(); - launchMainConsumers(); + startConsumers(); + } + + protected void startConsumers() { + nfConsumer.subscribe(); + nfConsumer.launch(); } @Override @@ -114,58 +128,42 @@ public abstract class AbstractConsumerService { - while (!stopped) { - try { - List> msgs = nfConsumer.poll(getNotificationPollDuration()); - if (msgs.isEmpty()) { - continue; - } - List> orderedMsgList = msgs.stream().map(msg -> new IdMsgPair<>(UUID.randomUUID(), msg)).collect(Collectors.toList()); - ConcurrentMap> pendingMap = orderedMsgList.stream().collect( - Collectors.toConcurrentMap(IdMsgPair::getUuid, IdMsgPair::getMsg)); - CountDownLatch processingTimeoutLatch = new CountDownLatch(1); - TbPackProcessingContext> ctx = new TbPackProcessingContext<>( - processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>()); - orderedMsgList.forEach(element -> { - UUID id = element.getUuid(); - TbProtoQueueMsg msg = element.getMsg(); - log.trace("[{}] Creating notification callback for message: {}", id, msg.getValue()); - TbCallback callback = new TbPackCallback<>(id, ctx); - try { - handleNotification(id, msg, callback); - } catch (Throwable e) { - log.warn("[{}] Failed to process notification: {}", id, msg, e); - callback.onFailure(e); - } - }); - if (!processingTimeoutLatch.await(getNotificationPackProcessingTimeout(), TimeUnit.MILLISECONDS)) { - ctx.getAckMap().forEach((id, msg) -> log.warn("[{}] Timeout to process notification: {}", id, msg.getValue())); - ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process notification: {}", id, msg.getValue())); - } - nfConsumer.commit(); - } catch (Exception e) { - if (!stopped) { - log.warn("Failed to obtain notifications from queue.", e); - try { - Thread.sleep(getNotificationPollDuration()); - } catch (InterruptedException e2) { - log.trace("Failed to wait until the server has capacity to handle new notifications", e2); - } - } - } + protected abstract int getMgmtThreadPoolSize(); + + protected abstract TbQueueConsumer> createNotificationsConsumer(); + + protected void processNotifications(List> msgs, TbQueueConsumer> consumer) throws Exception { + List> orderedMsgList = msgs.stream().map(msg -> new IdMsgPair<>(UUID.randomUUID(), msg)).collect(Collectors.toList()); + ConcurrentMap> pendingMap = orderedMsgList.stream().collect( + Collectors.toConcurrentMap(IdMsgPair::getUuid, IdMsgPair::getMsg)); + CountDownLatch processingTimeoutLatch = new CountDownLatch(1); + TbPackProcessingContext> ctx = new TbPackProcessingContext<>( + processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>()); + orderedMsgList.forEach(element -> { + UUID id = element.getUuid(); + TbProtoQueueMsg msg = element.getMsg(); + log.trace("[{}] Creating notification callback for message: {}", id, msg.getValue()); + TbCallback callback = new TbPackCallback<>(id, ctx); + try { + handleNotification(id, msg, callback); + } catch (Throwable e) { + log.warn("[{}] Failed to process notification: {}", id, msg, e); + callback.onFailure(e); } - log.info("TB Notifications Consumer stopped."); }); + if (!processingTimeoutLatch.await(getNotificationPackProcessingTimeout(), TimeUnit.MILLISECONDS)) { + ctx.getAckMap().forEach((id, msg) -> log.warn("[{}] Timeout to process notification: {}", id, msg.getValue())); + ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process notification: {}", id, msg.getValue())); + } + consumer.commit(); } protected final void handleComponentLifecycleMsg(UUID id, ComponentLifecycleMsg componentLifecycleMsg) { @@ -219,13 +217,15 @@ public abstract class AbstractConsumerService partitions; private boolean drainQueue; @@ -37,8 +37,8 @@ public class TbQueueConsumerManagerTask { return new TbQueueConsumerManagerTask(QueueEvent.DELETE, null, null, drainQueue); } - public static TbQueueConsumerManagerTask configUpdate(Queue queue) { - return new TbQueueConsumerManagerTask(QueueEvent.CONFIG_UPDATE, queue, null, false); + public static TbQueueConsumerManagerTask configUpdate(QueueConfig config) { + return new TbQueueConsumerManagerTask(QueueEvent.CONFIG_UPDATE, config, null, false); } public static TbQueueConsumerManagerTask partitionChange(Set 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 0e25efd014..84539d4d7c 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 @@ -20,9 +20,8 @@ import lombok.RequiredArgsConstructor; import lombok.Setter; 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.TbQueueMsg; import java.util.Set; import java.util.concurrent.Future; @@ -30,12 +29,12 @@ import java.util.concurrent.TimeUnit; @RequiredArgsConstructor @Slf4j -public class TbQueueConsumerTask { +public class TbQueueConsumerTask { @Getter private final Object key; @Getter - private final TbQueueConsumer> consumer; + private final TbQueueConsumer consumer; @Setter private Future task; 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 cf876c6947..08d2931246 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 @@ -19,8 +19,6 @@ 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.ThingsBoardExecutors; -import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.queue.TbQueueAdmin; @@ -33,11 +31,6 @@ import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStr import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategyFactory; import org.thingsboard.server.service.stats.RuleEngineStatisticsService; -import javax.annotation.PostConstruct; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; - @Component @TbRuleEngineComponent @Slf4j @@ -68,22 +61,4 @@ public class TbRuleEngineConsumerContext { private final TbQueueProducerProvider producerProvider; private final TbQueueAdmin queueAdmin; - private ExecutorService consumersExecutor; - private ExecutorService mgmtExecutor; - private ScheduledExecutorService scheduler; - - private volatile boolean isReady = false; - - @PostConstruct - void init() { - this.consumersExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer")); - this.mgmtExecutor = ThingsBoardExecutors.newWorkStealingPool(mgmtThreadPoolSize, "tb-rule-engine-mgmt"); - this.scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-scheduler")); - } - - public void stop() { - scheduler.shutdownNow(); - consumersExecutor.shutdownNow(); - mgmtExecutor.shutdownNow(); - } } 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 2da3dbc6dc..b93313b104 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 @@ -16,9 +16,8 @@ package org.thingsboard.server.service.queue.ruleengine; import com.google.protobuf.ProtocolStringList; -import lombok.Getter; +import lombok.Builder; 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; @@ -38,173 +37,54 @@ import org.thingsboard.server.queue.discovery.QueueKey; import org.thingsboard.server.service.queue.TbMsgPackCallback; import org.thingsboard.server.service.queue.TbMsgPackProcessingContext; import org.thingsboard.server.service.queue.TbRuleEngineConsumerStats; +import org.thingsboard.server.service.queue.consumer.QueueConsumerManager; 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.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.ConcurrentLinkedQueue; -import java.util.concurrent.Future; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; @Slf4j -public class TbRuleEngineQueueConsumerManager { +public class TbRuleEngineQueueConsumerManager extends QueueConsumerManager, Queue> { public static final String SUCCESSFUL_STATUS = "successful"; public static final String FAILED_STATUS = "failed"; private final TbRuleEngineConsumerContext ctx; - private final QueueKey queueKey; private final TbRuleEngineConsumerStats stats; - private final ReentrantLock lock = new ReentrantLock(); //NonfairSync - @Getter - private volatile Queue queue; - @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 queueKey) { + @Builder(builderMethodName = "create") // not to conflict with super.builder() + public TbRuleEngineQueueConsumerManager(TbRuleEngineConsumerContext ctx, + QueueKey queueKey, + ExecutorService consumerExecutor, + ScheduledExecutorService scheduler, + ExecutorService taskExecutor) { + super(queueKey, null, null, ctx.getQueueFactory()::createToRuleEngineMsgConsumer, consumerExecutor, scheduler, taskExecutor); this.ctx = ctx; - this.queueKey = queueKey; this.stats = new TbRuleEngineConsumerStats(queueKey, ctx.getStatsFactory()); } - public void init(Queue queue) { - this.queue = queue; - if (queue.isConsumerPerPartition()) { - this.consumerWrapper = new ConsumerPerPartitionWrapper(); - } else { - this.consumerWrapper = new SingleConsumerWrapper(); - } - log.debug("[{}] Initialized consumer for queue: {}", queueKey, queue); - } - - public void update(Queue queue) { - addTask(TbQueueConsumerManagerTask.configUpdate(queue)); - } - - public void update(Set partitions) { - addTask(TbQueueConsumerManagerTask.partitionChange(partitions)); - } - public void delete(boolean drainQueue) { addTask(TbQueueConsumerManagerTask.delete(drainQueue)); } - private void addTask(TbQueueConsumerManagerTask todo) { - if (stopped) { - return; - } - tasks.add(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 (!stopped) { - TbQueueConsumerManagerTask task = tasks.poll(); - 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.DELETE) { - doDelete(task.isDrainQueue()); - return; - } - } - if (stopped) { - return; - } - if (newConfiguration != null) { - doUpdate(newConfiguration); - } - if (newPartitions != null) { - doUpdate(newPartitions); - } - } catch (Exception e) { - log.error("[{}] Failed to process tasks", queueKey, e); - } finally { - lock.unlock(); - } - } else { - log.trace("[{}] Failed to acquire lock", queueKey); - ctx.getScheduler().schedule(this::tryProcessTasks, 1, TimeUnit.SECONDS); - } - }); - } - - private void doUpdate(Queue newQueue) { - log.info("[{}] Processing queue update: {}", queueKey, newQueue); - var oldQueue = this.queue; - this.queue = newQueue; - if (log.isTraceEnabled()) { - log.trace("[{}] Old queue configuration: {}", queueKey, oldQueue); - log.trace("[{}] New queue configuration: {}", queueKey, newQueue); - } - - if (oldQueue == null) { - init(queue); - } else if (newQueue.isConsumerPerPartition() != oldQueue.isConsumerPerPartition()) { - consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::initiateStop); - consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitCompletion); - - init(queue); - if (partitions != null) { - doUpdate(partitions); // even if partitions number was changed, there can be no partition change event - } - } else { - // 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 + @Override + protected void processTask(TbQueueConsumerManagerTask task) { + if (task.getEvent() == QueueEvent.DELETE) { + doDelete(task.isDrainQueue()); } } - private void doUpdate(Set partitions) { - this.partitions = partitions; - consumerWrapper.updatePartitions(partitions); - } - - public void stop() { - log.debug("[{}] Stopping consumers", queueKey); - consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::initiateStop); - stopped = true; - } - - public void awaitStop() { - consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitCompletion); - log.debug("[{}] Unsubscribed and stopped consumers", queueKey); - } - private void doDelete(boolean drainQueue) { stopped = true; log.info("[{}] Handling queue deletion", queueKey); @@ -212,7 +92,7 @@ public class TbRuleEngineQueueConsumerManager { List>> queueConsumers = consumerWrapper.getConsumers().stream() .map(TbQueueConsumerTask::getConsumer).collect(Collectors.toList()); - ctx.getConsumersExecutor().submit(() -> { + consumerExecutor.submit(() -> { if (drainQueue) { drainQueue(queueConsumers); } @@ -235,47 +115,10 @@ public class TbRuleEngineQueueConsumerManager { }); } - private void launchConsumer(TbQueueConsumerTask consumerTask) { - log.info("[{}] Launching consumer", consumerTask.getKey()); - Future consumerLoop = ctx.getConsumersExecutor().submit(() -> { - ThingsBoardThreadFactory.updateCurrentThreadName(consumerTask.getKey().toString()); - try { - consumerLoop(consumerTask.getConsumer()); - } catch (Throwable e) { - log.error("Failure in consumer loop", e); - } - }); - consumerTask.setTask(consumerLoop); - } - - private void consumerLoop(TbQueueConsumer> consumer) { - while (!stopped && !consumer.isStopped()) { - try { - List> msgs = consumer.poll(queue.getPollInterval()); - if (msgs.isEmpty()) { - continue; - } - processMsgs(msgs, consumer, queue); - } catch (Exception e) { - if (!consumer.isStopped()) { - 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); - } - } - } - } - if (consumer.isStopped()) { - consumer.unsubscribe(); - } - log.info("Rule Engine consumer stopped"); - } - - private void processMsgs(List> msgs, - TbQueueConsumer> consumer, - Queue queue) throws InterruptedException { + @Override + protected void processMsgs(List> msgs, + TbQueueConsumer> consumer, + Queue queue) throws Exception { TbRuleEngineSubmitStrategy submitStrategy = getSubmitStrategy(queue); TbRuleEngineProcessingStrategy ackStrategy = getProcessingStrategy(queue); submitStrategy.init(msgs); @@ -320,7 +163,7 @@ public class TbRuleEngineQueueConsumerManager { } private void submitMessage(TbMsgPackProcessingContext packCtx, UUID id, TbProtoQueueMsg msg) { - log.trace("[{}] Creating callback for topic {} message: {}", id, queue.getName(), msg.getValue()); + log.trace("[{}] Creating callback for topic {} message: {}", id, config.getName(), msg.getValue()); ToRuleEngineMsg toRuleEngineMsg = msg.getValue(); TenantId tenantId = TenantId.fromUUID(new UUID(toRuleEngineMsg.getTenantIdMSB(), toRuleEngineMsg.getTenantIdLSB())); TbMsgCallback callback = ctx.isPrometheusStatsEnabled() ? @@ -328,7 +171,7 @@ public class TbRuleEngineQueueConsumerManager { new TbMsgPackCallback(id, tenantId, packCtx); try { if (!toRuleEngineMsg.getTbMsg().isEmpty()) { - forwardToRuleEngineActor(queue.getName(), tenantId, toRuleEngineMsg, callback); + forwardToRuleEngineActor(config.getName(), tenantId, toRuleEngineMsg, callback); } else { callback.onSuccess(); } @@ -356,7 +199,7 @@ public class TbRuleEngineQueueConsumerManager { 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); + TbMsg tmpMsg = TbMsg.fromBytes(config.getName(), tmp.getTbMsg().toByteArray(), TbMsgCallback.EMPTY); RuleNodeInfo ruleNodeInfo = ctx.getLastVisitedRuleNode(pending.getKey()); if (printAll) { log.trace("[{}][{}] {} to process message: {}, Last Rule Node: {}", queueKey, TenantId.fromUUID(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo); @@ -379,7 +222,7 @@ public class TbRuleEngineQueueConsumerManager { int n = 0; while (System.currentTimeMillis() <= finishTs) { for (TbQueueConsumer> consumer : consumers) { - List> msgs = consumer.poll(queue.getPollInterval()); + List> msgs = consumer.poll(config.getPollInterval()); if (msgs.isEmpty()) { continue; } @@ -388,7 +231,7 @@ public class TbRuleEngineQueueConsumerManager { 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); + TopicPartitionInfo tpi = ctx.getPartitionService().resolve(ServiceType.TB_RULE_ENGINE, config.getName(), TenantId.SYS_TENANT_ID, originator); ctx.getProducerProvider().getRuleEngineMsgProducer().send(tpi, msg, null); n++; } catch (Throwable e) { @@ -399,90 +242,11 @@ public class TbRuleEngineQueueConsumerManager { } } if (n > 0) { - log.info("Moved {} messages from {} to system {}", n, queueKey, queue.getName()); + log.info("Moved {} messages from {} to system {}", n, queueKey, config.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).awaitCompletion(); - }); - - 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.awaitCompletion(); - } - 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/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java b/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java index 6bd0cd9c0e..a00f45b8db 100644 --- a/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java +++ b/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java @@ -27,6 +27,8 @@ import org.mockito.Mockito; import org.mockito.junit.MockitoJUnitRunner; import org.testcontainers.shaded.org.apache.commons.lang3.RandomUtils; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.QueueId; @@ -65,6 +67,9 @@ import java.util.Objects; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; @@ -86,7 +91,6 @@ import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; @@ -112,6 +116,9 @@ public class TbRuleEngineQueueConsumerManagerTest { @Mock private TbQueueAdmin queueAdmin; private TbRuleEngineConsumerContext ruleEngineConsumerContext; + private ExecutorService consumersExecutor; + private ScheduledExecutorService scheduler; + private ExecutorService mgmtExecutor; private TbRuleEngineQueueConsumerManager consumerManager; private Queue queue; @@ -141,10 +148,10 @@ public class TbRuleEngineQueueConsumerManagerTest { }).when(actorContext).tell(any()); ruleEngineMsgProducer = mock(TbQueueProducer.class); when(producerProvider.getRuleEngineMsgProducer()).thenReturn(ruleEngineMsgProducer); - ruleEngineConsumerContext.setMgmtThreadPoolSize(2); + consumersExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer")); + mgmtExecutor = ThingsBoardExecutors.newWorkStealingPool(3, "tb-rule-engine-mgmt"); + scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-scheduler")); ruleEngineConsumerContext.setTopicDeletionDelayInSec(5); - ruleEngineConsumerContext.init(); - ruleEngineConsumerContext.setReady(false); queue = new Queue(); queue.setName("Test"); @@ -174,14 +181,23 @@ public class TbRuleEngineQueueConsumerManagerTest { }).when(queueFactory).createToRuleEngineMsgConsumer(any()); QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queue); - consumerManager = new TbRuleEngineQueueConsumerManager(ruleEngineConsumerContext, queueKey); + consumerManager = TbRuleEngineQueueConsumerManager.create() + .ctx(ruleEngineConsumerContext) + .queueKey(queueKey) + .consumerExecutor(consumersExecutor) + .scheduler(scheduler) + .taskExecutor(mgmtExecutor) + .build(); } @After public void afterEach() { consumerManager.stop(); consumerManager.awaitStop(); - ruleEngineConsumerContext.stop(); + + consumersExecutor.shutdownNow(); + scheduler.shutdownNow(); + mgmtExecutor.shutdownNow(); if (generateQueueMsgs) { await().atMost(10, TimeUnit.SECONDS) @@ -199,14 +215,7 @@ public class TbRuleEngineQueueConsumerManagerTest { Set partitions = createTpis(2, 3, 4); consumerManager.update(partitions); - partitions = createTpis(3, 4, 5); - consumerManager.update(partitions); - partitions = createTpis(1, 2, 3); - consumerManager.update(partitions); - // simulated multiple partition change events before consumer is ready; only latest partitions should be processed - verifyNoInteractions(queueFactory); - ruleEngineConsumerContext.setReady(true); await().atMost(2, TimeUnit.SECONDS) .until(() -> consumers.size() == 3); for (TopicPartitionInfo partition : partitions) { @@ -222,14 +231,7 @@ public class TbRuleEngineQueueConsumerManagerTest { Set partitions = createTpis(2, 3, 4); consumerManager.update(partitions); - partitions = createTpis(3, 4, 5); - consumerManager.update(partitions); - partitions = createTpis(1, 2, 3); - consumerManager.update(partitions); - - verifyNoInteractions(queueFactory); - ruleEngineConsumerContext.setReady(true); await().atMost(2, TimeUnit.SECONDS) .until(() -> consumers.size() == 1); TestConsumer consumer = getConsumer(); @@ -240,7 +242,6 @@ public class TbRuleEngineQueueConsumerManagerTest { public void testPartitionsUpdate_singleConsumer() { queue.setConsumerPerPartition(false); consumerManager.init(queue); - ruleEngineConsumerContext.setReady(true); Set partitions = Collections.emptySet(); consumerManager.update(partitions); @@ -273,7 +274,6 @@ public class TbRuleEngineQueueConsumerManagerTest { public void testPartitionsUpdate_consumerPerPartition() { queue.setConsumerPerPartition(true); consumerManager.init(queue); - ruleEngineConsumerContext.setReady(true); consumerManager.update(Collections.emptySet()); verify(queueFactory, after(1000).never()).createToRuleEngineMsgConsumer(any()); @@ -316,7 +316,6 @@ public class TbRuleEngineQueueConsumerManagerTest { public void testConfigUpdate_singleConsumer() { queue.setConsumerPerPartition(false); consumerManager.init(queue); - ruleEngineConsumerContext.setReady(true); Set partitions = createTpis(1, 2, 3); consumerManager.update(partitions); TestConsumer consumer = getConsumer(); @@ -342,7 +341,6 @@ public class TbRuleEngineQueueConsumerManagerTest { public void testConfigUpdate_consumerPerPartition() { queue.setConsumerPerPartition(true); consumerManager.init(queue); - ruleEngineConsumerContext.setReady(true); Set partitions = createTpis(1, 2, 3); consumerManager.update(partitions); TestConsumer consumer1 = getConsumer(1); @@ -375,7 +373,6 @@ public class TbRuleEngineQueueConsumerManagerTest { public void testConfigUpdate_fromSingleToConsumerPerPartition() { queue.setConsumerPerPartition(false); consumerManager.init(queue); - ruleEngineConsumerContext.setReady(true); Set partitions = createTpis(1, 2, 3); consumerManager.update(partitions); TestConsumer consumer = getConsumer(); @@ -395,7 +392,6 @@ public class TbRuleEngineQueueConsumerManagerTest { public void testConfigUpdate_fromConsumerPerPartitionToSingle() { queue.setConsumerPerPartition(true); consumerManager.init(queue); - ruleEngineConsumerContext.setReady(true); Set partitions = createTpis(1, 2, 3); consumerManager.update(partitions); TestConsumer consumer1 = getConsumer(1); @@ -419,7 +415,6 @@ public class TbRuleEngineQueueConsumerManagerTest { public void testStop() { queue.setConsumerPerPartition(true); consumerManager.init(queue); - ruleEngineConsumerContext.setReady(true); consumerManager.update(createTpis(1)); TestConsumer consumer = getConsumer(1); verifySubscribedAndLaunched(consumer, 1); @@ -437,7 +432,6 @@ public class TbRuleEngineQueueConsumerManagerTest { public void testDelete_consumerPerPartition() { queue.setConsumerPerPartition(true); consumerManager.init(queue); - ruleEngineConsumerContext.setReady(true); Set partitions = createTpis(1, 2); consumerManager.update(partitions); TestConsumer consumer1 = getConsumer(1); @@ -459,7 +453,7 @@ public class TbRuleEngineQueueConsumerManagerTest { int msgCount = totalConsumedMsgs.get(); await().atLeast(2, TimeUnit.SECONDS) // based on topicDeletionDelayInSec(5) = 5 - ( 3 seconds the code may execute starting consumerManager.delete() call) - .atMost(7, TimeUnit.SECONDS) + .atMost(10, TimeUnit.SECONDS) .untilAsserted(() -> { partitions.stream() .map(TopicPartitionInfo::getFullTopicName) @@ -481,7 +475,6 @@ public class TbRuleEngineQueueConsumerManagerTest { public void testDelete_singleConsumer() { queue.setConsumerPerPartition(false); consumerManager.init(queue); - ruleEngineConsumerContext.setReady(true); Set partitions = createTpis(1, 2); consumerManager.update(partitions); TestConsumer consumer = getConsumer(); @@ -499,7 +492,7 @@ public class TbRuleEngineQueueConsumerManagerTest { int msgCount = totalConsumedMsgs.get(); await().atLeast(2, TimeUnit.SECONDS) // based on topicDeletionDelayInSec(5) = 5 - ( 3 seconds the code may execute starting consumerManager.delete() call) - .atMost(7, TimeUnit.SECONDS) + .atMost(10, TimeUnit.SECONDS) .untilAsserted(() -> { partitions.stream() .map(TopicPartitionInfo::getFullTopicName) @@ -520,10 +513,9 @@ public class TbRuleEngineQueueConsumerManagerTest { public void testManyDifferentUpdates() throws Exception { queue.setConsumerPerPartition(RandomUtils.nextBoolean()); consumerManager.init(queue); - ruleEngineConsumerContext.setReady(true); Supplier queueConfigUpdater = () -> { - Queue oldConfig = consumerManager.getQueue(); + Queue oldConfig = consumerManager.getConfig(); Queue newConfig = JacksonUtil.clone(oldConfig); newConfig.setConsumerPerPartition(RandomUtils.nextBoolean()); newConfig.setPollInterval(RandomUtils.nextInt(100, 501)); @@ -571,7 +563,7 @@ public class TbRuleEngineQueueConsumerManagerTest { Set expectedPartitions = latestPartitions; await().atMost(5, TimeUnit.SECONDS) .untilAsserted(() -> { - assertThat(consumerManager.getQueue()).isEqualTo(expectedConfig); + assertThat(consumerManager.getConfig()).isEqualTo(expectedConfig); assertThat(consumerManager.getPartitions()).isEqualTo(expectedPartitions); }); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/queue/Queue.java b/common/data/src/main/java/org/thingsboard/server/common/data/queue/Queue.java index e3c6836aab..1c08072e61 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/queue/Queue.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/queue/Queue.java @@ -30,7 +30,7 @@ import org.thingsboard.server.common.data.validation.NoXss; import java.util.Optional; @Data -public class Queue extends BaseDataWithAdditionalInfo implements HasName, HasTenantId { +public class Queue extends BaseDataWithAdditionalInfo implements HasName, HasTenantId, QueueConfig { private TenantId tenantId; @NoXss @Length(fieldName = "name") diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/queue/QueueConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/queue/QueueConfig.java new file mode 100644 index 0000000000..383c613a38 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/queue/QueueConfig.java @@ -0,0 +1,24 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.queue; + +public interface QueueConfig { + + boolean isConsumerPerPartition(); + + int getPollInterval(); + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java index 3459096e08..88470296ed 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java @@ -39,8 +39,8 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.TopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; +import org.thingsboard.server.queue.discovery.TopicService; import org.thingsboard.server.queue.kafka.TbKafkaAdmin; import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService; import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; @@ -212,7 +212,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(topicService.buildTopicName(coreSettings.getTopic())); - consumerBuilder.clientId("monolith-core-consumer-" + serviceInfoProvider.getServiceId()); + consumerBuilder.clientId("monolith-core-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()); consumerBuilder.groupId(topicService.buildTopicName("monolith-core-consumer")); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(coreAdmin); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java index 1852693304..21078d7f94 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java @@ -38,8 +38,8 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.TopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; +import org.thingsboard.server.queue.discovery.TopicService; import org.thingsboard.server.queue.kafka.TbKafkaAdmin; import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService; import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; @@ -55,6 +55,7 @@ import org.thingsboard.server.queue.settings.TbQueueVersionControlSettings; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicLong; @Component @ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='tb-core'") @@ -81,6 +82,8 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { private final TbQueueAdmin fwUpdatesAdmin; private final TbQueueAdmin vcAdmin; + private final AtomicLong consumerCount = new AtomicLong(); + public KafkaTbCoreQueueFactory(TopicService topicService, TbKafkaSettings kafkaSettings, TbServiceInfoProvider serviceInfoProvider, @@ -169,7 +172,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(topicService.buildTopicName(coreSettings.getTopic())); - consumerBuilder.clientId("tb-core-consumer-" + serviceInfoProvider.getServiceId()); + consumerBuilder.clientId("tb-core-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()); consumerBuilder.groupId(topicService.buildTopicName("tb-core-node")); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(coreAdmin); diff --git a/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java b/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java index de097cfd6c..dc263b80e7 100644 --- a/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java +++ b/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java @@ -51,6 +51,11 @@ public class ThingsBoardThreadFactory implements ThreadFactory { Thread.currentThread().setName(name); } + public static void addThreadNamePrefix(String prefix) { + String name = Thread.currentThread().getName(); + name = prefix + name; + Thread.currentThread().setName(name); + } @Override public Thread newThread(Runnable r) { From 134db4249472f094523e35d2bb3efeb6f64be26f Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Tue, 19 Mar 2024 13:49:46 +0200 Subject: [PATCH 2/6] Configurable consumer per partition for tb-core queue --- .../server/service/queue/DefaultTbCoreConsumerService.java | 5 ++++- .../service/queue/processing/AbstractConsumerService.java | 3 +-- application/src/main/resources/thingsboard.yml | 2 ++ 3 files changed, 7 insertions(+), 3 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index c0ccaea9d0..474efc5ed7 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -126,6 +126,8 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService, CoreQueueConfig>builder() .queueKey(new QueueKey(ServiceType.TB_CORE)) - .config(CoreQueueConfig.of(true, (int) pollInterval)) + .config(CoreQueueConfig.of(consumerPerPartitionEnabled, (int) pollInterval)) .msgPackProcessor(this::processMsgs) .consumerCreator(config -> queueFactory.createToCoreMsgConsumer()) .consumerExecutor(consumersExecutor) @@ -764,6 +766,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService Date: Wed, 20 Mar 2024 12:49:40 +0200 Subject: [PATCH 3/6] Minor refactoring for consumer managers --- .../service/queue/DefaultTbCoreConsumerService.java | 12 ++++++++---- .../queue/DefaultTbRuleEngineConsumerService.java | 5 +---- .../queue/consumer/BasicQueueConsumerManager.java | 2 +- .../service/queue/consumer/QueueConsumerManager.java | 8 +++++--- .../queue/processing/AbstractConsumerService.java | 2 +- application/src/main/resources/thingsboard.yml | 2 +- .../TbRuleEngineQueueConsumerManagerTest.java | 4 ++-- 7 files changed, 19 insertions(+), 16 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index a2bed09151..e1bd18c7bf 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -127,8 +127,8 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService, CoreQueueConfig>builder() .queueKey(new QueueKey(ServiceType.TB_CORE)) - .config(CoreQueueConfig.of(consumerPerPartitionEnabled, (int) pollInterval)) + .config(CoreQueueConfig.of(consumerPerPartition, (int) pollInterval)) .msgPackProcessor(this::processMsgs) .consumerCreator(config -> queueFactory.createToCoreMsgConsumer()) .consumerExecutor(consumersExecutor) @@ -325,7 +325,9 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService log.debug("[{}] Timeout to process message: {}", id, msg.getValue())); + if (log.isDebugEnabled()) { + ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue())); + } ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue())); } consumer.commit(); @@ -452,6 +454,8 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService { } catch (Throwable e) { log.error("Failure in consumer loop", e); } + log.info("[{}] Consumer stopped", name); }); } @@ -94,7 +95,6 @@ public class BasicQueueConsumerManager { } } } - log.info("{} Consumer stopped", name); } public void stop() { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java b/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java index 04538289ea..acd28f25ba 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java @@ -176,8 +176,9 @@ public class QueueConsumerManager { doUpdate(partitions); // even if partitions number was changed, there can be no partition change event } } else { + log.trace("[{}] Silently applied new config, because consumer-per-partition not changed", queueKey); // 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 changes to other config values will be picked up by consumer on the fly, // and queue topic and name are immutable } } @@ -196,6 +197,7 @@ public class QueueConsumerManager { } catch (Throwable e) { log.error("Failure in consumer loop", e); } + log.info("[{}] Consumer stopped", consumerTask.getKey()); }); consumerTask.setTask(consumerLoop); } @@ -222,7 +224,6 @@ public class QueueConsumerManager { if (consumer.isStopped()) { consumer.unsubscribe(); } - log.info("{} Consumer stopped", queueKey); } protected void processMsgs(List msgs, TbQueueConsumer consumer, C config) throws Exception { @@ -236,6 +237,7 @@ public class QueueConsumerManager { } public void awaitStop() { + log.debug("[{}] Waiting for consumers to stop", queueKey); consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitCompletion); log.debug("[{}] Unsubscribed and stopped consumers", queueKey); } @@ -272,7 +274,7 @@ public class QueueConsumerManager { removedPartitions.forEach((tpi) -> consumers.remove(tpi).awaitCompletion()); addedPartitions.forEach((tpi) -> { - String key = queueKey + "-" + tpi.getPartition().orElse(-999999); + String key = queueKey + "-" + tpi.getPartition().orElse(-1); TbQueueConsumerTask consumer = new TbQueueConsumerTask<>(key, consumerCreator.apply(config)); consumers.put(tpi, consumer); consumer.subscribe(Set.of(tpi)); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index fa43eaa78d..79942cf6df 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.service.queue.processing; +import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.context.ApplicationEventPublisher; @@ -48,7 +49,6 @@ import org.thingsboard.server.service.queue.TbPackProcessingContext; import org.thingsboard.server.service.queue.consumer.BasicQueueConsumerManager; import org.thingsboard.server.service.security.auth.jwt.settings.JwtSettingsService; -import jakarta.annotation.PreDestroy; import java.util.List; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index d38c83a9ef..ca186ed0a5 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1596,7 +1596,7 @@ queue: # Timeout for processing a message pack by Core microservices pack-processing-timeout: "${TB_QUEUE_CORE_PACK_PROCESSING_TIMEOUT_MS:2000}" # Enable/disable a separate consumer per partition for Core queue - consumer-per-partition-enabled: "${TB_QUEUE_CORE_CONSUMER_PER_PARTITION_ENABLED:true}" + consumer-per-partition: "${TB_QUEUE_CORE_CONSUMER_PER_PARTITION:true}" ota: # Default topic name for OTA updates topic: "${TB_QUEUE_CORE_OTA_TOPIC:tb_ota_package}" diff --git a/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java b/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java index a00f45b8db..75df04333d 100644 --- a/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java +++ b/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java @@ -453,7 +453,7 @@ public class TbRuleEngineQueueConsumerManagerTest { int msgCount = totalConsumedMsgs.get(); await().atLeast(2, TimeUnit.SECONDS) // based on topicDeletionDelayInSec(5) = 5 - ( 3 seconds the code may execute starting consumerManager.delete() call) - .atMost(10, TimeUnit.SECONDS) + .atMost(20, TimeUnit.SECONDS) .untilAsserted(() -> { partitions.stream() .map(TopicPartitionInfo::getFullTopicName) @@ -492,7 +492,7 @@ public class TbRuleEngineQueueConsumerManagerTest { int msgCount = totalConsumedMsgs.get(); await().atLeast(2, TimeUnit.SECONDS) // based on topicDeletionDelayInSec(5) = 5 - ( 3 seconds the code may execute starting consumerManager.delete() call) - .atMost(10, TimeUnit.SECONDS) + .atMost(20, TimeUnit.SECONDS) .untilAsserted(() -> { partitions.stream() .map(TopicPartitionInfo::getFullTopicName) From 47cfebe2a7a8646df5ef225a0f2034129031efb6 Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Wed, 20 Mar 2024 13:25:30 +0200 Subject: [PATCH 4/6] Use QueueConsumerManager for other consumers --- .../queue/DefaultTbCoreConsumerService.java | 20 ++--- ...ger.java => MainQueueConsumerManager.java} | 15 ++-- .../processing/AbstractConsumerService.java | 8 +- .../TbRuleEngineQueueConsumerManager.java | 4 +- .../common/consumer/QueueConsumerManager.java | 14 ++- .../service/DefaultTransportService.java | 71 +++++++-------- .../common/util/ThingsBoardThreadFactory.java | 2 +- .../DefaultClusterVersionControlService.java | 87 ++++++++----------- 8 files changed, 96 insertions(+), 125 deletions(-) rename application/src/main/java/org/thingsboard/server/service/queue/consumer/{QueueConsumerManager.java => MainQueueConsumerManager.java} (95%) rename application/src/main/java/org/thingsboard/server/service/queue/consumer/BasicQueueConsumerManager.java => common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManager.java (87%) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index e1bd18c7bf..fdbbde3222 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -90,8 +90,8 @@ import org.thingsboard.server.service.notification.NotificationSchedulerService; import org.thingsboard.server.service.ota.OtaPackageStateService; import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; -import org.thingsboard.server.service.queue.consumer.BasicQueueConsumerManager; -import org.thingsboard.server.service.queue.consumer.QueueConsumerManager; +import org.thingsboard.server.queue.common.consumer.QueueConsumerManager; +import org.thingsboard.server.service.queue.consumer.MainQueueConsumerManager; import org.thingsboard.server.service.queue.processing.AbstractConsumerService; import org.thingsboard.server.service.queue.processing.IdMsgPair; import org.thingsboard.server.service.resource.TbImageService; @@ -151,9 +151,9 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService, CoreQueueConfig> mainConsumer; - private BasicQueueConsumerManager> usageStatsConsumer; - private BasicQueueConsumerManager> firmwareStatesConsumer; + private MainQueueConsumerManager, CoreQueueConfig> mainConsumer; + private QueueConsumerManager> usageStatsConsumer; + private QueueConsumerManager> firmwareStatesConsumer; private volatile ListeningExecutorService deviceActivityEventsExecutor; @@ -199,7 +199,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService, CoreQueueConfig>builder() + this.mainConsumer = MainQueueConsumerManager., CoreQueueConfig>builder() .queueKey(new QueueKey(ServiceType.TB_CORE)) .config(CoreQueueConfig.of(consumerPerPartition, (int) pollInterval)) .msgPackProcessor(this::processMsgs) @@ -208,19 +208,19 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService>builder() + this.usageStatsConsumer = QueueConsumerManager.>builder() .key("usage-stats") .name("TB Usage Stats") - .pollInterval(pollInterval) .msgPackProcessor(this::processUsageStatsMsg) + .pollInterval(pollInterval) .consumerCreator(queueFactory::createToUsageStatsServiceMsgConsumer) .consumerExecutor(consumersExecutor) .build(); - this.firmwareStatesConsumer = BasicQueueConsumerManager.>builder() + this.firmwareStatesConsumer = QueueConsumerManager.>builder() .key("firmware") .name("TB Ota Package States") - .pollInterval(pollInterval) .msgPackProcessor(this::processFirmwareMsgs) + .pollInterval(pollInterval) .consumerCreator(queueFactory::createToOtaPackageStateServiceMsgConsumer) .consumerExecutor(consumersExecutor) .build(); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java b/application/src/main/java/org/thingsboard/server/service/queue/consumer/MainQueueConsumerManager.java similarity index 95% rename from application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java rename to application/src/main/java/org/thingsboard/server/service/queue/consumer/MainQueueConsumerManager.java index acd28f25ba..6792422b57 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/consumer/MainQueueConsumerManager.java @@ -45,7 +45,7 @@ import java.util.function.Function; import java.util.stream.Collectors; @Slf4j -public class QueueConsumerManager { +public class MainQueueConsumerManager { protected final QueueKey queueKey; @Getter @@ -65,13 +65,12 @@ public class QueueConsumerManager { protected volatile boolean stopped; @Builder - public QueueConsumerManager(QueueKey queueKey, - C config, - MsgPackProcessor msgPackProcessor, - Function> consumerCreator, - ExecutorService consumerExecutor, - ScheduledExecutorService scheduler, - ExecutorService taskExecutor) { + public MainQueueConsumerManager(QueueKey queueKey, C config, + MsgPackProcessor msgPackProcessor, + Function> consumerCreator, + ExecutorService consumerExecutor, + ScheduledExecutorService scheduler, + ExecutorService taskExecutor) { this.queueKey = queueKey; this.config = config; this.msgPackProcessor = msgPackProcessor; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index 79942cf6df..3c769c911a 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -46,7 +46,7 @@ import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.service.queue.TbPackCallback; import org.thingsboard.server.service.queue.TbPackProcessingContext; -import org.thingsboard.server.service.queue.consumer.BasicQueueConsumerManager; +import org.thingsboard.server.queue.common.consumer.QueueConsumerManager; import org.thingsboard.server.service.security.auth.jwt.settings.JwtSettingsService; import java.util.List; @@ -73,7 +73,7 @@ public abstract class AbstractConsumerService> nfConsumer; + protected QueueConsumerManager> nfConsumer; protected ExecutorService consumersExecutor; protected ExecutorService mgmtExecutor; @@ -84,11 +84,11 @@ public abstract class AbstractConsumerService>builder() + this.nfConsumer = QueueConsumerManager.>builder() .key("notifications") .name("TB Notifications") - .pollInterval(getNotificationPollDuration()) .msgPackProcessor(this::processNotifications) + .pollInterval(getNotificationPollDuration()) .consumerCreator(this::createNotificationsConsumer) .consumerExecutor(consumersExecutor) .build(); 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 b93313b104..c2823d3c00 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 @@ -37,7 +37,7 @@ import org.thingsboard.server.queue.discovery.QueueKey; import org.thingsboard.server.service.queue.TbMsgPackCallback; import org.thingsboard.server.service.queue.TbMsgPackProcessingContext; import org.thingsboard.server.service.queue.TbRuleEngineConsumerStats; -import org.thingsboard.server.service.queue.consumer.QueueConsumerManager; +import org.thingsboard.server.service.queue.consumer.MainQueueConsumerManager; import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingDecision; import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingResult; import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategy; @@ -55,7 +55,7 @@ import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @Slf4j -public class TbRuleEngineQueueConsumerManager extends QueueConsumerManager, Queue> { +public class TbRuleEngineQueueConsumerManager extends MainQueueConsumerManager, Queue> { public static final String SUCCESSFUL_STATUS = "successful"; public static final String FAILED_STATUS = "failed"; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/consumer/BasicQueueConsumerManager.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManager.java similarity index 87% rename from application/src/main/java/org/thingsboard/server/service/queue/consumer/BasicQueueConsumerManager.java rename to common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManager.java index 4b4866c6b4..bf103bef60 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/consumer/BasicQueueConsumerManager.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManager.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.service.queue.consumer; +package org.thingsboard.server.queue.common.consumer; import lombok.Builder; import lombok.Getter; @@ -29,12 +29,12 @@ import java.util.concurrent.ExecutorService; import java.util.function.Supplier; @Slf4j -public class BasicQueueConsumerManager { +public class QueueConsumerManager { private final String key; private final String name; - private final long pollInterval; private final MsgPackProcessor msgPackProcessor; + private final long pollInterval; private final ExecutorService consumerExecutor; @Getter @@ -42,11 +42,9 @@ public class BasicQueueConsumerManager { private volatile boolean stopped; @Builder - public BasicQueueConsumerManager(String key, String name, - long pollInterval, - MsgPackProcessor msgPackProcessor, - Supplier> consumerCreator, - ExecutorService consumerExecutor) { + public QueueConsumerManager(String key, String name, MsgPackProcessor msgPackProcessor, + long pollInterval, Supplier> consumerCreator, + ExecutorService consumerExecutor) { this.key = key; this.name = name; this.pollInterval = pollInterval; diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index 8854ad3a7d..8f814ec052 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -20,6 +20,8 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import com.google.gson.Gson; import com.google.gson.JsonObject; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import org.springframework.beans.factory.annotation.Autowired; @@ -96,6 +98,7 @@ import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.AsyncCallbackTemplate; import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.common.consumer.QueueConsumerManager; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.discovery.TopicService; @@ -105,14 +108,12 @@ import org.thingsboard.server.queue.scheduler.SchedulerComponent; import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.TbTransportComponent; -import jakarta.annotation.PostConstruct; -import jakarta.annotation.PreDestroy; -import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Random; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; @@ -179,19 +180,17 @@ public class DefaultTransportService extends TransportActivityManager implements protected TbQueueRequestTemplate, TbProtoQueueMsg> transportApiRequestTemplate; protected TbQueueProducer> ruleEngineMsgProducer; protected TbQueueProducer> tbCoreMsgProducer; - protected TbQueueConsumer> transportNotificationsConsumer; + protected QueueConsumerManager> transportNotificationsConsumer; protected MessagesStats ruleEngineProducerStats; protected MessagesStats tbCoreProducerStats; protected MessagesStats transportApiStats; protected ExecutorService transportCallbackExecutor; - private ExecutorService mainConsumerExecutor; + private ExecutorService consumerExecutor; private final Map toServerRpcPendingMap = new ConcurrentHashMap<>(); - private volatile boolean stopped = false; - public DefaultTransportService(PartitionService partitionService, TbServiceInfoProvider serviceInfoProvider, TbTransportQueueFactory queueProvider, @@ -232,42 +231,34 @@ public class DefaultTransportService extends TransportActivityManager implements transportApiRequestTemplate.setMessagesStats(transportApiStats); ruleEngineMsgProducer = producerProvider.getRuleEngineMsgProducer(); tbCoreMsgProducer = producerProvider.getTbCoreMsgProducer(); - transportNotificationsConsumer = queueProvider.createTransportNotificationsConsumer(); - TopicPartitionInfo tpi = topicService.getNotificationsTopic(ServiceType.TB_TRANSPORT, serviceInfoProvider.getServiceId()); - transportNotificationsConsumer.subscribe(Collections.singleton(tpi)); transportApiRequestTemplate.init(); - mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("transport-consumer")); + consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("consumer")); + transportNotificationsConsumer = QueueConsumerManager.>builder() + .key("transport") + .name("TB Transport") + .msgPackProcessor(this::processNotificationMsgs) + .pollInterval(notificationsPollDuration) + .consumerCreator(queueProvider::createTransportNotificationsConsumer) + .consumerExecutor(consumerExecutor) + .build(); } @AfterStartUp(order = AfterStartUp.TRANSPORT_SERVICE) public void start() { - mainConsumerExecutor.execute(() -> { - while (!stopped) { - try { - List> records = transportNotificationsConsumer.poll(notificationsPollDuration); - if (records.size() == 0) { - continue; - } - records.forEach(record -> { - try { - processToTransportMsg(record.getValue()); - } catch (Throwable e) { - log.warn("Failed to process the notification.", e); - } - }); - transportNotificationsConsumer.commit(); - } catch (Exception e) { - if (!stopped) { - log.warn("Failed to obtain messages from queue.", e); - try { - Thread.sleep(notificationsPollDuration); - } catch (InterruptedException e2) { - log.trace("Failed to wait until the server has capacity to handle new requests", e2); - } - } - } + TopicPartitionInfo tpi = topicService.getNotificationsTopic(ServiceType.TB_TRANSPORT, serviceInfoProvider.getServiceId()); + transportNotificationsConsumer.subscribe(Set.of(tpi)); + transportNotificationsConsumer.launch(); + } + + private void processNotificationMsgs(List> msgs, TbQueueConsumer> consumer) { + msgs.forEach(msg -> { + try { + processToTransportMsg(msg.getValue()); + } catch (Throwable e) { + log.warn("Failed to process the notification.", e); } }); + consumer.commit(); } private void invalidateRateLimits() { @@ -276,16 +267,14 @@ public class DefaultTransportService extends TransportActivityManager implements @PreDestroy public void destroy() { - stopped = true; - if (transportNotificationsConsumer != null) { - transportNotificationsConsumer.unsubscribe(); + transportNotificationsConsumer.stop(); } if (transportCallbackExecutor != null) { transportCallbackExecutor.shutdownNow(); } - if (mainConsumerExecutor != null) { - mainConsumerExecutor.shutdownNow(); + if (consumerExecutor != null) { + consumerExecutor.shutdownNow(); } if (transportApiRequestTemplate != null) { transportApiRequestTemplate.stop(); diff --git a/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java b/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java index dc263b80e7..4d0876bf8e 100644 --- a/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java +++ b/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java @@ -53,7 +53,7 @@ public class ThingsBoardThreadFactory implements ThreadFactory { public static void addThreadNamePrefix(String prefix) { String name = Thread.currentThread().getName(); - name = prefix + name; + name = prefix + "-" + name; Thread.currentThread().setName(name); } diff --git a/common/version-control/src/main/java/org/thingsboard/server/service/sync/vc/DefaultClusterVersionControlService.java b/common/version-control/src/main/java/org/thingsboard/server/service/sync/vc/DefaultClusterVersionControlService.java index edf998dc8e..a5a7021d8d 100644 --- a/common/version-control/src/main/java/org/thingsboard/server/service/sync/vc/DefaultClusterVersionControlService.java +++ b/common/version-control/src/main/java/org/thingsboard/server/service/sync/vc/DefaultClusterVersionControlService.java @@ -21,14 +21,13 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.checkerframework.checker.nullness.qual.Nullable; import org.eclipse.jgit.errors.LargeObjectException; import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.context.event.ApplicationReadyEvent; -import org.springframework.context.event.EventListener; -import org.springframework.core.annotation.Order; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.data.EntityType; @@ -68,16 +67,16 @@ import org.thingsboard.server.gen.transport.TransportProtos.VersionedEntityInfoP import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.common.consumer.QueueConsumerManager; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.queue.discovery.TopicService; import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.provider.TbVersionControlQueueFactory; +import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.TbVersionControlComponent; -import jakarta.annotation.PostConstruct; -import jakarta.annotation.PreDestroy; import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; @@ -115,9 +114,8 @@ public class DefaultClusterVersionControlService extends TbApplicationEventListe private final Map pendingCommitMap = new HashMap<>(); private volatile ExecutorService consumerExecutor; - private volatile TbQueueConsumer> consumer; + private volatile QueueConsumerManager> consumer; private volatile TbQueueProducer> producer; - private volatile boolean stopped = false; @Value("${queue.vc.poll-interval:25}") private long pollDuration; @@ -134,20 +132,26 @@ public class DefaultClusterVersionControlService extends TbApplicationEventListe @PostConstruct public void init() { - consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("vc-consumer")); + consumerExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("consumer")); var threadFactory = ThingsBoardThreadFactory.forName("vc-io-thread"); for (int i = 0; i < ioPoolSize; i++) { ioThreads.add(MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor(threadFactory))); } producer = producerProvider.getTbCoreNotificationsMsgProducer(); - consumer = queueFactory.createToVersionControlMsgConsumer(); + consumer = QueueConsumerManager.>builder() + .key("vc") + .name("TB Version Control") + .msgPackProcessor(this::processMsgs) + .pollInterval(pollDuration) + .consumerCreator(queueFactory::createToVersionControlMsgConsumer) + .consumerExecutor(consumerExecutor) + .build(); } @PreDestroy public void stop() { - stopped = true; if (consumer != null) { - consumer.unsubscribe(); + consumer.stop(); } if (consumerExecutor != null) { consumerExecutor.shutdownNow(); @@ -179,48 +183,29 @@ public class DefaultClusterVersionControlService extends TbApplicationEventListe return ServiceType.TB_VC_EXECUTOR.equals(event.getServiceType()); } - @EventListener(ApplicationReadyEvent.class) - @Order(value = 2) - public void onApplicationEvent(ApplicationReadyEvent event) { - consumerExecutor.execute(() -> consumerLoop(consumer)); + @AfterStartUp(order = 2) + public void afterStartUp() { + consumer.launch(); } - void consumerLoop(TbQueueConsumer> consumer) { - while (!stopped && !consumer.isStopped()) { - List> futures = new ArrayList<>(); - try { - List> msgs = consumer.poll(pollDuration); - if (msgs.isEmpty()) { - continue; - } - for (TbProtoQueueMsg msgWrapper : msgs) { - ToVersionControlServiceMsg msg = msgWrapper.getValue(); - var ctx = new VersionControlRequestCtx(msg, msg.hasClearRepositoryRequest() ? null : ProtoUtils.fromProto(msg.getVcSettings())); - long startTs = System.currentTimeMillis(); - log.trace("[{}][{}] RECEIVED task: {}", ctx.getTenantId(), ctx.getRequestId(), msg); - int threadIdx = Math.abs(ctx.getTenantId().hashCode() % ioPoolSize); - ListenableFuture future = ioThreads.get(threadIdx).submit(() -> processMessage(ctx, msg)); - logTaskExecution(ctx, future, startTs); - futures.add(future); - } - try { - Futures.allAsList(futures).get(packProcessingTimeout, TimeUnit.MILLISECONDS); - } catch (TimeoutException e) { - log.info("Timeout for processing the version control tasks.", e); - } - consumer.commit(); - } catch (Exception e) { - if (!stopped) { - log.warn("Failed to obtain version control requests from queue.", e); - try { - Thread.sleep(pollDuration); - } catch (InterruptedException e2) { - log.trace("Failed to wait until the server has capacity to handle new version control messages", e2); - } - } - } + void processMsgs(List> msgs, TbQueueConsumer> consumer) throws Exception { + List> futures = new ArrayList<>(); + for (TbProtoQueueMsg msgWrapper : msgs) { + ToVersionControlServiceMsg msg = msgWrapper.getValue(); + var ctx = new VersionControlRequestCtx(msg, msg.hasClearRepositoryRequest() ? null : ProtoUtils.fromProto(msg.getVcSettings())); + long startTs = System.currentTimeMillis(); + log.trace("[{}][{}] RECEIVED task: {}", ctx.getTenantId(), ctx.getRequestId(), msg); + int threadIdx = Math.abs(ctx.getTenantId().hashCode() % ioPoolSize); + ListenableFuture future = ioThreads.get(threadIdx).submit(() -> processMessage(ctx, msg)); + logTaskExecution(ctx, future, startTs); + futures.add(future); + } + try { + Futures.allAsList(futures).get(packProcessingTimeout, TimeUnit.MILLISECONDS); + } catch (TimeoutException e) { + log.info("Timeout for processing the version control tasks.", e); } - log.info("TB Version Control request consumer stopped."); + consumer.commit(); } private Void processMessage(VersionControlRequestCtx ctx, ToVersionControlServiceMsg msg) { @@ -273,7 +258,7 @@ public class DefaultClusterVersionControlService extends TbApplicationEventListe var ids = vcService.listEntitiesAtVersion(ctx.getTenantId(), request.getVersionId(), path) .stream().skip(request.getOffset()).limit(request.getLimit()).collect(Collectors.toList()); if (!ids.isEmpty()) { - for (int i = 0; i < ids.size(); i++){ + for (int i = 0; i < ids.size(); i++) { VersionedEntityInfo info = ids.get(i); var data = vcService.getFileContentAtCommit(ctx.getTenantId(), getRelativePath(info.getExternalId().getEntityType(), info.getExternalId().getId().toString()), request.getVersionId()); From 15d189ead730245e29dcf07d959e34cfb8693e48 Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Tue, 7 May 2024 15:13:52 +0300 Subject: [PATCH 5/6] Refactoring for consumer manager --- .../HousekeeperReprocessingService.java | 2 +- .../housekeeper/HousekeeperService.java | 2 +- .../queue/DefaultTbCoreConsumerService.java | 4 +- .../queue/consumer/QueueConsumerManager.java | 109 ------------------ .../processing/AbstractConsumerService.java | 2 +- .../common/consumer/QueueConsumerManager.java | 13 ++- .../service/DefaultTransportService.java | 3 +- .../DefaultClusterVersionControlService.java | 3 +- 8 files changed, 15 insertions(+), 123 deletions(-) delete mode 100644 application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java diff --git a/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java b/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java index ae16f925a0..37a55c99d6 100644 --- a/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java @@ -27,11 +27,11 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperService import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.common.consumer.QueueConsumerManager; import org.thingsboard.server.queue.housekeeper.HousekeeperConfig; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.TbCoreComponent; -import org.thingsboard.server.service.queue.consumer.QueueConsumerManager; import javax.annotation.PreDestroy; import java.util.LinkedHashSet; diff --git a/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperService.java b/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperService.java index 06e38d0160..eea92eb75b 100644 --- a/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperService.java +++ b/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperService.java @@ -27,13 +27,13 @@ import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.common.consumer.QueueConsumerManager; import org.thingsboard.server.queue.housekeeper.HousekeeperConfig; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.housekeeper.processor.HousekeeperTaskProcessor; import org.thingsboard.server.service.housekeeper.stats.HousekeeperStatsService; -import org.thingsboard.server.service.queue.consumer.QueueConsumerManager; import javax.annotation.PreDestroy; import java.util.List; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index abfcd47a46..fbef0cfd29 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -210,20 +210,20 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService>builder() - .key("usage-stats") .name("TB Usage Stats") .msgPackProcessor(this::processUsageStatsMsg) .pollInterval(pollInterval) .consumerCreator(queueFactory::createToUsageStatsServiceMsgConsumer) .consumerExecutor(consumersExecutor) + .threadPrefix("usage-stats") .build(); this.firmwareStatesConsumer = QueueConsumerManager.>builder() - .key("firmware") .name("TB Ota Package States") .msgPackProcessor(this::processFirmwareMsgs) .pollInterval(pollInterval) .consumerCreator(queueFactory::createToOtaPackageStateServiceMsgConsumer) .consumerExecutor(consumersExecutor) + .threadPrefix("firmware") .build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java b/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java deleted file mode 100644 index e7b3a2f2b7..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java +++ /dev/null @@ -1,109 +0,0 @@ -/** - * Copyright © 2016-2024 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.service.queue.consumer; - -import lombok.Builder; -import lombok.Getter; -import lombok.extern.slf4j.Slf4j; -import org.thingsboard.common.util.ThingsBoardThreadFactory; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; -import org.thingsboard.server.queue.TbQueueConsumer; -import org.thingsboard.server.queue.TbQueueMsg; - -import java.util.List; -import java.util.Set; -import java.util.concurrent.ExecutorService; -import java.util.function.Supplier; - -@Slf4j -public class QueueConsumerManager { - - private final String name; - private final MsgPackProcessor msgPackProcessor; - private final long pollInterval; - private final ExecutorService consumerExecutor; - private final String threadPrefix; - - @Getter - private final TbQueueConsumer consumer; - private volatile boolean stopped; - - @Builder - public QueueConsumerManager(String name, MsgPackProcessor msgPackProcessor, - long pollInterval, Supplier> consumerCreator, - ExecutorService consumerExecutor, String threadPrefix) { - this.name = name; - this.pollInterval = pollInterval; - this.msgPackProcessor = msgPackProcessor; - this.consumerExecutor = consumerExecutor; - this.threadPrefix = threadPrefix; - this.consumer = consumerCreator.get(); - } - - public void subscribe() { - consumer.subscribe(); - } - - public void subscribe(Set partitions) { - consumer.subscribe(partitions); - } - - public void launch() { - log.info("[{}] Launching consumer", name); - consumerExecutor.submit(() -> { - if (threadPrefix != null) { - ThingsBoardThreadFactory.addThreadNamePrefix(threadPrefix); - } - try { - consumerLoop(consumer); - } catch (Throwable e) { - log.error("Failure in consumer loop", e); - } - log.info("[{}] Consumer stopped", name); - }); - } - - private void consumerLoop(TbQueueConsumer consumer) { - while (!stopped && !consumer.isStopped()) { - try { - List msgs = consumer.poll(pollInterval); - if (msgs.isEmpty()) { - continue; - } - msgPackProcessor.process(msgs, consumer); - } catch (Exception e) { - if (!consumer.isStopped()) { - log.warn("Failed to process messages from queue", e); - try { - Thread.sleep(pollInterval); - } catch (InterruptedException interruptedException) { - log.trace("Failed to wait until the server has capacity to handle new requests", interruptedException); - } - } - } - } - } - - public void stop() { - log.debug("[{}] Stopping consumer", name); - stopped = true; - consumer.unsubscribe(); - } - - public interface MsgPackProcessor { - void process(List msgs, TbQueueConsumer consumer) throws Exception; - } -} \ No newline at end of file diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index 2967143dd4..85227e50ab 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -85,12 +85,12 @@ public abstract class AbstractConsumerService>builder() - .key("notifications") .name("TB Notifications") .msgPackProcessor(this::processNotifications) .pollInterval(getNotificationPollDuration()) .consumerCreator(this::createNotificationsConsumer) .consumerExecutor(consumersExecutor) + .threadPrefix("notifications") .build(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManager.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManager.java index bf103bef60..90f4cf050c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManager.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManager.java @@ -31,25 +31,25 @@ import java.util.function.Supplier; @Slf4j public class QueueConsumerManager { - private final String key; private final String name; private final MsgPackProcessor msgPackProcessor; private final long pollInterval; private final ExecutorService consumerExecutor; + private final String threadPrefix; @Getter private final TbQueueConsumer consumer; private volatile boolean stopped; @Builder - public QueueConsumerManager(String key, String name, MsgPackProcessor msgPackProcessor, + public QueueConsumerManager(String name, MsgPackProcessor msgPackProcessor, long pollInterval, Supplier> consumerCreator, - ExecutorService consumerExecutor) { - this.key = key; + ExecutorService consumerExecutor, String threadPrefix) { this.name = name; this.pollInterval = pollInterval; this.msgPackProcessor = msgPackProcessor; this.consumerExecutor = consumerExecutor; + this.threadPrefix = threadPrefix; this.consumer = consumerCreator.get(); } @@ -64,7 +64,9 @@ public class QueueConsumerManager { public void launch() { log.info("[{}] Launching consumer", name); consumerExecutor.submit(() -> { - ThingsBoardThreadFactory.addThreadNamePrefix(key); + if (threadPrefix != null) { + ThingsBoardThreadFactory.addThreadNamePrefix(threadPrefix); + } try { consumerLoop(consumer); } catch (Throwable e) { @@ -104,4 +106,5 @@ public class QueueConsumerManager { public interface MsgPackProcessor { void process(List msgs, TbQueueConsumer consumer) throws Exception; } + } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index d208fab29e..8b7b88c261 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -232,9 +232,8 @@ public class DefaultTransportService extends TransportActivityManager implements ruleEngineMsgProducer = producerProvider.getRuleEngineMsgProducer(); tbCoreMsgProducer = producerProvider.getTbCoreMsgProducer(); transportApiRequestTemplate.init(); - consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("consumer")); + consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("transport-consumer")); transportNotificationsConsumer = QueueConsumerManager.>builder() - .key("transport") .name("TB Transport") .msgPackProcessor(this::processNotificationMsgs) .pollInterval(notificationsPollDuration) diff --git a/common/version-control/src/main/java/org/thingsboard/server/service/sync/vc/DefaultClusterVersionControlService.java b/common/version-control/src/main/java/org/thingsboard/server/service/sync/vc/DefaultClusterVersionControlService.java index a5a7021d8d..34171c19cd 100644 --- a/common/version-control/src/main/java/org/thingsboard/server/service/sync/vc/DefaultClusterVersionControlService.java +++ b/common/version-control/src/main/java/org/thingsboard/server/service/sync/vc/DefaultClusterVersionControlService.java @@ -132,14 +132,13 @@ public class DefaultClusterVersionControlService extends TbApplicationEventListe @PostConstruct public void init() { - consumerExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("consumer")); + consumerExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("vc-consumer")); var threadFactory = ThingsBoardThreadFactory.forName("vc-io-thread"); for (int i = 0; i < ioPoolSize; i++) { ioThreads.add(MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor(threadFactory))); } producer = producerProvider.getTbCoreNotificationsMsgProducer(); consumer = QueueConsumerManager.>builder() - .key("vc") .name("TB Version Control") .msgPackProcessor(this::processMsgs) .pollInterval(pollDuration) From b16d8484e91798ec77a6e73f1726e5159de160b8 Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Wed, 8 May 2024 14:50:11 +0300 Subject: [PATCH 6/6] Minor refactoring for notifications consumers --- .../queue/processing/AbstractConsumerService.java | 2 +- .../server/common/msg/queue/ServiceType.java | 14 +++++++++++++- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index 85227e50ab..a264f912cb 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -85,7 +85,7 @@ public abstract class AbstractConsumerService>builder() - .name("TB Notifications") + .name(getServiceType().getLabel() + " Notifications") .msgPackProcessor(this::processNotifications) .pollInterval(getNotificationPollDuration()) .consumerCreator(this::createNotificationsConsumer) diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/ServiceType.java b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/ServiceType.java index c823ea4d73..f3a0e47d09 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/ServiceType.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/ServiceType.java @@ -15,11 +15,23 @@ */ package org.thingsboard.server.common.msg.queue; +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +@RequiredArgsConstructor +@Getter public enum ServiceType { - TB_CORE, TB_RULE_ENGINE, TB_TRANSPORT, JS_EXECUTOR, TB_VC_EXECUTOR; + TB_CORE("TB Core"), + TB_RULE_ENGINE("TB Rule Engine"), + TB_TRANSPORT("TB Transport"), + JS_EXECUTOR("JS Executor"), + TB_VC_EXECUTOR("TB VC Executor"); + + private final String label; public static ServiceType of(String serviceType) { return ServiceType.valueOf(serviceType.replace("-", "_").toUpperCase()); } + }