From 86102aea06735fcd7f7f5d306a55817eb2b23d34 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Tue, 4 Aug 2020 12:38:58 +0300 Subject: [PATCH] added other parameters for queue kafka --- application/src/main/resources/thingsboard.yml | 10 ++++++++++ .../server/queue/kafka/TbKafkaSettings.java | 5 ++++- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 73db8c6735..2a58209560 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -605,6 +605,16 @@ queue: max_poll_records: "${TB_QUEUE_KAFKA_MAX_POLL_RECORDS:8192}" max_partition_fetch_bytes: "${TB_QUEUE_KAFKA_MAX_PARTITION_FETCH_BYTES:16777216}" fetch_max_bytes: "${TB_QUEUE_KAFKA_FETCH_MAX_BYTES:134217728}" + other: +# Properties for Confluent cloud +# - key: "ssl.endpoint.identification.algorithm" +# value: "https" +# - key: "sasl.mechanism" +# value: "PLAIN" +# - key: "sasl.jaas.config" +# value: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"CLUSTER_API_KEY\" password=\"CLUSTER_API_SECRET\";" +# - key: "security.protocol" +# value: "SASL_SSL" topic-properties: rule-engine: "${TB_QUEUE_KAFKA_RE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" core: "${TB_QUEUE_KAFKA_CORE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" 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 659dd19bda..d9da969eb9 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 @@ -16,10 +16,12 @@ package org.thingsboard.server.queue.kafka; import lombok.Getter; +import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.producer.ProducerConfig; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; import java.util.List; @@ -30,6 +32,7 @@ import java.util.Properties; */ @Slf4j @ConditionalOnExpression("'${queue.type:null}'=='kafka'") +@ConfigurationProperties(prefix = "queue.kafka") @Component public class TbKafkaSettings { @@ -67,7 +70,7 @@ public class TbKafkaSettings { @Getter private int fetchMaxBytes; - @Value("${kafka.other:#{null}}") + @Setter private List other; public Properties toProps() {