Fix wrong configuration key for Compression Type

Fix #4678

(cherry picked from commit c176ca94)

@@ -26,7 +26,7 @@ const acks = Number(config.get('kafka.acks'));
 const maxBatchSize = Number(config.get('kafka.batch_size'));
 const maxBatchSize = Number(config.get('kafka.batch_size'));
 const linger = Number(config.get('kafka.linger_ms'));
 const linger = Number(config.get('kafka.linger_ms'));
 const requestTimeout = Number(config.get('kafka.requestTimeout'));
 const requestTimeout = Number(config.get('kafka.requestTimeout'));
-const compressionType = (config.get('kafka.requestTimeout') === "gzip") ? CompressionTypes.GZIP : CompressionTypes.None;
+const compressionType = (config.get('kafka.compression') === "gzip") ? CompressionTypes.GZIP : CompressionTypes.None;
 
 
 let kafkaClient;
 let kafkaClient;
 let kafkaAdmin;
 let kafkaAdmin;