Fix tests

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;

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<TenantId> 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)));
-//            }
         }
     }
 

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}"

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<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> beginningOffsets =
                     settings.getAdminClient().listOffsets(Collections.singletonMap(topicPartition, OffsetSpec.earliest())).all().get();
             Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> 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;