From 2658de715a9b1672bbf9abefd12b1479208a26cc Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Fri, 23 Apr 2021 10:56:49 +0300 Subject: [PATCH] implemented consumerPropertiesPerTopic --- application/src/main/resources/thingsboard.yml | 4 ++++ .../queue/kafka/TbKafkaConsumerStatsService.java | 2 +- .../server/queue/kafka/TbKafkaConsumerTemplate.java | 2 +- .../server/queue/kafka/TbKafkaSettings.java | 11 ++++++++++- 4 files changed, 16 insertions(+), 3 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index ae20c8c97c..29cd38d805 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -743,6 +743,10 @@ queue: sasl.mechanism: "${TB_QUEUE_KAFKA_CONFLUENT_SASL_MECHANISM:PLAIN}" sasl.config: "${TB_QUEUE_KAFKA_CONFLUENT_SASL_JAAS_CONFIG:org.apache.kafka.common.security.plain.PlainLoginModule required username=\"CLUSTER_API_KEY\" password=\"CLUSTER_API_SECRET\";}" security.protocol: "${TB_QUEUE_KAFKA_CONFLUENT_SECURITY_PROTOCOL:SASL_SSL}" + consumerPropertiesPerTopic: + tb_firmware: + - key: max.poll.records + value: 10 other: topic-properties: rule-engine: "${TB_QUEUE_KAFKA_RE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java index 0ccf9ba42c..b29944c1e3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java @@ -69,7 +69,7 @@ public class TbKafkaConsumerStatsService { this.adminClient = AdminClient.create(kafkaSettings.toAdminProps()); this.statsPrintScheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("kafka-consumer-stats")); - Properties consumerProps = kafkaSettings.toConsumerProps(); + Properties consumerProps = kafkaSettings.toConsumerProps(null); consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer-stats-loader-client"); consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-stats-loader-client-group"); this.consumer = new KafkaConsumer<>(consumerProps); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java index 4429f3518f..fe4300a421 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java @@ -50,7 +50,7 @@ public class TbKafkaConsumerTemplate extends AbstractTbQue String clientId, String groupId, String topic, TbQueueAdmin admin, TbKafkaConsumerStatsService statsService) { super(topic); - Properties props = settings.toConsumerProps(); + Properties props = settings.toConsumerProps(topic); props.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId); if (groupId != null) { props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); 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 86738acea8..f4a10f12c5 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 @@ -31,7 +31,9 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; +import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Properties; /** @@ -95,6 +97,9 @@ public class TbKafkaSettings { @Setter private List other; + @Setter + private Map> consumerPropertiesPerTopic; + public Properties toAdminProps() { Properties props = toProps(); props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, servers); @@ -103,7 +108,7 @@ public class TbKafkaSettings { return props; } - public Properties toConsumerProps() { + public Properties toConsumerProps(String topic) { Properties props = toProps(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers); props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords); @@ -113,6 +118,10 @@ public class TbKafkaSettings { props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); + + consumerPropertiesPerTopic + .getOrDefault(topic, Collections.emptyList()) + .forEach(kv -> props.put(kv.getKey(), kv.getValue())); return props; }