(Attempt to) Fix deadlock between consumer and heartbeat (#1628)

Author: dpkp

kafka/client_async.py

@@ -571,9 +571,7 @@ def poll(self, timeout_ms=None, future=None):
 
                 self._poll(timeout)
 
-            # called without the lock to avoid deadlock potential
-            # if handlers need to acquire locks
-            responses.extend(self._fire_pending_completed_requests())
+                responses.extend(self._fire_pending_completed_requests())
 
             # If all we had was a timeout (future is None) - only do one poll
             # If we do have a future, we keep looping until it is done

kafka/coordinator/base.py

@@ -347,7 +347,7 @@ def _handle_join_failure(self, _):
 
     def ensure_active_group(self):
         """Ensure that the group is active (i.e. joined and synced)"""
-        with self._lock:
+        with self._client._lock, self._lock:
             if self._heartbeat_thread is None:
                 self._start_heartbeat_thread()