From 3bd2afad8337cde345072ced0e6835267c3335c5 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Fri, 11 Apr 2025 09:27:15 +0300 Subject: [PATCH 1/4] Use consumer-properties-per-topic-inline to configure custom topics --- .../src/main/resources/thingsboard.yml | 5 +++ .../server/common/data/TbProperty.java | 8 +++-- .../server/queue/kafka/TbKafkaSettings.java | 36 +++++++++++++++++++ 3 files changed, 46 insertions(+), 3 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 716797c2d2..5d5962554d 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1628,6 +1628,11 @@ queue: - key: max.poll.records # Max poll records for edqs.state topic value: "${TB_QUEUE_KAFKA_EDQS_STATE_MAX_POLL_RECORDS:512}" + # If you override any default Kafka topic name using environment variables, you must also specify the related consumer properties + # for the new topic in `consumer-properties-per-topic-inline`. Otherwise, the topic will not inherit its expected configuration (e.g., max.poll.records, timeouts, etc). + # Format: "topic1:key1=value1,key2=value2;topic2:key=value" + # Example: "tb_core_modified.notifications:max.poll.records=10;tb_edge_modified:max.poll.records=10,enable.auto.commit=true" + consumer-properties-per-topic-inline: "${TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE:}" other-inline: "${TB_QUEUE_KAFKA_OTHER_PROPERTIES:}" # In this section you can specify custom parameters (semicolon separated) for Kafka consumer/producer/admin # Example "metrics.recording.level:INFO;metrics.sample.window.ms:30000" other: # DEPRECATED. In this section, you can specify custom parameters for Kafka consumer/producer and expose the env variables to configure outside # - key: "request.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_request.timeout.ms diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/TbProperty.java b/common/data/src/main/java/org/thingsboard/server/common/data/TbProperty.java index 98fd521ea3..a72bd6d085 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/TbProperty.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/TbProperty.java @@ -15,14 +15,16 @@ */ package org.thingsboard.server.common.data; +import lombok.AllArgsConstructor; import lombok.Data; +import lombok.NoArgsConstructor; -/** - * Created by ashvayka on 25.09.18. - */ @Data +@NoArgsConstructor +@AllArgsConstructor public class TbProperty { private String key; private String value; + } 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 82b5af179f..e0682f6364 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 @@ -15,6 +15,7 @@ */ package org.thingsboard.server.queue.kafka; +import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.Getter; import lombok.Setter; @@ -36,7 +37,9 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.TbProperty; import org.thingsboard.server.queue.util.PropertyUtils; +import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; @@ -138,6 +141,9 @@ public class TbKafkaSettings { @Value("${queue.kafka.other-inline:}") private String otherInline; + @Value("${queue.kafka.consumer-properties-per-topic-inline:}") + private String consumerPropertiesPerTopicInline; + @Deprecated @Setter private List other; @@ -147,6 +153,14 @@ public class TbKafkaSettings { private volatile AdminClient adminClient; + @PostConstruct + public void initInlineTopicProperties() { + Map> inlineProps = parseTopicPropertyList(consumerPropertiesPerTopicInline); + if (!inlineProps.isEmpty()) { + consumerPropertiesPerTopic.putAll(inlineProps); + } + } + public Properties toConsumerProps(String topic) { Properties props = toProps(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers); @@ -245,6 +259,28 @@ public class TbKafkaSettings { return props; } + private Map> parseTopicPropertyList(String inlineProperties) { + Map> result = new HashMap<>(); + Map rawTopicToPropertyString = PropertyUtils.getProps(inlineProperties); + + for (Map.Entry entry : rawTopicToPropertyString.entrySet()) { + String topic = entry.getKey().trim(); + String propertiesStr = entry.getValue(); + + List tbProperties = Arrays.stream(propertiesStr.split(",")) + .map(kv -> kv.split("=", 2)) + .filter(kvArr -> kvArr.length == 2) + .map(kvArr -> new TbProperty(kvArr[0].trim(), kvArr[1].trim())) + .toList(); + + if (!tbProperties.isEmpty()) { + result.put(topic, tbProperties); + } + } + + return result; + } + @PreDestroy private void destroy() { if (adminClient != null) { From edd7d6392a4fe740f074520a478a52dfe150f408 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Thu, 24 Apr 2025 16:52:31 +0300 Subject: [PATCH 2/4] Improvement after review --- .../src/main/resources/thingsboard.yml | 7 ++-- .../server/queue/kafka/TbKafkaSettings.java | 32 +++++++++---------- .../server/queue/util/PropertyUtils.java | 17 ++++++++++ .../queue/kafka/TbKafkaSettingsTest.java | 20 +++++++++++- 4 files changed, 55 insertions(+), 21 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index ebd001aa97..e013473f98 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1630,9 +1630,10 @@ queue: value: "${TB_QUEUE_KAFKA_EDQS_STATE_MAX_POLL_RECORDS:512}" # If you override any default Kafka topic name using environment variables, you must also specify the related consumer properties # for the new topic in `consumer-properties-per-topic-inline`. Otherwise, the topic will not inherit its expected configuration (e.g., max.poll.records, timeouts, etc). - # Format: "topic1:key1=value1,key2=value2;topic2:key=value" - # Example: "tb_core_modified.notifications:max.poll.records=10;tb_edge_modified:max.poll.records=10,enable.auto.commit=true" - consumer-properties-per-topic-inline: "${TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE:}" + # Each entry sets a single property for a specific topic. To define multiple properties for a topic, repeat the topic key. + # Format: "topic1:key=value;topic1:key=value;topic2:key=value" + # Example: tb_core_updated:max.poll.records=10;tb_core_updated:bootstrap.servers=kafka1:9092,kafka2:9092;tb_edge_updated:auto.offset.reset=latest + consumer-properties-per-topic-inline: "${TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE:tb_core_updated:max.poll.records=10;tb_core_updated:enable.auto.commit=true;tb_core_updated:bootstrap.servers=kafka1:9092,kafka2:9092;tb_edge_updated:max.poll.records=5;tb_edge_updated:auto.offset.reset=latest}" other-inline: "${TB_QUEUE_KAFKA_OTHER_PROPERTIES:}" # In this section you can specify custom parameters (semicolon separated) for Kafka consumer/producer/admin # Example "metrics.recording.level:INFO;metrics.sample.window.ms:30000" other: # DEPRECATED. In this section, you can specify custom parameters for Kafka consumer/producer and expose the env variables to configure outside # - key: "request.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_request.timeout.ms 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 e0682f6364..06ccfe3a69 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 @@ -37,9 +37,8 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.TbProperty; import org.thingsboard.server.queue.util.PropertyUtils; -import java.util.Arrays; -import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Properties; @@ -149,7 +148,7 @@ public class TbKafkaSettings { private List other; @Setter - private Map> consumerPropertiesPerTopic = Collections.emptyMap(); + private Map> consumerPropertiesPerTopic = new HashMap<>(); private volatile AdminClient adminClient; @@ -260,23 +259,22 @@ public class TbKafkaSettings { } private Map> parseTopicPropertyList(String inlineProperties) { + Map> grouped = PropertyUtils.getGroupedProps(inlineProperties); Map> result = new HashMap<>(); - Map rawTopicToPropertyString = PropertyUtils.getProps(inlineProperties); - for (Map.Entry entry : rawTopicToPropertyString.entrySet()) { - String topic = entry.getKey().trim(); - String propertiesStr = entry.getValue(); - - List tbProperties = Arrays.stream(propertiesStr.split(",")) - .map(kv -> kv.split("=", 2)) - .filter(kvArr -> kvArr.length == 2) - .map(kvArr -> new TbProperty(kvArr[0].trim(), kvArr[1].trim())) - .toList(); - - if (!tbProperties.isEmpty()) { - result.put(topic, tbProperties); + grouped.forEach((topic, entries) -> { + Map merged = new LinkedHashMap<>(); + for (String entry : entries) { + String[] kv = entry.split("=", 2); + if (kv.length == 2) { + merged.put(kv[0].trim(), kv[1].trim()); + } } - } + List props = merged.entrySet().stream() + .map(e -> new TbProperty(e.getKey(), e.getValue())) + .toList(); + result.put(topic, props); + }); return result; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/util/PropertyUtils.java b/common/queue/src/main/java/org/thingsboard/server/queue/util/PropertyUtils.java index 6030eb278d..629ee29f6f 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/util/PropertyUtils.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/util/PropertyUtils.java @@ -17,7 +17,9 @@ package org.thingsboard.server.queue.util; import org.thingsboard.server.common.data.StringUtils; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.function.Function; @@ -38,6 +40,21 @@ public class PropertyUtils { return configs; } + public static Map> getGroupedProps(String properties) { + Map> configs = new HashMap<>(); + if (StringUtils.isNotEmpty(properties)) { + for (String property : properties.split(";")) { + if (StringUtils.isNotEmpty(property)) { + int delimiterPosition = property.indexOf(":"); + String topic = property.substring(0, delimiterPosition).trim(); + String value = property.substring(delimiterPosition + 1).trim(); + configs.computeIfAbsent(topic, k -> new ArrayList<>()).add(value); + } + } + } + return configs; + } + public static Map getProps(Map defaultProperties, String propertiesStr) { return getProps(defaultProperties, propertiesStr, PropertyUtils::getProps); } diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/kafka/TbKafkaSettingsTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/kafka/TbKafkaSettingsTest.java index 5cdebc3996..ad026c63aa 100644 --- a/common/queue/src/test/java/org/thingsboard/server/queue/kafka/TbKafkaSettingsTest.java +++ b/common/queue/src/test/java/org/thingsboard/server/queue/kafka/TbKafkaSettingsTest.java @@ -33,6 +33,12 @@ import static org.mockito.Mockito.spy; "queue.type=kafka", "queue.kafka.bootstrap.servers=localhost:9092", "queue.kafka.other-inline=metrics.recording.level:INFO;metrics.sample.window.ms:30000", + "queue.kafka.consumer-properties-per-topic-inline=" + + "tb_core_updated:max.poll.records=10;" + + "tb_core_updated:enable.auto.commit=true;" + + "tb_core_updated:bootstrap.servers=kafka1:9092,kafka2:9092;" + + "tb_edge_updated:max.poll.records=5;" + + "tb_edge_updated:auto.offset.reset=latest" }) class TbKafkaSettingsTest { @@ -79,4 +85,16 @@ class TbKafkaSettingsTest { Mockito.verify(settings).configureSSL(any()); } -} \ No newline at end of file + @Test + void givenMultipleTopicsInInlineConfig_whenParsed_thenEachTopicGetsExpectedProperties() { + Properties coreProps = settings.toConsumerProps("tb_core_updated"); + assertThat(coreProps.getProperty("max.poll.records")).isEqualTo("10"); + assertThat(coreProps.getProperty("enable.auto.commit")).isEqualTo("true"); + assertThat(coreProps.getProperty("bootstrap.servers")).isEqualTo("kafka1:9092,kafka2:9092"); + + Properties edgeProps = settings.toConsumerProps("tb_edge_updated"); + assertThat(edgeProps.getProperty("max.poll.records")).isEqualTo("5"); + assertThat(edgeProps.getProperty("auto.offset.reset")).isEqualTo("latest"); + } + +} From b516583c6dbcf75c0cc0afcb390d7316fdf8576b Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Thu, 24 Apr 2025 16:53:57 +0300 Subject: [PATCH 3/4] Remove env setup config --- application/src/main/resources/thingsboard.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index e013473f98..9710f38a3a 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1633,7 +1633,7 @@ queue: # Each entry sets a single property for a specific topic. To define multiple properties for a topic, repeat the topic key. # Format: "topic1:key=value;topic1:key=value;topic2:key=value" # Example: tb_core_updated:max.poll.records=10;tb_core_updated:bootstrap.servers=kafka1:9092,kafka2:9092;tb_edge_updated:auto.offset.reset=latest - consumer-properties-per-topic-inline: "${TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE:tb_core_updated:max.poll.records=10;tb_core_updated:enable.auto.commit=true;tb_core_updated:bootstrap.servers=kafka1:9092,kafka2:9092;tb_edge_updated:max.poll.records=5;tb_edge_updated:auto.offset.reset=latest}" + consumer-properties-per-topic-inline: "${TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE:}" other-inline: "${TB_QUEUE_KAFKA_OTHER_PROPERTIES:}" # In this section you can specify custom parameters (semicolon separated) for Kafka consumer/producer/admin # Example "metrics.recording.level:INFO;metrics.sample.window.ms:30000" other: # DEPRECATED. In this section, you can specify custom parameters for Kafka consumer/producer and expose the env variables to configure outside # - key: "request.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_request.timeout.ms From b74d88284c0099998fa98a9e71bb2064acd00c3b Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Tue, 6 May 2025 15:37:11 +0300 Subject: [PATCH 4/4] Add TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE to ymls --- edqs/src/main/resources/edqs.yml | 6 +++++- msa/vc-executor/src/main/resources/tb-vc-executor.yml | 5 +++++ transport/coap/src/main/resources/tb-coap-transport.yml | 5 +++++ transport/http/src/main/resources/tb-http-transport.yml | 5 +++++ transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml | 5 +++++ transport/mqtt/src/main/resources/tb-mqtt-transport.yml | 5 +++++ transport/snmp/src/main/resources/tb-snmp-transport.yml | 5 +++++ 7 files changed, 35 insertions(+), 1 deletion(-) diff --git a/edqs/src/main/resources/edqs.yml b/edqs/src/main/resources/edqs.yml index 1cc32a4230..6e6e975d68 100644 --- a/edqs/src/main/resources/edqs.yml +++ b/edqs/src/main/resources/edqs.yml @@ -148,7 +148,11 @@ queue: - key: max.poll.records # Max poll records for edqs.state topic value: "${TB_QUEUE_KAFKA_EDQS_STATE_MAX_POLL_RECORDS:512}" - + # If you override any default Kafka topic name using environment variables, you must also specify the related consumer properties + # for the new topic in `consumer-properties-per-topic-inline`. Otherwise, the topic will not inherit its expected configuration (e.g., max.poll.records, timeouts, etc). + # Format: "topic1:key1=value1,key2=value2;topic2:key=value" + # Example: "tb_core_modified.notifications:max.poll.records=10;tb_edge_modified:max.poll.records=10,enable.auto.commit=true" + consumer-properties-per-topic-inline: "${TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE:}" other-inline: "${TB_QUEUE_KAFKA_OTHER_PROPERTIES:}" # In this section you can specify custom parameters (semicolon separated) for Kafka consumer/producer/admin # Example "metrics.recording.level:INFO;metrics.sample.window.ms:30000" other: # DEPRECATED. In this section, you can specify custom parameters for Kafka consumer/producer and expose the env variables to configure outside # - key: "request.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_request.timeout.ms 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 bb7f607ca0..7d1166e512 100644 --- a/msa/vc-executor/src/main/resources/tb-vc-executor.yml +++ b/msa/vc-executor/src/main/resources/tb-vc-executor.yml @@ -124,6 +124,11 @@ queue: # tb_rule_engine.sq: # - key: max.poll.records # value: "${TB_QUEUE_KAFKA_SQ_MAX_POLL_RECORDS:1024}" + # If you override any default Kafka topic name using environment variables, you must also specify the related consumer properties + # for the new topic in `consumer-properties-per-topic-inline`. Otherwise, the topic will not inherit its expected configuration (e.g., max.poll.records, timeouts, etc). + # Format: "topic1:key1=value1,key2=value2;topic2:key=value" + # Example: "tb_core_modified.notifications:max.poll.records=10;tb_edge_modified:max.poll.records=10,enable.auto.commit=true" + consumer-properties-per-topic-inline: "${TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE:}" other-inline: "${TB_QUEUE_KAFKA_OTHER_PROPERTIES:}" # In this section you can specify custom parameters (semicolon separated) for Kafka consumer/producer/admin # Example "metrics.recording.level:INFO;metrics.sample.window.ms:30000" other: # DEPRECATED. In this section you can specify custom parameters for Kafka consumer/producer and expose the env variables to configure outside # - key: "request.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_request.timeout.ms diff --git a/transport/coap/src/main/resources/tb-coap-transport.yml b/transport/coap/src/main/resources/tb-coap-transport.yml index f60a6bd47e..0c5d160625 100644 --- a/transport/coap/src/main/resources/tb-coap-transport.yml +++ b/transport/coap/src/main/resources/tb-coap-transport.yml @@ -310,6 +310,11 @@ queue: 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\";}" # Protocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL security.protocol: "${TB_QUEUE_KAFKA_CONFLUENT_SECURITY_PROTOCOL:SASL_SSL}" + # If you override any default Kafka topic name using environment variables, you must also specify the related consumer properties + # for the new topic in `consumer-properties-per-topic-inline`. Otherwise, the topic will not inherit its expected configuration (e.g., max.poll.records, timeouts, etc). + # Format: "topic1:key1=value1,key2=value2;topic2:key=value" + # Example: "tb_core_modified.notifications:max.poll.records=10;tb_edge_modified:max.poll.records=10,enable.auto.commit=true" + consumer-properties-per-topic-inline: "${TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE:}" other-inline: "${TB_QUEUE_KAFKA_OTHER_PROPERTIES:}" # In this section you can specify custom parameters (semicolon separated) for Kafka consumer/producer/admin # Example "metrics.recording.level:INFO;metrics.sample.window.ms:30000" other: # DEPRECATED. In this section you can specify custom parameters for Kafka consumer/producer and expose the env variables to configure outside # - key: "request.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_request.timeout.ms diff --git a/transport/http/src/main/resources/tb-http-transport.yml b/transport/http/src/main/resources/tb-http-transport.yml index c921f9f9ae..81f4719e45 100644 --- a/transport/http/src/main/resources/tb-http-transport.yml +++ b/transport/http/src/main/resources/tb-http-transport.yml @@ -259,6 +259,11 @@ queue: 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\";}" # Protocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL security.protocol: "${TB_QUEUE_KAFKA_CONFLUENT_SECURITY_PROTOCOL:SASL_SSL}" + # If you override any default Kafka topic name using environment variables, you must also specify the related consumer properties + # for the new topic in `consumer-properties-per-topic-inline`. Otherwise, the topic will not inherit its expected configuration (e.g., max.poll.records, timeouts, etc). + # Format: "topic1:key1=value1,key2=value2;topic2:key=value" + # Example: "tb_core_modified.notifications:max.poll.records=10;tb_edge_modified:max.poll.records=10,enable.auto.commit=true" + consumer-properties-per-topic-inline: "${TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE:}" other-inline: "${TB_QUEUE_KAFKA_OTHER_PROPERTIES:}" # In this section you can specify custom parameters (semicolon separated) for Kafka consumer/producer/admin # Example "metrics.recording.level:INFO;metrics.sample.window.ms:30000" other: # DEPRECATED. In this section you can specify custom parameters for Kafka consumer/producer and expose the env variables to configure outside # - key: "request.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_request.timeout.ms diff --git a/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml b/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml index 85e865e60e..2c62a35dbb 100644 --- a/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml +++ b/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml @@ -360,6 +360,11 @@ queue: 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\";}" # Protocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL security.protocol: "${TB_QUEUE_KAFKA_CONFLUENT_SECURITY_PROTOCOL:SASL_SSL}" + # If you override any default Kafka topic name using environment variables, you must also specify the related consumer properties + # for the new topic in `consumer-properties-per-topic-inline`. Otherwise, the topic will not inherit its expected configuration (e.g., max.poll.records, timeouts, etc). + # Format: "topic1:key1=value1,key2=value2;topic2:key=value" + # Example: "tb_core_modified.notifications:max.poll.records=10;tb_edge_modified:max.poll.records=10,enable.auto.commit=true" + consumer-properties-per-topic-inline: "${TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE:}" other-inline: "${TB_QUEUE_KAFKA_OTHER_PROPERTIES:}" # In this section you can specify custom parameters (semicolon separated) for Kafka consumer/producer/admin # Example "metrics.recording.level:INFO;metrics.sample.window.ms:30000" other: # DEPRECATED. In this section you can specify custom parameters for Kafka consumer/producer and expose the env variables to configure outside # - key: "request.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_request.timeout.ms diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index a6ca2f1a6e..51f9a17005 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -293,6 +293,11 @@ queue: 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\";}" # Protocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL security.protocol: "${TB_QUEUE_KAFKA_CONFLUENT_SECURITY_PROTOCOL:SASL_SSL}" + # If you override any default Kafka topic name using environment variables, you must also specify the related consumer properties + # for the new topic in `consumer-properties-per-topic-inline`. Otherwise, the topic will not inherit its expected configuration (e.g., max.poll.records, timeouts, etc). + # Format: "topic1:key1=value1,key2=value2;topic2:key=value" + # Example: "tb_core_modified.notifications:max.poll.records=10;tb_edge_modified:max.poll.records=10,enable.auto.commit=true" + consumer-properties-per-topic-inline: "${TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE:}" other-inline: "${TB_QUEUE_KAFKA_OTHER_PROPERTIES:}" # In this section you can specify custom parameters (semicolon separated) for Kafka consumer/producer/admin # Example "metrics.recording.level:INFO;metrics.sample.window.ms:30000" other: # DEPRECATED. In this section you can specify custom parameters for Kafka consumer/producer and expose the env variables to configure outside # - key: "request.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_request.timeout.ms diff --git a/transport/snmp/src/main/resources/tb-snmp-transport.yml b/transport/snmp/src/main/resources/tb-snmp-transport.yml index 6848e8af26..f4811b3326 100644 --- a/transport/snmp/src/main/resources/tb-snmp-transport.yml +++ b/transport/snmp/src/main/resources/tb-snmp-transport.yml @@ -239,6 +239,11 @@ queue: 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\";}" # Protocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL security.protocol: "${TB_QUEUE_KAFKA_CONFLUENT_SECURITY_PROTOCOL:SASL_SSL}" + # If you override any default Kafka topic name using environment variables, you must also specify the related consumer properties + # for the new topic in `consumer-properties-per-topic-inline`. Otherwise, the topic will not inherit its expected configuration (e.g., max.poll.records, timeouts, etc). + # Format: "topic1:key1=value1,key2=value2;topic2:key=value" + # Example: "tb_core_modified.notifications:max.poll.records=10;tb_edge_modified:max.poll.records=10,enable.auto.commit=true" + consumer-properties-per-topic-inline: "${TB_QUEUE_KAFKA_CONSUMER_PROPERTIES_PER_TOPIC_INLINE:}" other-inline: "${TB_QUEUE_KAFKA_OTHER_PROPERTIES:}" # In this section you can specify custom parameters (semicolon separated) for Kafka consumer/producer/admin # Example "metrics.recording.level:INFO;metrics.sample.window.ms:30000" other: # DEPRECATED. In this section you can specify custom parameters for Kafka consumer/producer and expose the env variables to configure outside # - key: "request.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_request.timeout.ms