diff options
| author | Dana Powers <dana.powers@rd.io> | 2015-06-05 23:09:50 -0700 |
|---|---|---|
| committer | Dana Powers <dana.powers@rd.io> | 2015-06-06 19:27:56 -0700 |
| commit | 2ba22bf4cebf5e25351816b38cd3cb70e2ea4cb8 (patch) | |
| tree | 554ccead53165a80588e94f4189f9928cece2e67 /kafka | |
| parent | c76bc9dd179044811a5e0d3fde2e437f1ee6d46c (diff) | |
| download | kafka-python-2ba22bf4cebf5e25351816b38cd3cb70e2ea4cb8.tar.gz | |
Dont stop async producer until all pending messages have been processed
Diffstat (limited to 'kafka')
| -rw-r--r-- | kafka/producer/base.py | 2 |
1 files changed, 1 insertions, 1 deletions
diff --git a/kafka/producer/base.py b/kafka/producer/base.py index a0bf47c..0fd742d 100644 --- a/kafka/producer/base.py +++ b/kafka/producer/base.py @@ -73,7 +73,7 @@ def _send_upstream(queue, client, codec, batch_time, batch_size, request_tries = {} client.reinit() - while not stop_event.is_set(): + while not (stop_event.is_set() and queue.empty() and not request_tries): timeout = batch_time count = batch_size send_at = time.time() + timeout |