Handling Large Messages in Kafka

Kafka is well optimized for small messages. Around 10KB seem to be the sweet spot for getting the best throughput (as seen in LinkedIn’s famous benchmark). But sometimes, we need to send larger messages – XML documents or even JSON can get a bit bulky and we end up with messages in the 10M-100M range. What do we do in this case?

Here are few suggestions:

  1. The best way to send large messages is not to send them at all. If shared storage is available (NAS, HDFS, S3), placing large files on the shared storage and using Kafka just to send a message about the file’s location is a much better use of Kafka.
  2. The second best way to send large messages is to slice and dice them. Use the producing client to split the message into small 10K portions, use partition key to make sure all the portions will be sent to the same Kafka partition (so the order will be preserved) and have the consuming client sew them back up into a large message.
  3. Kafka producer can be used to compress messages. If the original message is XML, there’s a good chance that the compressed message will not be very large at all. Use compression.codec and compressed.topics configuration parameters in the producer to enable compression. GZip and Snappy are both supported.

But what if you really need to use Kafka with large messages? Well, in this case, there are few configuration parameters you need to set:

broker configs:

  • message.max.bytes (default:1000000) – Maximum size of a message the broker will accept. This has to be smaller than the consumer fetch.message.max.bytes, or the broker will have messages that can’t be consumed, causing consumers to hang.
  • log.segment.bytes (default: 1GB) – size of a Kafka data file. Make sure its larger than 1 message. Default should be fine (i.e. large messages probably shouldn’t exceed 1GB in any case. Its a messaging system, not a file system)
  • replica.fetch.max.bytes (default: 1MB) – Maximum size of data that a broker can replicate. This has to be larger than message.max.bytes, or
    a broker will accept messages and fail to replicate them. Leading to potential data loss.

Consumer configs:

  • fetch.message.max.bytes (default 1MB) – Maximum size of message a consumer can read. This should be the size of message.max.bytes or larger.

If you choose to configure Kafka for large messages, there are few things you need to take into consideration. Large messages are not an afterthought, they have impact on overall design of the cluster and the topics:

  1. Performance: As we’ve mentioned, benchmarks indicate that Kafka reaches maximum throughput with message size of around 10K. Larger messages will show decreased throughput. Its important to remember this when doing capacity planning for a cluster.
  2. Available memory and number of partitions: Brokers will need to allocate a buffer the size of replica.fetch.max.bytes for each partition they replicate. So if replica.fetch.max.bytes = 1MB and you have 1000 partitions, that will take around 1GB of RAM. Do the math and make sure the number of partitions * the size of the largest message does not exceed available memory, or you’ll see OOM errors. Same for consumers and fetch.message.max.bytes – make sure there’s enough memory for the largest message for each partition the consumer replicates. This may mean that you end up with fewer partitions if you have large messages, or you may need servers with more RAM.
  3. Garbage collection - I did not personally see this issue, but I think its a reasonable concern. Large messages may cause longer garbage collection pauses (as brokers need to allocate large chunks), keep an eye on the GC log and on the server log. If long GC pauses cause Kafka to lose the zookeeper session, you may need to configure longer timeout values for zookeeper.session.timeout.ms.

Weight the pros, cons and headaches – and decide which solution works best for you.

Ran into additional issues with large messages? Know about parameters that I missed? Let us know in the comments.

 

Tweet about this on TwitterShare on FacebookShare on LinkedIn

'Handling Large Messages in Kafka' have 5 comments

  1. January 23, 2015 @ 6:07 pm uwo

    Thanks for this! At what size would you say that messages are too large and Kafka is not a good solution, and is there a general rule of thumb for message queues and an upper bound on message size? Also, if messages are consistently large, but still within what’s reasonable, do you have any other technologies you’d recommend evaluating?

    Reply

  2. January 26, 2015 @ 1:16 pm Hari Sekhon

    Can you share some practical examples where messages would be this large, 10-100MB+?

    Seems like an application paradigm problem, my first thought to solve is also option 1 or fix the application rather than just increasing the buffers.

    Regards,

    Hari Sekhon

    Reply

    • August 13, 2015 @ 6:11 pm Aniruddha Jawanjal

      One real time example would be mail aggregation client which also gets attachment of different size in the queue, where Kafka is used for HA & distributed architecture.

      Reply

  3. May 15, 2015 @ 1:20 am Rajesh Datla

    Great

    Great

    How to integrate Kafka with Hadoop using parquet file format, is it possible to store messages in Hbase and hive using parquet format or we need to write our own parquet custom input file format, record reader

    Reply

  4. October 2, 2015 @ 1:18 pm Rahul Jain

    Thanks for this very informative post.
    I tried this and realized that if you are using the new producer then the property “max.request.size” also needs to be set accordingly.

    Reply


Would you like to share your thoughts?

Your email address will not be published.