Browse Source

Improvement to buildTopicName for KafkaEdgeTopicsCleanUpService

pull/11924/head
Andrii Landiak 2 years ago
parent
commit
4b7e49a8f4
  1. 5
      application/src/main/java/org/thingsboard/server/service/ttl/KafkaEdgeTopicsCleanUpService.java

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

@ -64,9 +64,6 @@ public class KafkaEdgeTopicsCleanUpService extends AbstractCleanUpService {
private final TbKafkaSettings kafkaSettings; private final TbKafkaSettings kafkaSettings;
private final TbKafkaTopicConfigs kafkaTopicConfigs; private final TbKafkaTopicConfigs kafkaTopicConfigs;
@Value("${queue.prefix:}")
private String prefix;
@Value("${sql.ttl.edge_events.edge_events_ttl:2628000}") @Value("${sql.ttl.edge_events.edge_events_ttl:2628000}")
private long ttlSeconds; private long ttlSeconds;
@ -95,7 +92,7 @@ public class KafkaEdgeTopicsCleanUpService extends AbstractCleanUpService {
return; return;
} }
String edgeTopicPrefix = prefix.isBlank() ? EDGE_EVENT_TOPIC_NAME : prefix + "." + EDGE_EVENT_TOPIC_NAME; String edgeTopicPrefix = topicService.buildTopicName(EDGE_EVENT_TOPIC_NAME);
List<String> matchingTopics = topics.stream().filter(topic -> topic.startsWith(edgeTopicPrefix)).toList(); List<String> matchingTopics = topics.stream().filter(topic -> topic.startsWith(edgeTopicPrefix)).toList();
if (matchingTopics.isEmpty()) { if (matchingTopics.isEmpty()) {
log.info("No matching topics found with prefix [{}]. Skipping cleanup.", edgeTopicPrefix); log.info("No matching topics found with prefix [{}]. Skipping cleanup.", edgeTopicPrefix);

Loading…
Cancel
Save