From b981281205092194c65d7f39a09271a1d1123a38 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Fri, 1 Nov 2024 11:37:11 +0200 Subject: [PATCH] Fix tests --- .../service/edge/EdgeEventSourcingListener.java | 1 - .../service/ttl/KafkaEdgeTopicsCleanUpService.java | 12 +----------- application/src/main/resources/thingsboard.yml | 3 +-- .../thingsboard/server/queue/kafka/TbKafkaAdmin.java | 4 +--- 4 files changed, 3 insertions(+), 17 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/EdgeEventSourcingListener.java b/application/src/main/java/org/thingsboard/server/service/edge/EdgeEventSourcingListener.java index 2d3f40b078..3653eb3a40 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/EdgeEventSourcingListener.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/EdgeEventSourcingListener.java @@ -77,7 +77,6 @@ import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; @Slf4j @Component @RequiredArgsConstructor -@ConditionalOnExpression("${edges.enabled:true}") public class EdgeEventSourcingListener { private final TopicService topicService; diff --git a/application/src/main/java/org/thingsboard/server/service/ttl/KafkaEdgeTopicsCleanUpService.java b/application/src/main/java/org/thingsboard/server/service/ttl/KafkaEdgeTopicsCleanUpService.java index 35a2578444..20e2ab6758 100644 --- a/application/src/main/java/org/thingsboard/server/service/ttl/KafkaEdgeTopicsCleanUpService.java +++ b/application/src/main/java/org/thingsboard/server/service/ttl/KafkaEdgeTopicsCleanUpService.java @@ -66,11 +66,7 @@ public class KafkaEdgeTopicsCleanUpService { private final ExecutorService executorService = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("kafka-edge-topic-cleanup")); - // @Scheduled(initialDelayString = "#{T(org.apache.commons.lang3.RandomUtils).nextLong(0, ${sql.ttl.edge_events.execution_interval_ms})}", fixedDelayString = "${sql.ttl.edge_events.execution_interval_ms}") - @Scheduled( - initialDelay = 60000, // 1 minute delay after startup - fixedDelayString = "${sql.ttl.edge_events.execution_interval_ms}" - ) + @Scheduled(initialDelayString = "#{T(org.apache.commons.lang3.RandomUtils).nextLong(0, ${sql.ttl.edge_events.execution_interval_ms})}", fixedDelayString = "${sql.ttl.edge_events.execution_interval_ms}") public void cleanUp() { executorService.submit(() -> { PageDataIterable tenants = new PageDataIterable<>(tenantService::findTenantsIds, 10_000); @@ -105,12 +101,6 @@ public class KafkaEdgeTopicsCleanUpService { } } } -// String topic = topicService.buildEdgeEventNotificationsTopicPartitionInfo(tenantId, edge.getId()).getTopic(); -// TbKafkaAdmin kafkaAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeEventConfigs()); -// if (kafkaAdmin.isTopicEmpty(topic)) { -// kafkaAdmin.deleteTopic(topic); -// log.info("Removed outdated topic for tenant {} and edge {} older than {}", tenantId, edge.getName(), Date.from(Instant.ofEpochMilli(currentTimeMillis - ONE_MONTH_MILLIS))); -// } } } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 448d5964f0..bf050d268a 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1560,8 +1560,7 @@ queue: # Kafka properties for Edge topic edge: "${TB_QUEUE_KAFKA_EDGE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" # Kafka properties for Edge event topic - # TODO : discuss and adjust properties - edge-event: "${TB_QUEUE_KAFKA_EDGE_EVENT_TOPIC_PROPERTIES:retention.ms:60000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" + edge-event: "${TB_QUEUE_KAFKA_EDGE_EVENT_TOPIC_PROPERTIES:retention.ms:2592000000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" consumer-stats: # Prints lag between consumer group offset and last messages offset in Kafka topics enabled: "${TB_QUEUE_KAFKA_CONSUMER_STATS_ENABLED:true}" diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java index 86fa0dad3c..c209c574e2 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java @@ -180,7 +180,6 @@ public class TbKafkaAdmin implements TbQueueAdmin { TopicDescription topicDescription = settings.getAdminClient().describeTopics(Collections.singletonList(topic)).topicNameValues().get(topic).get(); TopicPartition topicPartition = new TopicPartition(topic, topicDescription.partitions().get(0).partition()); - // Get the earliest and latest offsets Map beginningOffsets = settings.getAdminClient().listOffsets(Collections.singletonMap(topicPartition, OffsetSpec.earliest())).all().get(); Map endOffsets = @@ -189,8 +188,7 @@ public class TbKafkaAdmin implements TbQueueAdmin { long beginningOffset = beginningOffsets.get(topicPartition).offset(); long endOffset = endOffsets.get(topicPartition).offset(); - // If beginning and end offsets are both 0, the topic is empty - return beginningOffset == 0 && endOffset == 0; + return beginningOffset == endOffset; } catch (InterruptedException | ExecutionException e) { log.error("Failed to check if topic [{}] is empty.", topic, e); return false;