title: How do I set the size of messages in Kafka?
body: <p>I'm currently using Kafka 0.9.0.1. According to some sources I've found, the way to set the sizes of messages is to modify the following key values in <code>server.properties</code>.</p>
  
  <ul>
  <li>message.max.bytes</li>
  <li>replica.fetch.max.bytes</li>
  <li>fetch.message.max.bytes</li>
  </ul>
  
  <p>My <code>server.properties</code> file actually has these settings.</p>
  
  <pre><code>message.max.bytes=10485760
  replica.fetch.max.bytes=20971520
  fetch.message.max.bytes=10485760
  </code></pre>
  
  <p>Other settings that may be relevant are below.</p>
  
  <pre><code>socket.send.buffer.bytes=102400
  socket.receive.buffer.bytes=102400
  socket.request.max.bytes=104857600
  </code></pre>
  
  <p>However, when I attempt to send messages with payloads of 4 to 6 MB in size, the consumer never gets any messages. The producer seems to send the messages without any exceptions being thrown. If I do send smaller payloads (like &lt; 1 MB) then the consumer does actually receive the messages.</p>
  
  <p>Any idea on what I'm doing wrong in terms of configuration settings?</p>
  
  <p>Here is the example code to send a message.</p>
  
  <pre><code>Producer&lt;String, byte[]&gt; producer = new KafkaProducer&lt;&gt;(getProducerProps());
  File dir = new File("/path/to/dir");
  for(String s : dir.list()) {
    File f = new File(dir, s);
    byte[] data = Files.readAllBytes(f.toPath());
    Payload payload = new Payload(data); //a simple pojo to store payload
    String key = String.valueOf(System.currentTimeMillis());
    byte[] val = KryoUtil.toBytes(payload); //custom util to use kryo to get bytes[]
    producer.send(new ProducerRecord&lt;&gt;("test", key, val));
  }
  producer.close();
  </code></pre>
  
  <p>Here is the example code to receive a message.</p>
  
  <pre><code>KafkaConsumer consumer = new KafkaConsumer&lt;&gt;(getConsumerProps());
  consumer.subscribe(Arrays.asList("test"));
  while(true) {
    ConsumerRecord&lt;String, byte[]&gt; records = consumer.poll(100);
    for(ConsumerRecord&lt;String, byte[]&gt; record : records) {
      long offset = record.offset();
      String key = record.key();
      byte[] val = record.value();
      Payload payload = (Payload)KryoUtil.toObject(val, Payload.class); //custom util to use kryo to deserialize back to object
      System.out.println(
        System.format("offset=%d, key=%s", offset, key));
    }
  }
  </code></pre>
  
  <p>Here are the methods to populate the properties files for the producer and consumer.</p>
  
  <pre><code>public static Properties getProducerProps() {
    Properties props = new Properties();
    props.put("bootstrap.servers", "qc1:9092,qc2:9092,qc3:9092,qc4:9092");
    props.put("acks", "all");
    props.put("retries", 0);
    props.put("batch.size", 16384);
    props.put("linger.ms", 1);
    props.put("buffer.memory", 33554432);
    props.put("compression.type", "snappy");
    props.put("max.request.size", 10485760); //need this
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
    return props;
  }
  
  public static Properties getConsumerProps() {
    Properties props = new Properties();
    props.put("bootstrap.servers", "qc1:9092,qc2:9092,qc3:9092,qc4:9092");
    props.put("group.id", "test");
    props.put("enable.auto.commit", "true");
    props.put("auto.commit.interval.ms", "1000");
    props.put("session.timeout.ms", "30000");
    props.put("max.partition.fetch.bytes", 10485760); //need this too
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
    return props;
  }
  </code></pre>