dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

TbKafkaSettings - use startswith strategy for consumer props in case direct .get return null

Browse Source
This commit is contained in:
Volodymyr Babak 2025-04-09 17:56:47 +03:00
parent c443b5e8c6
commit 68289044c5
3 changed files with 19 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -62,7 +62,7 @@ public class KafkaEdgeTopicsCleanUpService extends AbstractCleanUpService {
@Value("${sql.ttl.edge_events.edge_events_ttl:2628000}") @Value("${sql.ttl.edge_events.edge_events_ttl:2628000}")
private long ttlSeconds; private long ttlSeconds;
@Value("${queue.edge.event_notifications_topic:tb_edge_event.notifications}") @Value("${queue.edge.event-notifications-topic:tb_edge_event.notifications}")
private String tbEdgeEventNotificationsTopic; private String tbEdgeEventNotificationsTopic;
public KafkaEdgeTopicsCleanUpService(PartitionService partitionService, EdgeService edgeService, public KafkaEdgeTopicsCleanUpService(PartitionService partitionService, EdgeService edgeService,

12
common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicService.java
View File

@ -32,22 +32,22 @@ public class TopicService {
@Value("${queue.prefix:}") @Value("${queue.prefix:}")
private String prefix; private String prefix;
@Value("${queue.core.notifications_topic:tb_core.notifications}") @Value("${queue.core.notifications-topic:tb_core.notifications}")
private String tbCoreNotificationsTopic; private String tbCoreNotificationsTopic;
@Value("${queue.rule-engine.notifications_topic:tb_rule_engine.notifications}") @Value("${queue.rule-engine.notifications-topic:tb_rule_engine.notifications}")
private String tbRuleEngineNotificationsTopic; private String tbRuleEngineNotificationsTopic;
@Value("${queue.transport.notifications_topics:tb_transport.notifications}") @Value("${queue.transport.notifications-topic:tb_transport.notifications}")
private String tbTransportNotificationsTopic; private String tbTransportNotificationsTopic;
@Value("${queue.edge.notifications_topic:tb_edge.notifications}") @Value("${queue.edge.notifications-topic:tb_edge.notifications}")
private String tbEdgeNotificationsTopic; private String tbEdgeNotificationsTopic;
@Value("${queue.edge.event_notifications_topic:tb_edge_event.notifications}") @Value("${queue.edge.event-notifications-topic:tb_edge_event.notifications}")
private String tbEdgeEventNotificationsTopic; private String tbEdgeEventNotificationsTopic;
@Value("${queue.calculated-fields.notifications_topic:calculated_field.notifications}") @Value("${queue.calculated-fields.notifications-topic:calculated_field.notifications}")
private String tbCalculatedFieldNotificationsTopic; private String tbCalculatedFieldNotificationsTopic;
private final ConcurrentMap<String, TopicPartitionInfo> tbCoreNotificationTopics = new ConcurrentHashMap<>(); private final ConcurrentMap<String, TopicPartitionInfo> tbCoreNotificationTopics = new ConcurrentHashMap<>();

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

@ -50,8 +50,6 @@ import java.util.Properties;
@Component @Component
public class TbKafkaSettings { public class TbKafkaSettings {
private static final List<String> DYNAMIC_TOPICS = List.of("tb_edge.notifications", "tb_edge_event.notifications");
@Value("${queue.kafka.bootstrap.servers}") @Value("${queue.kafka.bootstrap.servers}")
private String servers; private String servers;
@ -163,18 +161,20 @@ public class TbKafkaSettings {
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
consumerPropertiesPerTopic
.getOrDefault(topic, Collections.emptyList())
.forEach(kv -> props.put(kv.getKey(), kv.getValue()));
if (topic != null) { if (topic != null) {
DYNAMIC_TOPICS.stream() List<TbProperty> properties = consumerPropertiesPerTopic.get(topic);
.filter(topic::startsWith) if (properties == null) {
.findFirst() for (Map.Entry<String, List<TbProperty>> entry : consumerPropertiesPerTopic.entrySet()) {
.ifPresent(prefix -> consumerPropertiesPerTopic.getOrDefault(prefix, Collections.emptyList()) if (topic.startsWith(entry.getKey())) {
.forEach(kv -> props.put(kv.getKey(), kv.getValue()))); properties = entry.getValue();
break;
}
}
}
if (properties != null) {
properties.forEach(kv -> props.put(kv.getKey(), kv.getValue()));
}
} }
return props; return props;
} }