title: Kafka doesn't compress in snappy
body: <p>I'm curious and I want to check if the snappy compression works well with the java Kafka client. </p>
  
  <p>In order to handle this, I set up a small program. This program generate 1024 messages readable data. Their size are of 1024 byte. I send these messages on tree new topics and after I check the size of these topic directly on the broker filesystem. </p>
  
  <p>You can find this program through the following code :</p>
  
  <pre><code>package unit_test.testCompress;
  
  import java.util.HashMap;
  import java.util.Map;
  import java.util.Random;
  import java.util.concurrent.Future;
  
  import org.apache.kafka.clients.producer.KafkaProducer;
  import org.apache.kafka.clients.producer.ProducerRecord;
  import org.apache.kafka.clients.producer.RecordMetadata;
  
  
  /**
   * Can be use in order to execute some unit test on compression  
   */
  public class TestCompress {
  
      public static void compress(String type, String version){
          Map&lt;String,Object&gt; configs = new HashMap&lt;String,Object&gt;();
          configs.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
          configs.put("producer.type", "async");
          configs.put("compression.type", type);
          configs.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
          configs.put("partitioner.class", "com.kafkaproducer.RecordPartitioner");
          configs.put("bootstrap.servers", "kafka:9092");
  
  
          KafkaProducer&lt;String, byte[]&gt; producer = new KafkaProducer&lt;String, byte[]&gt;(configs);
  
          Random r = new Random(15415485);
          int size = 1024; //1 Ko
          byte[] buffer = new byte[size];
          for(int i = 0; i &lt; size; i++){
              buffer[i] = (byte) ('A' + (r.nextInt() % 26));
          }
          buffer[size-1] = 0;
          //System.out.println(new String(buffer));
          for(int i = 0; i &lt; size; i++ ){
              Future&lt;RecordMetadata&gt; result = producer.send( new ProducerRecord&lt;String, byte[]&gt;("unit_test_compress_"+version+ "_" + type , buffer));
          }
  
          producer.close();
      }
  
      public static void main(String[] args) {
  
          String version = "v10";
          compress("snappy",version);
          compress("gzip",version);
          compress("none",version);
  
      }
  
  }
  </code></pre>
  
  <p>I'm compiling this code with this following maven pom file :</p>
  
  <pre><code>    &lt;project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"&gt;
    &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
  
    &lt;groupId&gt;unit_test&lt;/groupId&gt;
    &lt;artifactId&gt;testCompress&lt;/artifactId&gt;
    &lt;version&gt;0.0.1-SNAPSHOT&lt;/version&gt;
    &lt;packaging&gt;jar&lt;/packaging&gt;
  
    &lt;name&gt;testCompress&lt;/name&gt;
    &lt;url&gt;http://maven.apache.org&lt;/url&gt;
  
    &lt;properties&gt;
      &lt;project.build.sourceEncoding&gt;UTF-8&lt;/project.build.sourceEncoding&gt;
    &lt;/properties&gt;
  
    &lt;dependencies&gt;  
       &lt;dependency&gt;
          &lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
          &lt;artifactId&gt;kafka_2.10&lt;/artifactId&gt;
          &lt;version&gt;0.8.2.2&lt;/version&gt;      
      &lt;/dependency&gt;
    &lt;/dependencies&gt;
  &lt;/project&gt;
  </code></pre>
  
  <p>This program executes very well on my computer. </p>
  
  <p>But when I check the results directly on my kafka broker it give the following output : 
  <a href="https://i.stack.imgur.com/7W1f5.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/7W1f5.png" alt="enter image description here"></a></p>
  
  <p>I think that's mean there are no compression on snappy topic (but gzip compression works very well). I checked though vi the file </p>
  
  <p>I'm aware about this issue on Kafka 8.2.1 : <a href="https://issues.apache.org/jira/browse/KAFKA-2189" rel="nofollow noreferrer">https://issues.apache.org/jira/browse/KAFKA-2189</a> 
  But I'm using Kafka 8.2.2 on producer and 8.2.1 on broker. I checked the dependency of Snappy as well. I'm using the 1.1.1.7</p>
  
  <p>Have you an idea of how to enable snappy compression on Kafak ? Did I forget a parameter to enable snappy compression on kafka ? </p>