tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Fix Kafka topics cache

pull/14587/head
Viacheslav Klimov 8 months ago
parent
8becf055db
commit
1044aabce3
10 changed files with 30 additions and 16 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      application/src/main/resources/thingsboard.yml
  2. 29
      common/queue/src/main/java/org/thingsboard/server/queue/kafka/KafkaAdmin.java
  3. 1
      common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java
  4. 2
      edqs/src/main/resources/edqs.yml
  5. 2
      msa/vc-executor/src/main/resources/tb-vc-executor.yml
  6. 2
      transport/coap/src/main/resources/tb-coap-transport.yml
  7. 2
      transport/http/src/main/resources/tb-http-transport.yml
  8. 2
      transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml
  9. 2
      transport/mqtt/src/main/resources/tb-mqtt-transport.yml
  10. 2
      transport/snmp/src/main/resources/tb-snmp-transport.yml

2
application/src/main/resources/thingsboard.yml
View File

@ -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:

29
common/queue/src/main/java/org/thingsboard/server/queue/kafka/KafkaAdmin.java
View File

@ -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> adminClient;
private final CachedValue<Set<String>> 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.<AdminClient>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<String> listTopics() {
try {
Set<String> topics = getClient().listTopics().names().get(requestTimeoutMs, TimeUnit.MILLISECONDS);
Set<String> 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<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> 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<TopicPartition, OffsetAndMetadata> 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<TopicPartition> allPartitions = getClient().describeTopics(existingTopics).allTopicNames().get(requestTimeoutMs, TimeUnit.MILLISECONDS)
List<TopicPartition> 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<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> 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<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> 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);
}

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

@ -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;

2
edqs/src/main/resources/edqs.yml
View File

@ -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

2
msa/vc-executor/src/main/resources/tb-vc-executor.yml
View File

@ -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:

2
transport/coap/src/main/resources/tb-coap-transport.yml
View File

@ -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:

2
transport/http/src/main/resources/tb-http-transport.yml
View File

@ -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:

2
transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml
View File

@ -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:

2
transport/mqtt/src/main/resources/tb-mqtt-transport.yml
View File

@ -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:

2
transport/snmp/src/main/resources/tb-snmp-transport.yml
View File

@ -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:

Write Preview
Loading…
Cancel
Save