async producer has unbounded internal queue

[milyenpabo]

I'm using the provided example code for SimpleProducer to send out 1 kB messages in an infinite loop, as fast as possible. What I see is that I can pass ~60k messages/sec to the kafka-python library, although only ~1k messages are actually sent out towards the kafka brokers. In the meantime, resident memory usage of the producer is going up steadily and without an upper bound.

I'm new to kafka-python, and I'm not familiar with its iternals, but it seems as if the communication between my client code and kafka-python was async (regardless of instantiating the SimpleProcuder with async=False), and messages were queued internally by the kafka-python library in an unlimited queue.

Is that intutition right? Is there a way to use the library in a kind of "blocking" mode where memory usage is limited?

I'm using kafka-python v0.9.2 with two v0.8.2 kafka brokers, all nodes are connected with a 10 Gbps network.

[dpkp]

this sounds odd -- the SimpleProducer should operate in a sync / blocking mode by default. can you paste a test case to reproduce?

[ediskandarov]

I confirm that memory leak in producer with batch_send=True.

[ediskandarov]

I guess this happens because growing queue.
We have to add queue maxsize parameter.

[milyenpabo]

Thanks for the quick patch!

Though it might be that from a client point of view, blocking on send would be more desirable than raising an error. What do you think?

[ediskandarov]

Kafka is about a big data.

If you can not send your data to kafka it is better to spill data to disk and try to resend later.
In my opinion there are more cases where no time to wait.

[dpkp]

there is currently no bounds on the async producer's internal queue size. setting either async=True or batch_send=True uses async mode and will not block but instead add messages to the internal queue. If you add messages to the producer's queue faster than it can send them over the network to kafka then you will see memory increasing as the unsent queue grows.

honestly I would not recommend the current async producer code for production use -- it needs to be fixed for several issues (this one included). at my company we wrote our own async producer class that wraps the sync simpleproducer. I will see if we can get that merged back upstream.

[wizzat]

FWIW, async and multiprocessing both have a host of issues. I'm still
firmly convinced that the multiprocessing consumer is worse for performance
in every way possible.

On Fri, Jan 23, 2015 at 9:32 AM, Dana Powers notifications@github.com
wrote:

there is currently no bounds on the async producer's internal queue size.
setting either async=True or batch_send=True uses async mode and will not
block but instead add messages to the internal queue. If you add messages
to the producer's queue faster than it can send them over the network to
kafka then you will see memory increasing as the unsent queue grows.

honestly I would not recommend the current async producer code for
production use -- it needs to be fixed for several issues (this one
included). at my company we wrote our own async producer class that wraps
the sync simpleproducer. I will see if we can get that merged back upstream.

—
Reply to this email directly or view it on GitHub
#297 (comment).

[dpkp]

several PRs to fix this issue: #283, #304, #375 . will need to pick one and merge. I'm going to try to take a closer look tonight. if anyone has thoughts on the relative merits, please drop a note. Otherwise I'll just make an executive decision.

[dpkp]

#331 also addresses this issue

[dpkp]

#331 is merged and supports this via async_queue_maxsize and async_queue_put_timeout configuration parameters to the producer