From d654e09d6a52c0a0d038ac176f36214c0dce25b2 Mon Sep 17 00:00:00 2001 From: Yevhen Bondarenko <56396344+YevhenBondarenko@users.noreply.github.com> Date: Wed, 15 Apr 2020 17:47:46 +0300 Subject: [PATCH] kafka topic settings * added topic-properties to kafka queue * added topic-properties to kafka queue to transport * kafka topic settings improvements --- .../src/main/resources/thingsboard.yml | 7 ++ .../server/queue/kafka/TBKafkaAdmin.java | 43 +++++++----- .../queue/kafka/TBKafkaConsumerTemplate.java | 8 ++- .../queue/kafka/TBKafkaProducerTemplate.java | 26 ++++++- .../server/queue/kafka/TbKafkaSettings.java | 8 ++- .../queue/kafka/TbKafkaTopicConfigs.java | 69 +++++++++++++++++++ .../provider/KafkaMonolithQueueFactory.java | 34 +++++++-- .../provider/KafkaTbCoreQueueFactory.java | 32 +++++++-- .../KafkaTbRuleEngineQueueFactory.java | 28 ++++++-- .../KafkaTbTransportQueueFactory.java | 42 +++++++---- .../src/main/resources/tb-coap-transport.yml | 7 ++ .../src/main/resources/tb-http-transport.yml | 7 ++ .../src/main/resources/tb-mqtt-transport.yml | 7 ++ 13 files changed, 267 insertions(+), 51 deletions(-) create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index c138c23fc9..7db2224d34 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -521,6 +521,13 @@ queue: batch.size: "${TB_KAFKA_BATCH_SIZE:16384}" linger.ms: "${TB_KAFKA_LINGER_MS:1}" buffer.memory: "${TB_BUFFER_MEMORY:33554432}" + replication_factor: "${TB_QUEUE_KAFKA_REPLICATION_FACTOR:1}" + 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}" + transport-api: "${TB_QUEUE_KAFKA_TA_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" + notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" + js-executor: "${TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:104857600}" aws_sqs: access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}" secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}" diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaAdmin.java index 76c977e25a..69ad6286a6 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaAdmin.java @@ -19,15 +19,14 @@ import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.CreateTopicsResult; import org.apache.kafka.clients.admin.NewTopic; -import org.apache.kafka.clients.admin.TopicDescription; -import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.errors.TopicExistsException; import org.thingsboard.server.queue.TbQueueAdmin; import java.util.Collections; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; /** * Created by ashvayka on 24.09.18. @@ -35,17 +34,34 @@ import java.util.concurrent.TimeoutException; @Slf4j public class TBKafkaAdmin implements TbQueueAdmin { - AdminClient client; + private final AdminClient client; + private final Map topicConfigs; + private final Set topics = ConcurrentHashMap.newKeySet(); - public TBKafkaAdmin(TbKafkaSettings settings) { + private final short replicationFactor; + + public TBKafkaAdmin(TbKafkaSettings settings, Map topicConfigs) { client = AdminClient.create(settings.toProps()); + this.topicConfigs = topicConfigs; + + try { + topics.addAll(client.listTopics().names().get()); + } catch (InterruptedException | ExecutionException e) { + log.error("Failed to get all topics.", e); + } + + replicationFactor = settings.getReplicationFactor(); } - //TODO 2.5 - ybondarenko Need to pass not only settings but also properties for topic creation. Somewhere in thingsboard.yml, in KV format. @Override public void createTopicIfNotExists(String topic) { + if (topics.contains(topic)) { + return; + } try { - createTopic(new NewTopic(topic, 1, (short) 1)).values().get(topic).get(); + NewTopic newTopic = new NewTopic(topic, 1, replicationFactor).configs(topicConfigs); + createTopic(newTopic).values().get(topic).get(); + topics.add(topic); } catch (ExecutionException ee) { if (ee.getCause() instanceof TopicExistsException) { //do nothing @@ -57,19 +73,10 @@ public class TBKafkaAdmin implements TbQueueAdmin { log.warn("[{}] Failed to create topic", topic, e); throw new RuntimeException(e); } + } public CreateTopicsResult createTopic(NewTopic topic) { return client.createTopics(Collections.singletonList(topic)); } - - private boolean topicExists(String topic) throws InterruptedException { - KafkaFuture topicDescriptionFuture = client.describeTopics(Collections.singleton(topic)).values().get(topic); - try { - topicDescriptionFuture.get(); - return true; - } catch (ExecutionException e) { - return false; - } - } } 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 d882153177..069f4a6455 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 @@ -23,6 +23,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueMsg; @@ -43,7 +44,7 @@ import java.util.stream.Collectors; @Slf4j public class TBKafkaConsumerTemplate implements TbQueueConsumer { - private final TBKafkaAdmin admin; + private final TbQueueAdmin admin; private final KafkaConsumer consumer; private final TbKafkaDecoder decoder; private volatile boolean subscribed; @@ -57,7 +58,8 @@ public class TBKafkaConsumerTemplate implements TbQueueCon private TBKafkaConsumerTemplate(TbKafkaSettings settings, TbKafkaDecoder decoder, String clientId, String groupId, String topic, boolean autoCommit, int autoCommitIntervalMs, - int maxPollRecords) { + int maxPollRecords, + TbQueueAdmin admin) { Properties props = settings.toProps(); props.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId); if (groupId != null) { @@ -70,7 +72,7 @@ public class TBKafkaConsumerTemplate implements TbQueueCon if (maxPollRecords > 0) { props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords); } - this.admin = new TBKafkaAdmin(settings); + this.admin = admin; this.consumer = new KafkaConsumer<>(props); this.decoder = decoder; this.topic = topic; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaProducerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaProducerTemplate.java index 2d82ed275d..f81381bc38 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaProducerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaProducerTemplate.java @@ -24,12 +24,15 @@ import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.header.Header; import org.apache.kafka.common.header.internals.RecordHeader; import org.springframework.util.StringUtils; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueMsg; import org.thingsboard.server.queue.TbQueueProducer; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import java.util.Properties; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; /** @@ -46,8 +49,12 @@ public class TBKafkaProducerTemplate implements TbQueuePro @Getter private final TbKafkaSettings settings; + private final TbQueueAdmin admin; + + private final Set topics; + @Builder - private TBKafkaProducerTemplate(TbKafkaSettings settings, TbKafkaPartitioner partitioner, String defaultTopic, String clientId) { + private TBKafkaProducerTemplate(TbKafkaSettings settings, TbKafkaPartitioner partitioner, String defaultTopic, String clientId, TbQueueAdmin admin) { Properties props = settings.toProps(); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); @@ -57,6 +64,8 @@ public class TBKafkaProducerTemplate implements TbQueuePro this.settings = settings; this.producer = new KafkaProducer<>(props); this.defaultTopic = defaultTopic; + this.admin = admin; + topics = ConcurrentHashMap.newKeySet(); } @Override @@ -65,6 +74,7 @@ public class TBKafkaProducerTemplate implements TbQueuePro @Override public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) { + createTopicIfNotExist(tpi); String key = msg.getKey().toString(); byte[] data = msg.getData(); ProducerRecord record; @@ -85,8 +95,18 @@ public class TBKafkaProducerTemplate implements TbQueuePro }); } + private void createTopicIfNotExist(TopicPartitionInfo tpi) { + if (topics.contains(tpi)) { + return; + } + admin.createTopicIfNotExists(tpi.getFullTopicName()); + topics.add(tpi); + } + @Override public void stop() { - + if (producer != null) { + producer.close(); + } } } 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 e8581d9719..66121cb215 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 lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.producer.ProducerConfig; import org.springframework.beans.factory.annotation.Value; @@ -32,9 +33,6 @@ import java.util.Properties; @Component public class TbKafkaSettings { - static final String REQUEST_ID_HEADER = "requestId"; - static final String RESPONSE_TOPIC_HEADER = "responseTopic"; - @Value("${queue.kafka.bootstrap.servers}") private String servers; @@ -53,6 +51,10 @@ public class TbKafkaSettings { @Value("${queue.kafka.buffer.memory}") private long bufferMemory; + @Value("${queue.kafka.replication_factor}") + @Getter + private short replicationFactor; + @Value("${kafka.other:#{null}}") private List other; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java new file mode 100644 index 0000000000..8682ca385f --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java @@ -0,0 +1,69 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.queue.kafka; + +import lombok.Getter; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +import javax.annotation.PostConstruct; +import java.util.HashMap; +import java.util.Map; + +@Component +public class TbKafkaTopicConfigs { + @Value("${queue.kafka.topic-properties.core}") + private String coreProperties; + @Value("${queue.kafka.topic-properties.rule-engine}") + private String ruleEngineProperties; + @Value("${queue.kafka.topic-properties.transport-api}") + private String transportApiProperties; + @Value("${queue.kafka.topic-properties.notifications}") + private String notificationsProperties; + @Value("${queue.kafka.topic-properties.js-executor}") + private String jsExecutorProperties; + + @Getter + private Map coreConfigs; + @Getter + private Map ruleEngineConfigs; + @Getter + private Map transportApiConfigs; + @Getter + private Map notificationsConfigs; + @Getter + private Map jsExecutorConfigs; + + @PostConstruct + private void init() { + coreConfigs = getConfigs(coreProperties); + ruleEngineConfigs = getConfigs(ruleEngineProperties); + transportApiConfigs = getConfigs(transportApiProperties); + notificationsConfigs = getConfigs(notificationsProperties); + jsExecutorConfigs = getConfigs(jsExecutorProperties); + } + + private Map getConfigs(String properties) { + Map configs = new HashMap<>(); + for (String property : properties.split(";")) { + int delimiterPosition = property.indexOf(":"); + String key = property.substring(0, delimiterPosition); + String value = property.substring(delimiterPosition + 1); + configs.put(key, value); + } + return configs; + } +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java index dae4221e91..9eb656c2ea 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java @@ -28,6 +28,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotifica import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; +import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueRequestTemplate; @@ -40,6 +41,7 @@ import org.thingsboard.server.queue.kafka.TBKafkaAdmin; import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; +import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; @@ -62,13 +64,20 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbQueueAdmin coreAdmin; + private final TbQueueAdmin ruleEngineAdmin; + private final TbQueueAdmin jsExecutorAdmin; + private final TbQueueAdmin transportApiAdmin; + private final TbQueueAdmin notificationAdmin; + public KafkaMonolithQueueFactory(PartitionService partitionService, TbKafkaSettings kafkaSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings, - TbQueueRemoteJsInvokeSettings jsInvokeSettings) { + TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbKafkaTopicConfigs kafkaTopicConfigs) { this.partitionService = partitionService; this.kafkaSettings = kafkaSettings; this.serviceInfoProvider = serviceInfoProvider; @@ -77,6 +86,12 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; this.jsInvokeSettings = jsInvokeSettings; + + this.coreAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); + this.ruleEngineAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); + this.jsExecutorAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getJsExecutorConfigs()); + this.transportApiAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs()); + this.notificationAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); } @Override @@ -85,6 +100,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-transport-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(transportNotificationSettings.getNotificationsTopic()); + requestBuilder.admin(notificationAdmin); return requestBuilder.build(); } @@ -94,6 +110,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-rule-engine-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); + requestBuilder.admin(ruleEngineAdmin); return requestBuilder.build(); } @@ -103,6 +120,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-rule-engine-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); + requestBuilder.admin(ruleEngineAdmin); return requestBuilder.build(); } @@ -112,6 +130,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-core-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); + requestBuilder.admin(coreAdmin); return requestBuilder.build(); } @@ -121,6 +140,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-core-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); + requestBuilder.admin(coreAdmin); return requestBuilder.build(); } @@ -133,6 +153,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.clientId("re-" + queueName + "-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("re-" + queueName + "-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(ruleEngineAdmin); return consumerBuilder.build(); } @@ -144,6 +165,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.clientId("monolith-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("monolith-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(notificationAdmin); return consumerBuilder.build(); } @@ -155,6 +177,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.clientId("monolith-core-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("monolith-core-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(coreAdmin); return consumerBuilder.build(); } @@ -166,6 +189,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.clientId("monolith-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("monolith-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(notificationAdmin); return consumerBuilder.build(); } @@ -177,6 +201,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.clientId("monolith-transport-api-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("monolith-transport-api-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(transportApiAdmin); return consumerBuilder.build(); } @@ -186,6 +211,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-transport-api-producer-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(transportApiSettings.getResponsesTopic()); + requestBuilder.admin(transportApiAdmin); return requestBuilder.build(); } @@ -196,24 +222,24 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi requestBuilder.settings(kafkaSettings); requestBuilder.clientId("producer-js-invoke-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(jsInvokeSettings.getRequestTopic()); + requestBuilder.admin(jsExecutorAdmin); TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); responseBuilder.settings(kafkaSettings); responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); responseBuilder.groupId("rule-engine-node-" + serviceInfoProvider.getServiceId()); -// responseBuilder.autoCommit(true); -// responseBuilder.autoCommitIntervalMs(autoCommitInterval); responseBuilder.decoder(msg -> { JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); return new TbProtoQueueMsg<>(msg.getKey(), builder.build(), msg.getHeaders()); } ); + responseBuilder.admin(jsExecutorAdmin); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> builder = DefaultTbQueueRequestTemplate.builder(); - builder.queueAdmin(new TBKafkaAdmin(kafkaSettings)); + builder.queueAdmin(jsExecutorAdmin); builder.requestTemplate(requestBuilder.build()); builder.responseTemplate(responseBuilder.build()); builder.maxPendingRequests(jsInvokeSettings.getMaxPendingRequests()); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java index e148416bb4..8800c1c55a 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java @@ -28,6 +28,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotifica import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; +import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueRequestTemplate; @@ -40,6 +41,7 @@ import org.thingsboard.server.queue.kafka.TBKafkaAdmin; import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; +import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; @@ -59,12 +61,19 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbQueueAdmin coreAdmin; + private final TbQueueAdmin ruleEngineAdmin; + private final TbQueueAdmin jsExecutorAdmin; + private final TbQueueAdmin transportApiAdmin; + private final TbQueueAdmin notificationAdmin; + public KafkaTbCoreQueueFactory(PartitionService partitionService, TbKafkaSettings kafkaSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbQueueTransportApiSettings transportApiSettings, - TbQueueRemoteJsInvokeSettings jsInvokeSettings) { + TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbKafkaTopicConfigs kafkaTopicConfigs) { this.partitionService = partitionService; this.kafkaSettings = kafkaSettings; this.serviceInfoProvider = serviceInfoProvider; @@ -72,6 +81,12 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { this.ruleEngineSettings = ruleEngineSettings; this.transportApiSettings = transportApiSettings; this.jsInvokeSettings = jsInvokeSettings; + + this.coreAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); + this.ruleEngineAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); + this.jsExecutorAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getJsExecutorConfigs()); + this.transportApiAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs()); + this.notificationAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); } @Override @@ -80,6 +95,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-transport-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); + requestBuilder.admin(coreAdmin); return requestBuilder.build(); } @@ -89,6 +105,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-rule-engine-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); + requestBuilder.admin(coreAdmin); return requestBuilder.build(); } @@ -98,6 +115,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-rule-engine-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); + requestBuilder.admin(ruleEngineAdmin); return requestBuilder.build(); } @@ -107,6 +125,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-to-core-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); + requestBuilder.admin(coreAdmin); return requestBuilder.build(); } @@ -116,6 +135,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-to-core-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); + requestBuilder.admin(coreAdmin); return requestBuilder.build(); } @@ -127,6 +147,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { consumerBuilder.clientId("tb-core-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("tb-core-node-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(coreAdmin); return consumerBuilder.build(); } @@ -138,6 +159,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { consumerBuilder.clientId("tb-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("tb-core-notifications-node-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(notificationAdmin); return consumerBuilder.build(); } @@ -149,6 +171,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { consumerBuilder.clientId("tb-core-transport-api-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("tb-core-transport-api-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(transportApiAdmin); return consumerBuilder.build(); } @@ -158,6 +181,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-transport-api-producer-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); + requestBuilder.admin(coreAdmin); return requestBuilder.build(); } @@ -168,24 +192,24 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("producer-js-invoke-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(jsInvokeSettings.getRequestTopic()); + requestBuilder.admin(jsExecutorAdmin); TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); responseBuilder.settings(kafkaSettings); responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); responseBuilder.groupId("rule-engine-node-" + serviceInfoProvider.getServiceId()); -// responseBuilder.autoCommit(true); -// responseBuilder.autoCommitIntervalMs(autoCommitInterval); responseBuilder.decoder(msg -> { JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); return new TbProtoQueueMsg<>(msg.getKey(), builder.build(), msg.getHeaders()); } ); + responseBuilder.admin(jsExecutorAdmin); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> builder = DefaultTbQueueRequestTemplate.builder(); - builder.queueAdmin(new TBKafkaAdmin(kafkaSettings)); + builder.queueAdmin(jsExecutorAdmin); builder.requestTemplate(requestBuilder.build()); builder.responseTemplate(responseBuilder.build()); builder.maxPendingRequests(jsInvokeSettings.getMaxPendingRequests()); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java index 8262d27354..b96d4ebe86 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java @@ -26,6 +26,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMs import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueRequestTemplate; @@ -38,6 +39,7 @@ import org.thingsboard.server.queue.kafka.TBKafkaAdmin; import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; +import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; @@ -56,17 +58,28 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbQueueAdmin coreAdmin; + private final TbQueueAdmin ruleEngineAdmin; + private final TbQueueAdmin jsExecutorAdmin; + private final TbQueueAdmin notificationAdmin; + public KafkaTbRuleEngineQueueFactory(PartitionService partitionService, TbKafkaSettings kafkaSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, - TbQueueRemoteJsInvokeSettings jsInvokeSettings) { + TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbKafkaTopicConfigs kafkaTopicConfigs) { this.partitionService = partitionService; this.kafkaSettings = kafkaSettings; this.serviceInfoProvider = serviceInfoProvider; this.coreSettings = coreSettings; this.ruleEngineSettings = ruleEngineSettings; this.jsInvokeSettings = jsInvokeSettings; + + this.coreAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); + this.ruleEngineAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); + this.jsExecutorAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getJsExecutorConfigs()); + this.notificationAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); } @Override @@ -75,6 +88,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-transport-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); + requestBuilder.admin(coreAdmin); return requestBuilder.build(); } @@ -84,6 +98,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-to-rule-engine-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); + requestBuilder.admin(coreAdmin); return requestBuilder.build(); } @@ -93,6 +108,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-to-rule-engine-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); + requestBuilder.admin(ruleEngineAdmin); return requestBuilder.build(); } @@ -103,6 +119,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-to-core-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); + requestBuilder.admin(coreAdmin); return requestBuilder.build(); } @@ -112,6 +129,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-to-core-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); + requestBuilder.admin(coreAdmin); return requestBuilder.build(); } @@ -124,6 +142,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { consumerBuilder.clientId("re-" + queueName + "-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("re-" + queueName + "-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(ruleEngineAdmin); return consumerBuilder.build(); } @@ -135,6 +154,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { consumerBuilder.clientId("tb-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("tb-rule-engine-notifications-node-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(notificationAdmin); return consumerBuilder.build(); } @@ -145,24 +165,24 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("producer-js-invoke-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(jsInvokeSettings.getRequestTopic()); + requestBuilder.admin(jsExecutorAdmin); TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); responseBuilder.settings(kafkaSettings); responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); responseBuilder.groupId("rule-engine-node-" + serviceInfoProvider.getServiceId()); -// responseBuilder.autoCommit(true); -// responseBuilder.autoCommitIntervalMs(autoCommitInterval); responseBuilder.decoder(msg -> { JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); return new TbProtoQueueMsg<>(msg.getKey(), builder.build(), msg.getHeaders()); } ); + responseBuilder.admin(jsExecutorAdmin); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> builder = DefaultTbQueueRequestTemplate.builder(); - builder.queueAdmin(new TBKafkaAdmin(kafkaSettings)); + builder.queueAdmin(jsExecutorAdmin); builder.requestTemplate(requestBuilder.build()); builder.responseTemplate(responseBuilder.build()); builder.maxPendingRequests(jsInvokeSettings.getMaxPendingRequests()); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java index 12f677af9d..f5a31cf1c7 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java @@ -18,25 +18,27 @@ package org.thingsboard.server.queue.provider; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Component; -import org.thingsboard.server.queue.TbQueueConsumer; -import org.thingsboard.server.queue.settings.TbQueueCoreSettings; -import org.thingsboard.server.queue.TbQueueProducer; -import org.thingsboard.server.queue.TbQueueRequestTemplate; -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; -import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; -import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; -import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; -import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; +import org.thingsboard.server.queue.TbQueueAdmin; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.kafka.TBKafkaAdmin; import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; +import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; +import org.thingsboard.server.queue.settings.TbQueueCoreSettings; +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; @Component @ConditionalOnExpression("'${queue.type:null}'=='kafka' && ('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport')") @@ -50,18 +52,29 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; + private final TbQueueAdmin coreAdmin; + private final TbQueueAdmin ruleEngineAdmin; + private final TbQueueAdmin transportApiAdmin; + private final TbQueueAdmin notificationAdmin; + public KafkaTbTransportQueueFactory(TbKafkaSettings kafkaSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbQueueTransportApiSettings transportApiSettings, - TbQueueTransportNotificationSettings transportNotificationSettings) { + TbQueueTransportNotificationSettings transportNotificationSettings, + TbKafkaTopicConfigs kafkaTopicConfigs) { this.kafkaSettings = kafkaSettings; this.serviceInfoProvider = serviceInfoProvider; this.coreSettings = coreSettings; this.ruleEngineSettings = ruleEngineSettings; this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; + + this.coreAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); + this.ruleEngineAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); + this.transportApiAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs()); + this.notificationAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); } @Override @@ -70,6 +83,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("transport-api-request-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(transportApiSettings.getRequestsTopic()); + requestBuilder.admin(transportApiAdmin); TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); responseBuilder.settings(kafkaSettings); @@ -77,10 +91,11 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { responseBuilder.clientId("transport-api-response-" + serviceInfoProvider.getServiceId()); responseBuilder.groupId("transport-node-" + serviceInfoProvider.getServiceId()); responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders())); + responseBuilder.admin(transportApiAdmin); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> templateBuilder = DefaultTbQueueRequestTemplate.builder(); - templateBuilder.queueAdmin(new TBKafkaAdmin(kafkaSettings)); + templateBuilder.queueAdmin(transportApiAdmin); templateBuilder.requestTemplate(requestBuilder.build()); templateBuilder.responseTemplate(responseBuilder.build()); templateBuilder.maxPendingRequests(transportApiSettings.getMaxPendingRequests()); @@ -93,8 +108,9 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { public TbQueueProducer> createRuleEngineMsgProducer() { TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); - requestBuilder.clientId("transport-node-rule-engine-"+ serviceInfoProvider.getServiceId()); + requestBuilder.clientId("transport-node-rule-engine-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); + requestBuilder.admin(ruleEngineAdmin); return requestBuilder.build(); } @@ -104,6 +120,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("transport-node-core-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); + requestBuilder.admin(coreAdmin); return requestBuilder.build(); } @@ -115,6 +132,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { responseBuilder.clientId("transport-api-notifications-" + serviceInfoProvider.getServiceId()); responseBuilder.groupId("transport-node-" + serviceInfoProvider.getServiceId()); responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); + responseBuilder.admin(notificationAdmin); return responseBuilder.build(); } } diff --git a/transport/coap/src/main/resources/tb-coap-transport.yml b/transport/coap/src/main/resources/tb-coap-transport.yml index b94148f608..e2fb7e149a 100644 --- a/transport/coap/src/main/resources/tb-coap-transport.yml +++ b/transport/coap/src/main/resources/tb-coap-transport.yml @@ -50,6 +50,13 @@ queue: batch.size: "${TB_KAFKA_BATCH_SIZE:16384}" linger.ms: "${TB_KAFKA_LINGER_MS:1}" buffer.memory: "${TB_BUFFER_MEMORY:33554432}" + replication_factor: "${TB_QUEUE_KAFKA_REPLICATION_FACTOR:1}" + 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}" + transport-api: "${TB_QUEUE_KAFKA_TA_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" + notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" + js-executor: "${TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:104857600}" aws_sqs: access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}" secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}" diff --git a/transport/http/src/main/resources/tb-http-transport.yml b/transport/http/src/main/resources/tb-http-transport.yml index 0fa2788cac..dfd2c14be2 100644 --- a/transport/http/src/main/resources/tb-http-transport.yml +++ b/transport/http/src/main/resources/tb-http-transport.yml @@ -51,6 +51,13 @@ queue: batch.size: "${TB_KAFKA_BATCH_SIZE:16384}" linger.ms: "${TB_KAFKA_LINGER_MS:1}" buffer.memory: "${TB_BUFFER_MEMORY:33554432}" + replication_factor: "${TB_QUEUE_KAFKA_REPLICATION_FACTOR:1}" + 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}" + transport-api: "${TB_QUEUE_KAFKA_TA_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" + notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" + js-executor: "${TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:104857600}" aws_sqs: access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}" secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}" diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index 747427caf8..81f0fb795d 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -81,6 +81,13 @@ queue: batch.size: "${TB_KAFKA_BATCH_SIZE:16384}" linger.ms: "${TB_KAFKA_LINGER_MS:1}" buffer.memory: "${TB_BUFFER_MEMORY:33554432}" + replication_factor: "${TB_QUEUE_KAFKA_REPLICATION_FACTOR:1}" + 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}" + transport-api: "${TB_QUEUE_KAFKA_TA_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" + notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" + js-executor: "${TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:104857600}" aws_sqs: access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}" secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}"