Browse Source

added other parameters for queue kafka

pull/3244/head
YevhenBondarenko 6 years ago
committed by Andrew Shvayka
parent
commit
86102aea06
  1. 10
      application/src/main/resources/thingsboard.yml
  2. 5
      common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java

10
application/src/main/resources/thingsboard.yml

@ -605,6 +605,16 @@ queue:
max_poll_records: "${TB_QUEUE_KAFKA_MAX_POLL_RECORDS:8192}" max_poll_records: "${TB_QUEUE_KAFKA_MAX_POLL_RECORDS:8192}"
max_partition_fetch_bytes: "${TB_QUEUE_KAFKA_MAX_PARTITION_FETCH_BYTES:16777216}" max_partition_fetch_bytes: "${TB_QUEUE_KAFKA_MAX_PARTITION_FETCH_BYTES:16777216}"
fetch_max_bytes: "${TB_QUEUE_KAFKA_FETCH_MAX_BYTES:134217728}" 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: topic-properties:
rule-engine: "${TB_QUEUE_KAFKA_RE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" 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}" core: "${TB_QUEUE_KAFKA_CORE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}"

5
common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java

@ -16,10 +16,12 @@
package org.thingsboard.server.queue.kafka; package org.thingsboard.server.queue.kafka;
import lombok.Getter; import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import java.util.List; import java.util.List;
@ -30,6 +32,7 @@ import java.util.Properties;
*/ */
@Slf4j @Slf4j
@ConditionalOnExpression("'${queue.type:null}'=='kafka'") @ConditionalOnExpression("'${queue.type:null}'=='kafka'")
@ConfigurationProperties(prefix = "queue.kafka")
@Component @Component
public class TbKafkaSettings { public class TbKafkaSettings {
@ -67,7 +70,7 @@ public class TbKafkaSettings {
@Getter @Getter
private int fetchMaxBytes; private int fetchMaxBytes;
@Value("${kafka.other:#{null}}") @Setter
private List<TbKafkaProperty> other; private List<TbKafkaProperty> other;
public Properties toProps() { public Properties toProps() {

Loading…
Cancel
Save