From 1044aabce371d1ee7e65870fa70c86055f66dda6 Mon Sep 17 00:00:00 2001 From: Viacheslav Klimov Date: Fri, 12 Dec 2025 16:13:32 +0200 Subject: [PATCH] Fix Kafka topics cache --- .../src/main/resources/thingsboard.yml | 2 ++ .../server/queue/kafka/KafkaAdmin.java | 29 +++++++++---------- .../server/queue/kafka/TbKafkaSettings.java | 1 + edqs/src/main/resources/edqs.yml | 2 ++ .../src/main/resources/tb-vc-executor.yml | 2 ++ .../src/main/resources/tb-coap-transport.yml | 2 ++ .../src/main/resources/tb-http-transport.yml | 2 ++ .../src/main/resources/tb-lwm2m-transport.yml | 2 ++ .../src/main/resources/tb-mqtt-transport.yml | 2 ++ .../src/main/resources/tb-snmp-transport.yml | 2 ++ 10 files changed, 30 insertions(+), 16 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index dfdd3a407c..cda7f5782a 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1720,6 +1720,8 @@ queue: print-interval-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_MIN_PRINT_INTERVAL_MS:60000}" # Time to wait for the stats-loading requests to Kafka to finish kafka-response-timeout-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_RESPONSE_TIMEOUT_MS:1000}" + # Topics cache TTL in milliseconds. 5 minutes by default + topics_cache_ttl_ms: "${TB_QUEUE_KAFKA_TOPICS_CACHE_TTL_MS:300000}" partitions: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256 transport_api: diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/KafkaAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/KafkaAdmin.java index 6261e81497..3e2c64de69 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/KafkaAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/KafkaAdmin.java @@ -58,15 +58,12 @@ public class KafkaAdmin { private final TbKafkaSettings settings; - @Value("${queue.kafka.request.timeout.ms:30000}") - private int requestTimeoutMs; - @Value("${queue.kafka.topics_cache_ttl_ms:300000}") // 5 minutes by default - private int topicsCacheTtlMs; - private final LazyInitializer adminClient; private final CachedValue> topics; - public KafkaAdmin(@Lazy TbKafkaSettings settings) { + public KafkaAdmin(@Lazy TbKafkaSettings settings, + @Value("${queue.kafka.topics_cache_ttl_ms:300000}") + int topicsCacheTtlMs) { this.settings = settings; this.adminClient = LazyInitializer.builder() .setInitializer(() -> AdminClient.create(settings.toAdminProps())) @@ -91,7 +88,7 @@ public class KafkaAdmin { NewTopic newTopic = new NewTopic(topic, partitions, settings.getReplicationFactor()).configs(properties); try { - getClient().createTopics(List.of(newTopic)).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); + getClient().createTopics(List.of(newTopic)).all().get(settings.getRequestTimeoutMs(), TimeUnit.MILLISECONDS); topics.add(topic); } catch (ExecutionException ee) { log.trace("Failed to create topic {} with properties {}", topic, properties, ee); @@ -110,7 +107,7 @@ public class KafkaAdmin { public void deleteTopic(String topic) { log.debug("Deleting topic {}", topic); try { - getClient().deleteTopics(List.of(topic)).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); + getClient().deleteTopics(List.of(topic)).all().get(settings.getRequestTimeoutMs(), TimeUnit.MILLISECONDS); } catch (Exception e) { log.error("Failed to delete kafka topic [{}].", topic, e); } @@ -122,7 +119,7 @@ public class KafkaAdmin { public Set listTopics() { try { - Set topics = getClient().listTopics().names().get(requestTimeoutMs, TimeUnit.MILLISECONDS); + Set topics = getClient().listTopics().names().get(settings.getRequestTimeoutMs(), TimeUnit.MILLISECONDS); log.trace("Listed topics: {}", topics); return topics; } catch (Exception e) { @@ -150,7 +147,7 @@ public class KafkaAdmin { .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest())); Map endOffsets = - getClient().listOffsets(latestOffsetsSpec).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); + getClient().listOffsets(latestOffsetsSpec).all().get(settings.getRequestTimeoutMs(), TimeUnit.MILLISECONDS); return committedOffsets.entrySet().stream() .mapToLong(entry -> { @@ -169,7 +166,7 @@ public class KafkaAdmin { @SneakyThrows public Map getConsumerGroupOffsets(String groupId) { - return getClient().listConsumerGroupOffsets(groupId).partitionsToOffsetAndMetadata().get(requestTimeoutMs, TimeUnit.MILLISECONDS); + return getClient().listConsumerGroupOffsets(groupId).partitionsToOffsetAndMetadata().get(settings.getRequestTimeoutMs(), TimeUnit.MILLISECONDS); } /** @@ -212,7 +209,7 @@ public class KafkaAdmin { } else { log.info("[{}] SHOULD alter topic offset [{}] less than old node group offset [{}]", tp, existingOffset.offset(), om.offset()); } - getClient().alterConsumerGroupOffsets(newGroupId, Map.of(tp, om)).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); + getClient().alterConsumerGroupOffsets(newGroupId, Map.of(tp, om)).all().get(settings.getRequestTimeoutMs(), TimeUnit.MILLISECONDS); log.info("[{}] altered new consumer groupId {}", tp, newGroupId); break; } @@ -229,7 +226,7 @@ public class KafkaAdmin { return true; } - List allPartitions = getClient().describeTopics(existingTopics).allTopicNames().get(requestTimeoutMs, TimeUnit.MILLISECONDS) + List allPartitions = getClient().describeTopics(existingTopics).allTopicNames().get(settings.getRequestTimeoutMs(), TimeUnit.MILLISECONDS) .entrySet().stream() .flatMap(entry -> { String topic = entry.getKey(); @@ -239,9 +236,9 @@ public class KafkaAdmin { .toList(); Map beginningOffsets = getClient().listOffsets(allPartitions.stream() - .collect(Collectors.toMap(partition -> partition, partition -> OffsetSpec.earliest()))).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); + .collect(Collectors.toMap(partition -> partition, partition -> OffsetSpec.earliest()))).all().get(settings.getRequestTimeoutMs(), TimeUnit.MILLISECONDS); Map endOffsets = getClient().listOffsets(allPartitions.stream() - .collect(Collectors.toMap(partition -> partition, partition -> OffsetSpec.latest()))).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); + .collect(Collectors.toMap(partition -> partition, partition -> OffsetSpec.latest()))).all().get(settings.getRequestTimeoutMs(), TimeUnit.MILLISECONDS); for (TopicPartition partition : allPartitions) { long beginningOffset = beginningOffsets.get(partition).offset(); @@ -261,7 +258,7 @@ public class KafkaAdmin { public void deleteConsumerGroup(String consumerGroupId) { try { - getClient().deleteConsumerGroups(List.of(consumerGroupId)).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); + getClient().deleteConsumerGroups(List.of(consumerGroupId)).all().get(settings.getRequestTimeoutMs(), TimeUnit.MILLISECONDS); } catch (Exception e) { log.warn("Failed to delete consumer group {}", consumerGroupId, e); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java index 11736f68cf..3dbcff1863 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java @@ -112,6 +112,7 @@ public class TbKafkaSettings { @Value("${queue.kafka.fetch_max_bytes:134217728}") private int fetchMaxBytes; + @Getter @Value("${queue.kafka.request.timeout.ms:30000}") private int requestTimeoutMs; diff --git a/edqs/src/main/resources/edqs.yml b/edqs/src/main/resources/edqs.yml index f6b611d5a9..9bdee541a3 100644 --- a/edqs/src/main/resources/edqs.yml +++ b/edqs/src/main/resources/edqs.yml @@ -177,6 +177,8 @@ queue: print-interval-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_MIN_PRINT_INTERVAL_MS:60000}" # Time to wait for the stats-loading requests to Kafka to finish kafka-response-timeout-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_RESPONSE_TIMEOUT_MS:1000}" + # Topics cache TTL in milliseconds. 5 minutes by default + topics_cache_ttl_ms: "${TB_QUEUE_KAFKA_TOPICS_CACHE_TTL_MS:300000}" partitions: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256 diff --git a/msa/vc-executor/src/main/resources/tb-vc-executor.yml b/msa/vc-executor/src/main/resources/tb-vc-executor.yml index 7d1166e512..ea89fb7d23 100644 --- a/msa/vc-executor/src/main/resources/tb-vc-executor.yml +++ b/msa/vc-executor/src/main/resources/tb-vc-executor.yml @@ -151,6 +151,8 @@ queue: print-interval-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_MIN_PRINT_INTERVAL_MS:60000}" # Time to wait for the stats-loading requests to Kafka to finis kafka-response-timeout-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_RESPONSE_TIMEOUT_MS:1000}" + # Topics cache TTL in milliseconds. 5 minutes by default + topics_cache_ttl_ms: "${TB_QUEUE_KAFKA_TOPICS_CACHE_TTL_MS:300000}" partitions: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256 core: diff --git a/transport/coap/src/main/resources/tb-coap-transport.yml b/transport/coap/src/main/resources/tb-coap-transport.yml index f40a09c753..02ec421366 100644 --- a/transport/coap/src/main/resources/tb-coap-transport.yml +++ b/transport/coap/src/main/resources/tb-coap-transport.yml @@ -332,6 +332,8 @@ queue: notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" # Kafka properties for Housekeeper tasks topic housekeeper: "${TB_QUEUE_KAFKA_HOUSEKEEPER_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:10;min.insync.replicas:1}" + # Topics cache TTL in milliseconds. 5 minutes by default + topics_cache_ttl_ms: "${TB_QUEUE_KAFKA_TOPICS_CACHE_TTL_MS:300000}" partitions: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256 transport_api: diff --git a/transport/http/src/main/resources/tb-http-transport.yml b/transport/http/src/main/resources/tb-http-transport.yml index 587894d5ce..7fe35a57d5 100644 --- a/transport/http/src/main/resources/tb-http-transport.yml +++ b/transport/http/src/main/resources/tb-http-transport.yml @@ -281,6 +281,8 @@ queue: notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" # Kafka properties for Housekeeper tasks topic housekeeper: "${TB_QUEUE_KAFKA_HOUSEKEEPER_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:10;min.insync.replicas:1}" + # Topics cache TTL in milliseconds. 5 minutes by default + topics_cache_ttl_ms: "${TB_QUEUE_KAFKA_TOPICS_CACHE_TTL_MS:300000}" partitions: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256 transport_api: diff --git a/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml b/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml index 0895bfa676..b1718a2099 100644 --- a/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml +++ b/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml @@ -382,6 +382,8 @@ queue: notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" # Kafka properties for Housekeeper tasks topic housekeeper: "${TB_QUEUE_KAFKA_HOUSEKEEPER_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:10;min.insync.replicas:1}" + # Topics cache TTL in milliseconds. 5 minutes by default + topics_cache_ttl_ms: "${TB_QUEUE_KAFKA_TOPICS_CACHE_TTL_MS:300000}" partitions: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256 transport_api: diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index fae10cc892..86e5dc5a5b 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -315,6 +315,8 @@ queue: notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" # Kafka properties for Housekeeper tasks topic housekeeper: "${TB_QUEUE_KAFKA_HOUSEKEEPER_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:10;min.insync.replicas:1}" + # Topics cache TTL in milliseconds. 5 minutes by default + topics_cache_ttl_ms: "${TB_QUEUE_KAFKA_TOPICS_CACHE_TTL_MS:300000}" partitions: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256 transport_api: diff --git a/transport/snmp/src/main/resources/tb-snmp-transport.yml b/transport/snmp/src/main/resources/tb-snmp-transport.yml index 567654cce4..9dfb9f0b41 100644 --- a/transport/snmp/src/main/resources/tb-snmp-transport.yml +++ b/transport/snmp/src/main/resources/tb-snmp-transport.yml @@ -270,6 +270,8 @@ queue: print-interval-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_MIN_PRINT_INTERVAL_MS:60000}" # Time to wait for the stats-loading requests to Kafka to finis kafka-response-timeout-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_RESPONSE_TIMEOUT_MS:1000}" + # Topics cache TTL in milliseconds. 5 minutes by default + topics_cache_ttl_ms: "${TB_QUEUE_KAFKA_TOPICS_CACHE_TTL_MS:300000}" partitions: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256 transport_api: