Develop/2.5.3 confluent cloud (#3259)

* added other parameters for queue kafka * Added support Confluent Cloud * fix js executor kafka connection * refactored
6 years ago · e318b193bd
11 changed files with 166 additions and 20 deletions
--- 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}"
--- 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<TbKafkaProperty> 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;
--- 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
--- a/docker/docker-compose.confluent.yml
+++ 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
--- a/docker/queue-confluent.env
+++ 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
--- 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"
--- 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"
--- 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();

--- 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}"
--- 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}"
--- 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}"