From d2f136073cac0c8379f357cd76b0ea163fd22a99 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Sun, 10 Jan 2016 19:53:07 -0800 Subject: Receive all available responses in client._poll --- kafka/client_async.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/kafka/client_async.py b/kafka/client_async.py index 3a1922e..88b8ec6 100644 --- a/kafka/client_async.py +++ b/kafka/client_async.py @@ -354,11 +354,12 @@ class KafkaClient(object): ready, _, _ = select.select(list(sockets.keys()), [], [], timeout) responses = [] - # list, not iterator, because inline callbacks may add to self._conns for sock in ready: conn = sockets[sock] - response = conn.recv() # Note: conn.recv runs callbacks / errbacks - if response: + while conn.in_flight_requests: + response = conn.recv() # Note: conn.recv runs callbacks / errbacks + if not response: + break responses.append(response) return responses -- cgit v1.2.1