Problem when sending huge message blocks to poseidon 

I experienced a problem when using the out_kafka_buffered plugin in my fluentd environment.  

It was caused by many messages (I use the forest plugin so effectively only a single chunk is destined for a single topic) being sent in a send_messages call.  It was on the order of 15k messages which added up to about 8MB of serialized data per invocation.

When this happened all sorts of bad things started happening in fluent, particularly my forward input source wouldn't/couldn't keep up, and as a whole the entire fluentd process bogged down and eventually wouldn't be able to shutdown cleanly.  I usually gave up for the send_messages call to return after about 5 minutes.

During the process of fixing this problem, I disabled all aggregation of records and invoked send_messages per record, and this would allow me to finish the above chunk in about 5 seconds.  

Using the buffer_chunk_limit to control the chunking wasn't going to work for me due to the possibility of an enormous number of files that would be created in the event my kafka server went down.  I felt aggregation of some records is nice (and doesn't need to be per topic), so I added the parameter kafka_agg_max_bytes to control this and defaulted to about 4KB.  If a single serialized record exceeds this value, the single record is sent.

I also cleaned up the producer per topic (not needed).

Lastly, I wanted to point out I attempted to use the :async poseidon producer to fix this problem, but it is "Not Implemented Yet".