Async producer should wait for messages to flush before closing

[santosh-d3vpl3x]

As per code in base.py, a shutdown method is registered with atexit.register(cleanup, self) statement. This registered method does not pass any timeout effectively causing data loss. Producer.close with custom timeout value works properly but most of the time when application receives exit signal, _run_exitfuncs calls close on producer before any other close call with set timeout..

We should allow users to set custom value of timeout in registered cleanup method allowing data to flush in Kafka or if we can wait for queue to drain then that would be most efficient approach.

[dpkp]

Have you tried the Producer option async_stop_timeout ? This timeout parameter is passed into the async thread when the producer is created. Although it might be useful in some use cases to not have to pick a timeout at the time the producer is started, I think in most cases the desired timeout will be known ahead of time and can be configured with the existing interface.

See http://kafka-python.readthedocs.org/en/latest/apidoc/kafka.producer.html#module-kafka.producer.base for more detail on the options

[santosh-d3vpl3x]

Ummm,while examining the code, I couldn't see async_stop_timeout parameter acting in someway to delay shutdown of thread. As of now, atexit handlers don't depend on such parameter. Please correct me if I am wrong. SIGTERM to python process directly terminates the thread after waiting for 1 second which is stop functions default timeout. If what you are saying is true, then there must be something wrong with the way I am trying to do the things or api implementation.

[dpkp]

Ah - I see the issue now, thanks!

The problem is that stop() doesn't block until the async thread has completed. The timeout parameter for stop() is a bit confusing here. I'll try to throw up a patch this weekend.

[santosh-d3vpl3x]

Thanks! I have extended base.py and modified close in such a way that blocking time depends on number of messages in queue and configuration of Producer thread.

Calculation to determine blocking time is very dumb at the moment. It depends on batch_size[batch_send_every_n ] and interval[batch_send_every_t].

EDIT: I could not grasp your comment entirely.

doesn't block until the async thread has completed

Could you please let me know what exactly completed mean in this statement.

[dpkp]

by completed, I mean that the _send_upstream thread function returns. that function will loop until the following 3 conditions are all met:

(1) thread stop_event is set [caused by producer.stop()]
(2) async message queue is empty
(3) no messages are pending retry

https://github.com/dpkp/kafka-python/blob/master/kafka/producer/base.py#L92

[santosh-d3vpl3x]

Thanks for elaboration!

[dpkp]

I should add that (2) and (3) are subject to the producer async_stop_timeout . If all messages cannot be delivered before that timeout, the thread will drop messages and return.

[dpkp]

This should be fixed in master now. Please reopen if the problem persists!