kafka: added TB_KAFKA_COMPRESSION_TYPE for producer: none (default), gzip.

@@ -730,6 +730,7 @@ queue:
     bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}"
     bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}"
     acks: "${TB_KAFKA_ACKS:all}"
     acks: "${TB_KAFKA_ACKS:all}"
     retries: "${TB_KAFKA_RETRIES:1}"
     retries: "${TB_KAFKA_RETRIES:1}"
+    compression.type: "${TB_KAFKA_COMPRESSION_TYPE:none}" # none or gzip
     batch.size: "${TB_KAFKA_BATCH_SIZE:16384}"
     batch.size: "${TB_KAFKA_BATCH_SIZE:16384}"
     linger.ms: "${TB_KAFKA_LINGER_MS:1}"
     linger.ms: "${TB_KAFKA_LINGER_MS:1}"
     buffer.memory: "${TB_BUFFER_MEMORY:33554432}"
     buffer.memory: "${TB_BUFFER_MEMORY:33554432}"

@@ -54,6 +54,9 @@ public class TbKafkaSettings {
     @Value("${queue.kafka.retries}")
     @Value("${queue.kafka.retries}")
     private int retries;
     private int retries;
 
 
+    @Value("${queue.kafka.compression.type:none}")
+    private String compressionType;
+
     @Value("${queue.kafka.batch.size}")
     @Value("${queue.kafka.batch.size}")
     private int batchSize;
     private int batchSize;
 
 
@@ -135,6 +138,7 @@ public class TbKafkaSettings {
         props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
         props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionType);
         return props;
         return props;
     }
     }