diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 2a58209560..b6894bb0e7 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -605,16 +605,13 @@ 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}" + use_confluent_cloud: "${TB_QUEUE_KAFKA_USE_CONFLUENT_CLOUD:false}" + confluent: + ssl.algorithm: "${TB_QUEUE_KAFKA_CONFLUENT_SSL_ALGORITHM:https}" + 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}" 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 d9da969eb9..37e978c07d 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 @@ -18,6 +18,7 @@ package org.thingsboard.server.queue.kafka; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.producer.ProducerConfig; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; @@ -68,7 +69,22 @@ public class TbKafkaSettings { @Value("${queue.kafka.fetch_max_bytes:134217728}") @Getter - private int fetchMaxBytes; + private int fetchMaxBytes; + + @Value("${queue.kafka.use_confluent_cloud:false}") + private boolean useConfluent; + + @Value("${queue.kafka.confluent.ssl.algorithm}") + private String sslAlgorithm; + + @Value("${queue.kafka.confluent.sasl.mechanism}") + private String saslMechanism; + + @Value("${queue.kafka.confluent.sasl.config}") + private String saslConfig; + + @Value("${queue.kafka.confluent.security.protocol}") + private String securityProtocol; @Setter private List other; @@ -76,12 +92,21 @@ public class TbKafkaSettings { public Properties toProps() { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers); - props.put(ProducerConfig.ACKS_CONFIG, acks); props.put(ProducerConfig.RETRIES_CONFIG, retries); - props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize); - props.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs); - props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory); - if(other != null){ + + if (useConfluent) { + props.put("ssl.endpoint.identification.algorithm", sslAlgorithm); + props.put("sasl.mechanism", saslMechanism); + props.put("sasl.jaas.config", saslConfig); + props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol); + } else { + props.put(ProducerConfig.ACKS_CONFIG, acks); + props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize); + props.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs); + props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory); + } + + if (other != null) { other.forEach(kv -> props.put(kv.getKey(), kv.getValue())); } return props; diff --git a/docker/compose-utils.sh b/docker/compose-utils.sh index ac8d4fce78..a1fe74ae3f 100755 --- a/docker/compose-utils.sh +++ b/docker/compose-utils.sh @@ -39,6 +39,9 @@ function additionalComposeQueueArgs() { kafka) ADDITIONAL_COMPOSE_QUEUE_ARGS="-f docker-compose.kafka.yml" ;; + confluent) + ADDITIONAL_COMPOSE_QUEUE_ARGS="-f docker-compose.confluent.yml" + ;; aws-sqs) ADDITIONAL_COMPOSE_QUEUE_ARGS="-f docker-compose.aws-sqs.yml" ;; @@ -52,7 +55,7 @@ function additionalComposeQueueArgs() { ADDITIONAL_COMPOSE_QUEUE_ARGS="-f docker-compose.service-bus.yml" ;; *) - echo "Unknown Queue service value specified: '${TB_QUEUE_TYPE}'. Should be either kafka or aws-sqs or pubsub or rabbitmq or service-bus." >&2 + echo "Unknown Queue service value specified: '${TB_QUEUE_TYPE}'. Should be either kafka or confluent or aws-sqs or pubsub or rabbitmq or service-bus." >&2 exit 1 esac echo $ADDITIONAL_COMPOSE_QUEUE_ARGS diff --git a/docker/docker-compose.confluent.yml b/docker/docker-compose.confluent.yml new file mode 100644 index 0000000000..1bbc3f96d6 --- /dev/null +++ b/docker/docker-compose.confluent.yml @@ -0,0 +1,57 @@ +# +# 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. +# + +version: '2.2' + +services: + tb-js-executor: + env_file: + - queue-confluent.env + tb-core1: + env_file: + - queue-confluent.env + depends_on: + - redis + tb-core2: + env_file: + - queue-confluent.env + depends_on: + - redis + tb-rule-engine1: + env_file: + - queue-confluent.env + depends_on: + - redis + tb-rule-engine2: + env_file: + - queue-confluent.env + depends_on: + - redis + tb-mqtt-transport1: + env_file: + - queue-confluent.env + tb-mqtt-transport2: + env_file: + - queue-confluent.env + tb-http-transport1: + env_file: + - queue-confluent.env + tb-http-transport2: + env_file: + - queue-confluent.env + tb-coap-transport: + env_file: + - queue-confluent.env diff --git a/docker/queue-confluent.env b/docker/queue-confluent.env new file mode 100644 index 0000000000..868a135de3 --- /dev/null +++ b/docker/queue-confluent.env @@ -0,0 +1,18 @@ +TB_QUEUE_TYPE=kafka + +TB_KAFKA_SERVERS=confluent.cloud:9092 +TB_QUEUE_KAFKA_REPLICATION_FACTOR=3 + +TB_QUEUE_KAFKA_USE_CONFLUENT_CLOUD=true +TB_QUEUE_KAFKA_CONFLUENT_SSL_ALGORITHM=https +TB_QUEUE_KAFKA_CONFLUENT_SASL_MECHANISM=PLAIN +TB_QUEUE_KAFKA_CONFLUENT_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="CLUSTER_API_KEY" password="CLUSTER_API_SECRET"; +TB_QUEUE_KAFKA_CONFLUENT_SECURITY_PROTOCOL=SASL_SSL +TB_QUEUE_KAFKA_CONFLUENT_USERNAME=CLUSTER_API_KEY +TB_QUEUE_KAFKA_CONFLUENT_PASSWORD=CLUSTER_API_SECRET + +TB_QUEUE_KAFKA_RE_TOPIC_PROPERTIES=retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000 +TB_QUEUE_KAFKA_CORE_TOPIC_PROPERTIES=retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000 +TB_QUEUE_KAFKA_TA_TOPIC_PROPERTIES=retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000 +TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES=retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000 +TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES=retention.ms:604800000;segment.bytes:52428800;retention.bytes:104857600 diff --git a/msa/js-executor/config/custom-environment-variables.yml b/msa/js-executor/config/custom-environment-variables.yml index c573274801..ccd33177fd 100644 --- a/msa/js-executor/config/custom-environment-variables.yml +++ b/msa/js-executor/config/custom-environment-variables.yml @@ -26,6 +26,12 @@ kafka: servers: "TB_KAFKA_SERVERS" replication_factor: "TB_QUEUE_KAFKA_REPLICATION_FACTOR" topic_properties: "TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES" + use_confluent_cloud: "TB_QUEUE_KAFKA_USE_CONFLUENT_CLOUD" + confluent: + sasl: + mechanism: "TB_QUEUE_KAFKA_CONFLUENT_SASL_MECHANISM" + username: "TB_QUEUE_KAFKA_CONFLUENT_USERNAME" + password: "TB_QUEUE_KAFKA_CONFLUENT_PASSWORD" pubsub: project_id: "TB_QUEUE_PUBSUB_PROJECT_ID" diff --git a/msa/js-executor/config/default.yml b/msa/js-executor/config/default.yml index f42b74745f..fac1ac7e8a 100644 --- a/msa/js-executor/config/default.yml +++ b/msa/js-executor/config/default.yml @@ -26,6 +26,10 @@ kafka: servers: "localhost:9092" replication_factor: "1" topic_properties: "retention.ms:604800000;segment.bytes:26214400;retention.bytes:104857600" + use_confluent_cloud: false + confluent: + sasl: + mechanism: "PLAIN" pubsub: queue_properties: "ackDeadlineInSec:30;messageRetentionInSec:604800" diff --git a/msa/js-executor/queue/kafkaTemplate.js b/msa/js-executor/queue/kafkaTemplate.js index 33637cee81..0420173188 100644 --- a/msa/js-executor/queue/kafkaTemplate.js +++ b/msa/js-executor/queue/kafkaTemplate.js @@ -61,15 +61,27 @@ function KafkaProducer() { const kafkaBootstrapServers = config.get('kafka.bootstrap.servers'); const requestTopic = config.get('request_topic'); + const useConfluent = config.get('kafka.use_confluent_cloud'); logger.info('Kafka Bootstrap Servers: %s', kafkaBootstrapServers); logger.info('Kafka Requests Topic: %s', requestTopic); - kafkaClient = new Kafka({ + let kafkaConfig = { brokers: kafkaBootstrapServers.split(','), - logLevel: logLevel.INFO, - logCreator: KafkaJsWinstonLogCreator - }); + logLevel: logLevel.INFO, + logCreator: KafkaJsWinstonLogCreator + }; + + if (useConfluent) { + kafkaConfig['sasl'] = { + mechanism: config.get('kafka.confluent.sasl.mechanism'), + username: config.get('kafka.confluent.username'), + password: config.get('kafka.confluent.password') + }; + kafkaConfig['ssl'] = true; + } + + kafkaClient = new Kafka(kafkaConfig); parseTopicProperties(); diff --git a/transport/coap/src/main/resources/tb-coap-transport.yml b/transport/coap/src/main/resources/tb-coap-transport.yml index c34dba7c90..f04a7559b8 100644 --- a/transport/coap/src/main/resources/tb-coap-transport.yml +++ b/transport/coap/src/main/resources/tb-coap-transport.yml @@ -69,6 +69,13 @@ queue: linger.ms: "${TB_KAFKA_LINGER_MS:1}" buffer.memory: "${TB_BUFFER_MEMORY:33554432}" replication_factor: "${TB_QUEUE_KAFKA_REPLICATION_FACTOR:1}" + use_confluent_cloud: "${TB_QUEUE_KAFKA_USE_CONFLUENT_CLOUD:false}" + confluent: + ssl.algorithm: "${TB_QUEUE_KAFKA_CONFLUENT_SSL_ALGORITHM:https}" + 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}" + other: 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}" @@ -76,6 +83,7 @@ queue: 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: + use_default_credential_provider_chain: "${TB_QUEUE_AWS_SQS_USE_DEFAULT_CREDENTIAL_PROVIDER_CHAIN:false}" access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}" secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}" region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" diff --git a/transport/http/src/main/resources/tb-http-transport.yml b/transport/http/src/main/resources/tb-http-transport.yml index 377c66e711..a97a58cd85 100644 --- a/transport/http/src/main/resources/tb-http-transport.yml +++ b/transport/http/src/main/resources/tb-http-transport.yml @@ -62,6 +62,13 @@ queue: linger.ms: "${TB_KAFKA_LINGER_MS:1}" buffer.memory: "${TB_BUFFER_MEMORY:33554432}" replication_factor: "${TB_QUEUE_KAFKA_REPLICATION_FACTOR:1}" + use_confluent_cloud: "${TB_QUEUE_KAFKA_USE_CONFLUENT_CLOUD:false}" + confluent: + ssl.algorithm: "${TB_QUEUE_KAFKA_CONFLUENT_SSL_ALGORITHM:https}" + 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}" + other: 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}" @@ -69,6 +76,7 @@ queue: 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: + use_default_credential_provider_chain: "${TB_QUEUE_AWS_SQS_USE_DEFAULT_CREDENTIAL_PROVIDER_CHAIN:false}" access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}" secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}" region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index bcc27d0755..b3804e3531 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -90,6 +90,13 @@ queue: linger.ms: "${TB_KAFKA_LINGER_MS:1}" buffer.memory: "${TB_BUFFER_MEMORY:33554432}" replication_factor: "${TB_QUEUE_KAFKA_REPLICATION_FACTOR:1}" + use_confluent_cloud: "${TB_QUEUE_KAFKA_USE_CONFLUENT_CLOUD:false}" + confluent: + ssl.algorithm: "${TB_QUEUE_KAFKA_CONFLUENT_SSL_ALGORITHM:https}" + 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}" + other: 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}" @@ -97,6 +104,7 @@ queue: 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: + use_default_credential_provider_chain: "${TB_QUEUE_AWS_SQS_USE_DEFAULT_CREDENTIAL_PROVIDER_CHAIN:false}" access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}" secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}" region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}"