Add poll properties for edge-event kafka poll

application/src/main/java/org/thingsboard/server/service/edge/EdgeEventSourcingListener.java

@@ -20,8 +20,10 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
 import jakarta.annotation.PostConstruct;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.event.TransactionalEventListener;
 import org.thingsboard.common.util.JacksonUtil;
@@ -84,8 +86,12 @@ public class EdgeEventSourcingListener {
     private final TenantService tenantService;
     private final EdgeSynchronizationManager edgeSynchronizationManager;
 
-    private final TbKafkaSettings kafkaSettings;
-    private final TbKafkaTopicConfigs kafkaTopicConfigs;
+    @Autowired(required = false)
+    @Lazy
+    private TbKafkaSettings kafkaSettings;
+    @Autowired(required = false)
+    @Lazy
+    private TbKafkaTopicConfigs kafkaTopicConfigs;
 
     @Value("#{'${queue.type:null}' == 'kafka'}")
     private boolean isKafkaSupported;

application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeEventService.java

@@ -20,7 +20,6 @@ import com.google.common.util.concurrent.ListenableFuture;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
-import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
 import org.thingsboard.server.cache.limits.RateLimitService;
@@ -65,6 +64,7 @@ public class KafkaEdgeEventService implements EdgeEventService {
             throw new TbRateLimitsException(EntityType.EDGE);
         }
         edgeEventValidator.validate(edgeEvent, EdgeEvent::getTenantId);
+
         TopicPartitionInfo tpi = topicService.getEdgeEventNotificationsTopic(edgeEvent.getTenantId(), edgeEvent.getEdgeId());
         ToEdgeEventNotificationMsg msg = ToEdgeEventNotificationMsg.newBuilder().setEdgeEventMsg(ProtoUtils.toProto(edgeEvent)).build();
         producerProvider.getTbEdgeEventsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), msg), null);
@@ -78,6 +78,7 @@ public class KafkaEdgeEventService implements EdgeEventService {
     }
 
     @Override
-    public void cleanupEvents(long ttl) {}
+    public void cleanupEvents(long ttl) {
+    }
 
 }

application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java

@@ -89,7 +89,7 @@ public class KafkaEdgeGrpcSession extends AbstractEdgeGrpcSession<KafkaEdgeGrpcS
                         if (Boolean.TRUE.equals(isInterrupted)) {
                             log.debug("[{}][{}][{}] Send downlink messages task was interrupted", tenantId, edge.getId(), sessionId);
                         } else {
-                            consumer.commit();
+                            consumerExecutor.submit(consumer::commit);
                         }
                     }
                     @Override

application/src/main/java/org/thingsboard/server/service/ttl/KafkaEdgeTopicsCleanUpService.java

@@ -66,7 +66,11 @@ 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.checking_interval})}", fixedDelayString = "${sql.ttl.edge_events.checking_interval}")
+    //    @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}"
+    )
     public void cleanUp() {
         executorService.submit(() -> {
             PageDataIterable<TenantId> tenants = new PageDataIterable<>(tenantService::findTenantsIds, 10_000);
@@ -101,6 +105,12 @@ 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

@@ -1517,6 +1517,11 @@ queue:
     #      tb_rule_engine.sq:
     #        - key: max.poll.records
     #          value: "${TB_QUEUE_KAFKA_SQ_MAX_POLL_RECORDS:1024}"
+      tb_edge_event.notifications:
+        # Example of specific consumer properties value per topic for edge event
+        - key: max.poll.records
+          # Example of specific consumer properties value per topic for edge event
+          value: "${TB_QUEUE_KAFKA_EDGE_EVENT_MAX_POLL_INTERVAL_MS:50}"
       tb_housekeeper:
         # Consumer properties for Housekeeper tasks topic
         - key: max.poll.records
@@ -1556,7 +1561,7 @@ queue:
       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:2592000000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}"
+      edge-event: "${TB_QUEUE_KAFKA_EDGE_EVENT_TOPIC_PROPERTIES:retention.ms:60000;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/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeEventService.java

@@ -33,4 +33,5 @@ public interface EdgeEventService {
      * @param ttl the ttl for edge events in seconds
      */
     void cleanupEvents(long ttl);
+
 }

common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java

@@ -17,7 +17,10 @@ package org.thingsboard.server.queue.kafka;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.ListOffsetsResult;
 import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.TopicExistsException;
@@ -174,8 +177,20 @@ public class TbKafkaAdmin implements TbQueueAdmin {
 
     public boolean isTopicEmpty(String topic) {
         try {
-            Map<TopicPartition, OffsetAndMetadata> offsets = settings.getAdminClient().listConsumerGroupOffsets("__consumer_offsets").partitionsToOffsetAndMetadata().get();
-            return offsets.entrySet().stream().noneMatch(entry -> entry.getKey().topic().equals(topic));
+            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 =
+                    settings.getAdminClient().listOffsets(Collections.singletonMap(topicPartition, OffsetSpec.latest())).all().get();
+
+            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;
         } catch (InterruptedException | ExecutionException e) {
             log.error("Failed to check if topic [{}] is empty.", topic, e);
             return false;

common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java

@@ -50,6 +50,8 @@ import java.util.Properties;
 @Component
 public class TbKafkaSettings {
 
+    private static final List<String> BASIC_TOPIC_PREFIXES = List.of("tb_edge_event.notifications");
+
     @Value("${queue.kafka.bootstrap.servers}")
     private String servers;
 
@@ -164,9 +166,22 @@ public class TbKafkaSettings {
         consumerPropertiesPerTopic
                 .getOrDefault(topic, Collections.emptyList())
                 .forEach(kv -> props.put(kv.getKey(), kv.getValue()));
+
+        applyBaseTopicProperties(props, topic);
+
         return props;
     }
 
+    private void applyBaseTopicProperties(Properties props, String topic) {
+        if (topic != null) {
+            BASIC_TOPIC_PREFIXES.stream()
+                    .filter(topic::startsWith)
+                    .findFirst()
+                    .ifPresent(prefix -> consumerPropertiesPerTopic.getOrDefault(prefix, Collections.emptyList())
+                            .forEach(kv -> props.put(kv.getKey(), kv.getValue())));
+        }
+    }
+
     public Properties toProducerProps() {
         Properties props = toProps();
         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);

common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java

@@ -132,7 +132,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
         this.housekeeperAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getHousekeeperConfigs());
         this.housekeeperReprocessingAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getHousekeeperReprocessingConfigs());
         this.edgeAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeConfigs());
-        this.edgeEventAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeConfigs());
+        this.edgeEventAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeEventConfigs());
     }
 
     @Override

