js-executor: parameter added for Kafka PARTITIONS_CONSUMED_CONCURRENTLY to decrease max latency while scale down replicas

msa/js-executor/config/custom-environment-variables.yml

@@ -29,6 +29,7 @@ kafka:
   acks: "TB_KAFKA_ACKS" # -1 = all; 0 = no acknowledgments; 1 = only waits for the leader to acknowledge
   batch_size: "${TB_KAFKA_BATCH_SIZE:128}" # for producer
   linger_ms: "${TB_KAFKA_LINGER_MS:1}" # for producer
+  partitions_consumed_concurrently: "${PARTITIONS_CONSUMED_CONCURRENTLY:1}" # increase this value if you are planning to handle more than one partition (scale up, scale down) - this will decrease the latency
   requestTimeout: "TB_QUEUE_KAFKA_REQUEST_TIMEOUT_MS"
   compression: "TB_QUEUE_KAFKA_COMPRESSION" # gzip or uncompressed
   topic_properties: "TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES"

msa/js-executor/config/default.yml

@@ -29,6 +29,7 @@ kafka:
   acks: "1" # -1 = all; 0 = no acknowledgments; 1 = only waits for the leader to acknowledge
   batch_size: "128" # for producer
   linger_ms: "1" # for producer
+  partitions_consumed_concurrently: "1" # increase this value if you are planning to handle more than one partition (scale up, scale down) - this will decrease the latency
   requestTimeout: "30000" # The default value in kafkajs is: 30000
   compression: "gzip" # gzip or uncompressed
   topic_properties: "retention.ms:604800000;segment.bytes:26214400;retention.bytes:104857600;partitions:100;min.insync.replicas:1"

msa/js-executor/queue/kafkaTemplate.js

@@ -27,6 +27,7 @@ const maxBatchSize = Number(config.get('kafka.batch_size'));
 const linger = Number(config.get('kafka.linger_ms'));
 const requestTimeout = Number(config.get('kafka.requestTimeout'));
 const compressionType = (config.get('kafka.compression') === "gzip") ? CompressionTypes.GZIP : CompressionTypes.None;
+const partitionsConsumedConcurrently = Number(config.get('kafka.partitions_consumed_concurrently'));
 
 let kafkaClient;
 let kafkaAdmin;
@@ -197,7 +198,7 @@ async function sendMessagesAsBatch(isImmediately) {
 
         logger.info('Started ThingsBoard JavaScript Executor Microservice.');
         await consumer.run({
-            //partitionsConsumedConcurrently: 1, // Default: 1
+            partitionsConsumedConcurrently: partitionsConsumedConcurrently,
             eachMessage: async ({topic, partition, message}) => {
                 let headers = message.headers;
                 let key = message.key;