diff options
| author | Dana Powers <dana.powers@gmail.com> | 2018-11-10 12:45:01 -0800 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2018-11-10 12:45:01 -0800 |
| commit | 0a2ccba3cb1b8636f615a30821123720773a8dfa (patch) | |
| tree | 41c244e136c3f02c3d811019f8da90b1cfd6fb34 | |
| parent | 9d44e3cf8b6a75b11f1087282157aa48eba68a64 (diff) | |
| download | kafka-python-0a2ccba3cb1b8636f615a30821123720773a8dfa.tar.gz | |
(Attempt to) Fix deadlock between consumer and heartbeat (#1628)
| -rw-r--r-- | kafka/client_async.py | 4 | ||||
| -rw-r--r-- | kafka/coordinator/base.py | 2 |
2 files changed, 2 insertions, 4 deletions
diff --git a/kafka/client_async.py b/kafka/client_async.py index ccf1e4b..0cb575c 100644 --- a/kafka/client_async.py +++ b/kafka/client_async.py @@ -571,9 +571,7 @@ class KafkaClient(object): 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 diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py index 7deeaf0..8ce9a24 100644 --- a/kafka/coordinator/base.py +++ b/kafka/coordinator/base.py @@ -347,7 +347,7 @@ class BaseCoordinator(object): 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() |