common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java

@@ -132,7 +132,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
         this.housekeeperAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getHousekeeperConfigs());
         this.housekeeperReprocessingAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getHousekeeperReprocessingConfigs());
         this.edgeAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeConfigs());
-        this.edgeEventAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeConfigs());
+        this.edgeEventAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeEventConfigs());
     }
 
     @Override

common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java

@@ -110,7 +110,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory {
         this.fwUpdatesAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getFwUpdatesConfigs());
         this.housekeeperAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getHousekeeperConfigs());
         this.edgeAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeConfigs());
-        this.edgeEventAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeConfigs());
+        this.edgeEventAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeEventConfigs());
     }
 
     @Override

dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java

@@ -251,7 +251,6 @@ public class EdgeServiceImpl extends AbstractCachedEntityService<EdgeCacheKey, E
         }
         edgeDao.removeById(tenantId, edgeId.getId());
 
-
         publishEvictEvent(new EdgeCacheEvictEvent(edge.getTenantId(), edge.getName(), null));
         eventPublisher.publishEvent(DeleteEntityEvent.builder().tenantId(tenantId).entityId(edgeId).build());
     }

dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java → dao/src/main/java/org/thingsboard/server/dao/edge/PostgresEdgeEventService.java

@@ -46,7 +46,7 @@ import java.util.concurrent.Executors;
 @Service
 @RequiredArgsConstructor
 @ConditionalOnExpression("'${queue.type:null}'!='kafka'")
-public class BaseEdgeEventService implements EdgeEventService {
+public class PostgresEdgeEventService implements EdgeEventService {
 
     private final EdgeEventDao edgeEventDao;
     private final RateLimitService rateLimitService;
@@ -82,8 +82,11 @@ public class BaseEdgeEventService implements EdgeEventService {
         Futures.addCallback(saveFuture, new FutureCallback<>() {
             @Override
             public void onSuccess(Void result) {
-                eventPublisher.publishEvent(SaveEntityEvent.builder().tenantId(edgeEvent.getTenantId())
-                        .entity(edgeEvent).entityId(edgeEvent.getEdgeId()).build());
+                eventPublisher.publishEvent(SaveEntityEvent.builder()
+                        .tenantId(edgeEvent.getTenantId())
+                        .entityId(edgeEvent.getEdgeId())
+                        .entity(edgeEvent)
+                        .build());
             }
 
             @Override

dao/src/main/java/org/thingsboard/server/dao/model/sql/EdgeEventEntity.java

@@ -129,4 +129,5 @@ public class EdgeEventEntity extends BaseSqlEntity<EdgeEvent> implements BaseEnt
     private static long getTs(UUID uuid) {
         return (uuid.timestamp() - EPOCH_DIFF) / 10000;
     }
+
 }