Fix possible request draining in ensure_active_group (#896)

kafka/client_async.py

@@ -453,7 +453,7 @@ class KafkaClient(object):
 
         return self._conns[node_id].send(request, expect_response=expect_response)
 
-    def poll(self, timeout_ms=None, future=None, sleep=True):
+    def poll(self, timeout_ms=None, future=None, sleep=True, delayed_tasks=True):
         """Try to read and write to sockets.
 
         This method will also attempt to complete node connections, refresh
@@ -488,14 +488,15 @@ class KafkaClient(object):
             metadata_timeout_ms = self._maybe_refresh_metadata()
 
             # Send scheduled tasks
-            for task, task_future in self._delayed_tasks.pop_ready():
-                try:
-                    result = task()
-                except Exception as e:
-                    log.error("Task %s failed: %s", task, e)
-                    task_future.failure(e)
-                else:
-                    task_future.success(result)
+            if delayed_tasks:
+                for task, task_future in self._delayed_tasks.pop_ready():
+                    try:
+                        result = task()
+                    except Exception as e:
+                        log.error("Task %s failed: %s", task, e)
+                        task_future.failure(e)
+                    else:
+                        task_future.success(result)
 
             # If we got a future that is already done, don't block in _poll
             if future and future.is_done:

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)