diff --git a/msa/js-executor/config/custom-environment-variables.yml b/msa/js-executor/config/custom-environment-variables.yml index 8d2cf4b466..5e2bfe8342 100644 --- a/msa/js-executor/config/custom-environment-variables.yml +++ b/msa/js-executor/config/custom-environment-variables.yml @@ -27,6 +27,7 @@ kafka: replication_factor: "TB_QUEUE_KAFKA_REPLICATION_FACTOR" acks: "TB_KAFKA_ACKS" # -1 = all; 0 = no acknowledgments; 1 = only waits for the leader to acknowledge requestTimeout: "TB_QUEUE_KAFKA_REQUEST_TIMEOUT_MS" + compression: "TB_QUEUE_KAFKA_COMPRESSION" # gzip or uncompressed topic_properties: "TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES" use_confluent_cloud: "TB_QUEUE_KAFKA_USE_CONFLUENT_CLOUD" client_id: "KAFKA_CLIENT_ID" #inject pod name to easy identify the client using /opt/kafka/bin/kafka-consumer-groups.sh diff --git a/msa/js-executor/config/default.yml b/msa/js-executor/config/default.yml index 84e927378a..cb6b87d050 100644 --- a/msa/js-executor/config/default.yml +++ b/msa/js-executor/config/default.yml @@ -27,6 +27,7 @@ kafka: replication_factor: "1" acks: "1" # -1 = all; 0 = no acknowledgments; 1 = only waits for the leader to acknowledge 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" use_confluent_cloud: false client_id: "kafkajs" #inject pod name to easy identify the client using /opt/kafka/bin/kafka-consumer-groups.sh diff --git a/msa/js-executor/queue/kafkaTemplate.js b/msa/js-executor/queue/kafkaTemplate.js index fb56630471..2f4fd8751d 100644 --- a/msa/js-executor/queue/kafkaTemplate.js +++ b/msa/js-executor/queue/kafkaTemplate.js @@ -24,6 +24,7 @@ const topicProperties = config.get('kafka.topic_properties'); const kafkaClientId = config.get('kafka.client_id'); const acks = Number(config.get('kafka.acks')); const requestTimeout = Number(config.get('kafka.requestTimeout')); +const compressionType = (config.get('kafka.requestTimeout') === "gzip") ? CompressionTypes.GZIP : CompressionTypes.None; let kafkaClient; let kafkaAdmin; @@ -38,7 +39,7 @@ function KafkaProducer() { { topic: responseTopic, acks: acks, - compression: CompressionTypes.GZIP, + compression: compressionType, messages: [ { key: scriptId,