From 721854c8793e5f89c71cfdf05fb2c2320ed6a62b Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Tue, 20 Feb 2024 18:50:29 +0200 Subject: [PATCH] Housekeeper: reprocessing improvements --- .../DefaultHousekeeperService.java | 20 ++--- .../HousekeeperReprocessingService.java | 52 ++++++++---- .../EventsDeletionTaskProcessor.java | 1 + common/proto/src/main/proto/queue.proto | 83 ++++++++++--------- .../provider/AwsSqsMonolithQueueFactory.java | 4 +- .../provider/AwsSqsTbCoreQueueFactory.java | 4 +- .../InMemoryMonolithQueueFactory.java | 8 +- .../provider/KafkaMonolithQueueFactory.java | 14 ++-- .../provider/KafkaTbCoreQueueFactory.java | 14 ++-- .../provider/PubSubMonolithQueueFactory.java | 4 +- .../provider/PubSubTbCoreQueueFactory.java | 4 +- .../RabbitMqMonolithQueueFactory.java | 4 +- .../provider/RabbitMqTbCoreQueueFactory.java | 4 +- .../ServiceBusMonolithQueueFactory.java | 4 +- .../ServiceBusTbCoreQueueFactory.java | 4 +- .../queue/provider/TbCoreQueueFactory.java | 5 +- .../provider/TbCoreQueueProducerProvider.java | 9 +- .../provider/TbQueueProducerProvider.java | 2 +- .../TbRuleEngineProducerProvider.java | 2 +- .../TbTransportQueueProducerProvider.java | 2 +- .../TbVersionControlProducerProvider.java | 2 +- .../queue/settings/TbQueueCoreSettings.java | 2 +- 22 files changed, 134 insertions(+), 114 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/housekeeper/DefaultHousekeeperService.java b/application/src/main/java/org/thingsboard/server/service/housekeeper/DefaultHousekeeperService.java index 10d53f3500..644e3cc2c7 100644 --- a/application/src/main/java/org/thingsboard/server/service/housekeeper/DefaultHousekeeperService.java +++ b/application/src/main/java/org/thingsboard/server/service/housekeeper/DefaultHousekeeperService.java @@ -99,7 +99,7 @@ public class DefaultHousekeeperService implements HousekeeperService { processTask(msg); } catch (Throwable e) { log.error("Unexpected error during message processing [{}]", msg, e); - reprocessingService.submitForReprocessing(msg); + reprocessingService.submitForReprocessing(msg, e); } } consumer.commit(); @@ -131,23 +131,23 @@ public class DefaultHousekeeperService implements HousekeeperService { return null; }); future.get(taskProcessingTimeout, TimeUnit.SECONDS); - } catch (ExecutionException executionException) { - Throwable error = executionException.getCause(); + } catch (Throwable e) { + Throwable error = e; + if (e instanceof ExecutionException) { + error = error.getCause(); + } else if (e instanceof TimeoutException) { + error = new TimeoutException("Timeout after " + taskProcessingTimeout + " seconds"); + } log.error("[{}][{}][{}] {} task processing failed, submitting for reprocessing (attempt {}): {}", task.getTenantId(), task.getEntityId().getEntityType(), task.getEntityId(), task.getTaskType(), msg.getTask().getAttempt(), task, error); - reprocessingService.submitForReprocessing(queueMsg); - } catch (TimeoutException timeoutException) { - log.error("[{}][{}][{}] {} task processing timeout after {} seconds, submitting for reprocessing (attempt {}): {}", - task.getTenantId(), task.getEntityId().getEntityType(), task.getEntityId(), - task.getTaskType(), taskProcessingTimeout, msg.getTask().getAttempt(), task); - reprocessingService.submitForReprocessing(queueMsg); // fixme: we should schedule such task for later (separate queue?) + reprocessingService.submitForReprocessing(queueMsg, error); } } @Override public void submitTask(HousekeeperTask task) { - submitTask(UUID.randomUUID(), task); + submitTask(task.getEntityId().getId(), task); } @Override 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 9d7114476b..2d2c3242f9 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 @@ -16,18 +16,23 @@ package org.thingsboard.server.service.housekeeper; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.exception.ExceptionUtils; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Lazy; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.common.data.StringUtils; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.gen.transport.TransportProtos.HousekeeperTaskProto; 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.discovery.PartitionService; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; -import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.TbCoreComponent; import javax.annotation.PreDestroy; @@ -42,35 +47,46 @@ import java.util.concurrent.TimeUnit; public class HousekeeperReprocessingService { private final DefaultHousekeeperService housekeeperService; + private final PartitionService partitionService; + private final TbCoreQueueFactory queueFactory; private final TbQueueProducerProvider producerProvider; - private final TbQueueConsumer> consumer; + private final ExecutorService consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("housekeeper-reprocessing-consumer")); + private static final int startDelay = 15; // fixme - to 5 minutes + private static final int reprocessingDelay = 30; // seconds + private static final int maxReprocessingAttempts = 5; @Value("${queue.core.housekeeper.poll-interval-ms:500}") private int pollInterval; - private final long startTs = System.currentTimeMillis(); // fixme: some other tb-core might start earlier and submit for reprocessing private boolean stopped; // todo: stats public HousekeeperReprocessingService(@Lazy DefaultHousekeeperService housekeeperService, - TbCoreQueueFactory queueFactory, + PartitionService partitionService, TbCoreQueueFactory queueFactory, TbQueueProducerProvider producerProvider) { this.housekeeperService = housekeeperService; - this.consumer = queueFactory.createHousekeeperDelayedMsgConsumer(); + this.partitionService = partitionService; + this.queueFactory = queueFactory; this.producerProvider = producerProvider; } - @AfterStartUp(order = AfterStartUp.REGULAR_SERVICE) - public void afterStartUp() { + @Scheduled(initialDelay = startDelay, fixedDelay = reprocessingDelay, timeUnit = TimeUnit.SECONDS) + public void startReprocessing() { + if (!partitionService.isMyPartition(ServiceType.TB_CORE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID)) { + return; + } + + var consumer = queueFactory.createHousekeeperReprocessingMsgConsumer(); consumer.subscribe(); consumerExecutor.submit(() -> { - while (!stopped && !consumer.isStopped()) { + log.info("Starting Housekeeper tasks reprocessing"); + long startTs = System.currentTimeMillis(); + while (!stopped) { try { List> msgs = consumer.poll(pollInterval); if (msgs.isEmpty() || msgs.stream().anyMatch(msg -> msg.getValue().getTask().getTs() >= startTs)) { // msg batch size should be 1. otherwise some tasks won't be reprocessed immediately - stop(); - return; + break; } for (TbProtoQueueMsg msg : msgs) { @@ -78,7 +94,7 @@ public class HousekeeperReprocessingService { housekeeperService.processTask(msg); } catch (Exception e) { log.error("Unexpected error during message reprocessing [{}]", msg, e); - submitForReprocessing(msg); + submitForReprocessing(msg, e); } } consumer.commit(); @@ -93,31 +109,33 @@ public class HousekeeperReprocessingService { } } } + consumer.unsubscribe(); + log.info("Stopped Housekeeper tasks reprocessing"); }); - log.info("Started Housekeeper tasks reprocessing"); } - public void submitForReprocessing(TbProtoQueueMsg queueMsg) { + // todo: dead letter queue if attempts count exceeds the configured maximum + public void submitForReprocessing(TbProtoQueueMsg queueMsg, Throwable error) { ToHousekeeperServiceMsg msg = queueMsg.getValue(); HousekeeperTaskProto task = msg.getTask(); int attempt = task.getAttempt() + 1; msg = msg.toBuilder() .setTask(task.toBuilder() .setAttempt(attempt) - .setTs(System.currentTimeMillis() + TimeUnit.HOURS.toMillis(1)) // so that it is not reprocessed by anyone in the nearest hour + .addErrors(StringUtils.truncate(ExceptionUtils.getStackTrace(error), 1024)) + .setTs(System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(reprocessingDelay)) .build()) .build(); - var producer = producerProvider.getHousekeeperDelayedMsgProducer(); + log.trace("Submitting for reprocessing: {}", msg); + var producer = producerProvider.getHousekeeperReprocessingMsgProducer(); TopicPartitionInfo tpi = TopicPartitionInfo.builder().topic(producer.getDefaultTopic()).build(); producer.send(tpi, new TbProtoQueueMsg<>(queueMsg.getKey(), msg), null); } @PreDestroy private void stop() { - log.info("Stopped Housekeeper tasks reprocessing"); stopped = true; - consumer.unsubscribe(); consumerExecutor.shutdownNow(); } diff --git a/application/src/main/java/org/thingsboard/server/service/housekeeper/processor/EventsDeletionTaskProcessor.java b/application/src/main/java/org/thingsboard/server/service/housekeeper/processor/EventsDeletionTaskProcessor.java index 56edf01637..b95d7fc1e3 100644 --- a/application/src/main/java/org/thingsboard/server/service/housekeeper/processor/EventsDeletionTaskProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/housekeeper/processor/EventsDeletionTaskProcessor.java @@ -29,6 +29,7 @@ public class EventsDeletionTaskProcessor implements HousekeeperTaskProcessor> createHousekeeperDelayedMsgProducer() { + public TbQueueProducer> createHousekeeperReprocessingMsgProducer() { return null; } @Override - public TbQueueConsumer> createHousekeeperDelayedMsgConsumer() { + public TbQueueConsumer> createHousekeeperReprocessingMsgConsumer() { return null; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java index a1be25f29d..1a4f92fcd3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java @@ -217,12 +217,12 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { } @Override - public TbQueueProducer> createHousekeeperDelayedMsgProducer() { + public TbQueueProducer> createHousekeeperReprocessingMsgProducer() { return null; } @Override - public TbQueueConsumer> createHousekeeperDelayedMsgConsumer() { + public TbQueueConsumer> createHousekeeperReprocessingMsgConsumer() { return null; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java index 1ee8b404a5..310bc75fe4 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java @@ -171,13 +171,13 @@ public class InMemoryMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE } @Override - public TbQueueProducer> createHousekeeperDelayedMsgProducer() { - return new InMemoryTbQueueProducer<>(storage, topicService.buildTopicName(coreSettings.getHousekeeperDelayedTopic())); + public TbQueueProducer> createHousekeeperReprocessingMsgProducer() { + return new InMemoryTbQueueProducer<>(storage, topicService.buildTopicName(coreSettings.getHousekeeperReprocessingTopic())); } @Override - public TbQueueConsumer> createHousekeeperDelayedMsgConsumer() { - return new InMemoryTbQueueConsumer<>(storage, topicService.buildTopicName(coreSettings.getHousekeeperDelayedTopic())); + public TbQueueConsumer> createHousekeeperReprocessingMsgConsumer() { + return new InMemoryTbQueueConsumer<>(storage, topicService.buildTopicName(coreSettings.getHousekeeperReprocessingTopic())); } @Scheduled(fixedRateString = "${queue.in_memory.stats.print-interval-ms:60000}") 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 5ee9b6f53b..b4c2220ac1 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 @@ -373,22 +373,22 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi } @Override - public TbQueueProducer> createHousekeeperDelayedMsgProducer() { + public TbQueueProducer> createHousekeeperReprocessingMsgProducer() { return TbKafkaProducerTemplate.>builder() .settings(kafkaSettings) - .clientId("monolith-housekeeper-delayed-producer-" + serviceInfoProvider.getServiceId()) - .defaultTopic(topicService.buildTopicName(coreSettings.getHousekeeperDelayedTopic())) + .clientId("monolith-housekeeper-reprocessing-producer-" + serviceInfoProvider.getServiceId()) + .defaultTopic(topicService.buildTopicName(coreSettings.getHousekeeperReprocessingTopic())) .admin(housekeeperAdmin) .build(); } @Override - public TbQueueConsumer> createHousekeeperDelayedMsgConsumer() { + public TbQueueConsumer> createHousekeeperReprocessingMsgConsumer() { return TbKafkaConsumerTemplate.>builder() .settings(kafkaSettings) - .topic(topicService.buildTopicName(coreSettings.getHousekeeperDelayedTopic())) - .clientId("monolith-housekeeper-delayed-consumer-" + serviceInfoProvider.getServiceId()) - .groupId(topicService.buildTopicName("monolith-housekeeper-delayed-consumer")) + .topic(topicService.buildTopicName(coreSettings.getHousekeeperReprocessingTopic())) + .clientId("monolith-housekeeper-reprocessing-consumer-" + serviceInfoProvider.getServiceId()) + .groupId(topicService.buildTopicName("monolith-housekeeper-reprocessing-consumer")) .decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToHousekeeperServiceMsg.parseFrom(msg.getData()), msg.getHeaders())) .admin(housekeeperAdmin) .statsService(consumerStatsService) 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 c01ba32f70..635eb3e300 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 @@ -330,22 +330,22 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { } @Override - public TbQueueProducer> createHousekeeperDelayedMsgProducer() { + public TbQueueProducer> createHousekeeperReprocessingMsgProducer() { return TbKafkaProducerTemplate.>builder() .settings(kafkaSettings) - .clientId("tb-core-housekeeper-delayed-producer-" + serviceInfoProvider.getServiceId()) - .defaultTopic(topicService.buildTopicName(coreSettings.getHousekeeperDelayedTopic())) + .clientId("tb-core-housekeeper-reprocessing-producer-" + serviceInfoProvider.getServiceId()) + .defaultTopic(topicService.buildTopicName(coreSettings.getHousekeeperReprocessingTopic())) .admin(housekeeperAdmin) .build(); } @Override - public TbQueueConsumer> createHousekeeperDelayedMsgConsumer() { + public TbQueueConsumer> createHousekeeperReprocessingMsgConsumer() { return TbKafkaConsumerTemplate.>builder() .settings(kafkaSettings) - .topic(topicService.buildTopicName(coreSettings.getHousekeeperDelayedTopic())) - .clientId("tb-core-housekeeper-delayed-consumer-" + serviceInfoProvider.getServiceId()) - .groupId(topicService.buildTopicName("tb-core-housekeeper-delayed-consumer")) + .topic(topicService.buildTopicName(coreSettings.getHousekeeperReprocessingTopic())) + .clientId("tb-core-housekeeper-reprocessing-consumer-" + serviceInfoProvider.getServiceId()) + .groupId(topicService.buildTopicName("tb-core-housekeeper-reprocessing-consumer")) .decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToHousekeeperServiceMsg.parseFrom(msg.getData()), msg.getHeaders())) .admin(housekeeperAdmin) .statsService(consumerStatsService) diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java index a63a9fc16b..e6a23b8068 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java @@ -239,12 +239,12 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng } @Override - public TbQueueProducer> createHousekeeperDelayedMsgProducer() { + public TbQueueProducer> createHousekeeperReprocessingMsgProducer() { return null; } @Override - public TbQueueConsumer> createHousekeeperDelayedMsgConsumer() { + public TbQueueConsumer> createHousekeeperReprocessingMsgConsumer() { return null; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java index 5f1f282e39..1b52a2ba42 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java @@ -210,12 +210,12 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { } @Override - public TbQueueProducer> createHousekeeperDelayedMsgProducer() { + public TbQueueProducer> createHousekeeperReprocessingMsgProducer() { return null; } @Override - public TbQueueConsumer> createHousekeeperDelayedMsgConsumer() { + public TbQueueConsumer> createHousekeeperReprocessingMsgConsumer() { return null; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java index 802b08773d..fa89956b64 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java @@ -236,12 +236,12 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE } @Override - public TbQueueProducer> createHousekeeperDelayedMsgProducer() { + public TbQueueProducer> createHousekeeperReprocessingMsgProducer() { return null; } @Override - public TbQueueConsumer> createHousekeeperDelayedMsgConsumer() { + public TbQueueConsumer> createHousekeeperReprocessingMsgConsumer() { return null; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java index ceb6e03ca5..b1f3de672f 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java @@ -209,12 +209,12 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { } @Override - public TbQueueProducer> createHousekeeperDelayedMsgProducer() { + public TbQueueProducer> createHousekeeperReprocessingMsgProducer() { return null; } @Override - public TbQueueConsumer> createHousekeeperDelayedMsgConsumer() { + public TbQueueConsumer> createHousekeeperReprocessingMsgConsumer() { return null; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java index 1e9c1cadaf..5f4be98ca0 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java @@ -235,12 +235,12 @@ public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRul } @Override - public TbQueueProducer> createHousekeeperDelayedMsgProducer() { + public TbQueueProducer> createHousekeeperReprocessingMsgProducer() { return null; } @Override - public TbQueueConsumer> createHousekeeperDelayedMsgConsumer() { + public TbQueueConsumer> createHousekeeperReprocessingMsgConsumer() { return null; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java index e39df45b82..579fba272c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java @@ -210,12 +210,12 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { } @Override - public TbQueueProducer> createHousekeeperDelayedMsgProducer() { + public TbQueueProducer> createHousekeeperReprocessingMsgProducer() { return null; } @Override - public TbQueueConsumer> createHousekeeperDelayedMsgConsumer() { + public TbQueueConsumer> createHousekeeperReprocessingMsgConsumer() { return null; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java index 279a469415..866eb931f2 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java @@ -16,7 +16,6 @@ package org.thingsboard.server.queue.provider; import org.thingsboard.server.gen.js.JsInvokeProtos; -import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; @@ -137,8 +136,8 @@ public interface TbCoreQueueFactory extends TbUsageStatsClientQueueFactory { TbQueueConsumer> createHousekeeperMsgConsumer(); - TbQueueProducer> createHousekeeperDelayedMsgProducer(); + TbQueueProducer> createHousekeeperReprocessingMsgProducer(); - TbQueueConsumer> createHousekeeperDelayedMsgConsumer(); + TbQueueConsumer> createHousekeeperReprocessingMsgConsumer(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueProducerProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueProducerProvider.java index 4fa22cb7e3..e6bc8f0484 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueProducerProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueProducerProvider.java @@ -16,7 +16,6 @@ package org.thingsboard.server.queue.provider; import org.springframework.stereotype.Service; -import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; @@ -44,7 +43,7 @@ public class TbCoreQueueProducerProvider implements TbQueueProducerProvider { private TbQueueProducer> toUsageStats; private TbQueueProducer> toVersionControl; private TbQueueProducer> toHousekeeper; - private TbQueueProducer> toHousekeeperDelayed; + private TbQueueProducer> toHousekeeperReprocessing; public TbCoreQueueProducerProvider(TbCoreQueueFactory tbQueueProvider) { this.tbQueueProvider = tbQueueProvider; @@ -60,7 +59,7 @@ public class TbCoreQueueProducerProvider implements TbQueueProducerProvider { this.toUsageStats = tbQueueProvider.createToUsageStatsServiceMsgProducer(); this.toVersionControl = tbQueueProvider.createVersionControlMsgProducer(); this.toHousekeeper = tbQueueProvider.createHousekeeperMsgProducer(); - this.toHousekeeperDelayed = tbQueueProvider.createHousekeeperDelayedMsgProducer(); + this.toHousekeeperReprocessing = tbQueueProvider.createHousekeeperReprocessingMsgProducer(); } @Override @@ -104,8 +103,8 @@ public class TbCoreQueueProducerProvider implements TbQueueProducerProvider { } @Override - public TbQueueProducer> getHousekeeperDelayedMsgProducer() { - return toHousekeeperDelayed; + public TbQueueProducer> getHousekeeperReprocessingMsgProducer() { + return toHousekeeperReprocessing; } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java index 5cdf390f47..ff56d2e8be 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java @@ -82,6 +82,6 @@ public interface TbQueueProducerProvider { TbQueueProducer> getHousekeeperMsgProducer(); - TbQueueProducer> getHousekeeperDelayedMsgProducer(); + TbQueueProducer> getHousekeeperReprocessingMsgProducer(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineProducerProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineProducerProvider.java index c5ddf57fc0..ee6700906c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineProducerProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineProducerProvider.java @@ -96,7 +96,7 @@ public class TbRuleEngineProducerProvider implements TbQueueProducerProvider { } @Override - public TbQueueProducer> getHousekeeperDelayedMsgProducer() { + public TbQueueProducer> getHousekeeperReprocessingMsgProducer() { throw new RuntimeException("Not Implemented! Should not be used by Rule Engine!"); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java index e8dd7431a6..869c588fc7 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java @@ -92,7 +92,7 @@ public class TbTransportQueueProducerProvider implements TbQueueProducerProvider } @Override - public TbQueueProducer> getHousekeeperDelayedMsgProducer() { + public TbQueueProducer> getHousekeeperReprocessingMsgProducer() { throw new RuntimeException("Not Implemented! Should not be used by Transport!"); } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbVersionControlProducerProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbVersionControlProducerProvider.java index 7ddd2109ba..3f4bd209e4 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbVersionControlProducerProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbVersionControlProducerProvider.java @@ -88,7 +88,7 @@ public class TbVersionControlProducerProvider implements TbQueueProducerProvider } @Override - public TbQueueProducer> getHousekeeperDelayedMsgProducer() { + public TbQueueProducer> getHousekeeperReprocessingMsgProducer() { throw new RuntimeException("Not Implemented! Should not be used by Version Control Service!"); } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueCoreSettings.java b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueCoreSettings.java index 0b66c2fa75..2aa01193b0 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueCoreSettings.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueCoreSettings.java @@ -38,7 +38,7 @@ public class TbQueueCoreSettings { private String housekeeperTopic; @Value("${queue.core.housekeeper.reprocessing-topic:tb_housekeeper.reprocessing}") - private String housekeeperDelayedTopic; + private String housekeeperReprocessingTopic; @Value("${queue.core.partitions}") private int partitions;