Fix possible request draining in ensure_active_group (#896)

1 files changed, 6 insertions, 3 deletions

diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py
index 22dffb4..e4ebcb0 100644
--- a/kafka/coordinator/base.py
+++ b/kafka/coordinator/base.py
@@ -246,9 +246,12 @@ class BaseCoordinator(object):
             # This is important in particular to avoid resending a pending
             # JoinGroup request.
             if self._client.in_flight_request_count(self.coordinator_id):
-                while self._client.in_flight_request_count(self.coordinator_id):
-                    self._client.poll()
-                continue
+                while not self.coordinator_unknown():
+                    self._client.poll(delayed_tasks=False)
+                    if not self._client.in_flight_request_count(self.coordinator_id):
+                        break
+                else:
+                    continue
 
             future = self._send_join_group_request()
             self._client.poll(future=future)