break up some circular references and close client wake pipe on __del__

2 files changed, 21 insertions, 6 deletions

diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py
index 6efdfd0..c49c38b 100644
--- a/kafka/coordinator/base.py
+++ b/kafka/coordinator/base.py
@@ -2,6 +2,7 @@ import abc
 import copy
 import logging
 import time
+import weakref
 
 import six
 
@@ -85,9 +86,12 @@ class BaseCoordinator(object):
         self.rejoin_needed = True
         self.needs_join_prepare = True
         self.heartbeat = Heartbeat(**self.config)
-        self.heartbeat_task = HeartbeatTask(self)
+        self.heartbeat_task = HeartbeatTask(weakref.proxy(self))
         #self.sensors = GroupCoordinatorMetrics(metrics, metric_group_prefix, metric_tags)
 
+    def __del__(self):
+        self.heartbeat_task.disable()
+
     @abc.abstractmethod
     def protocol_type(self):
         """
@@ -572,6 +576,12 @@ class HeartbeatTask(object):
         self._client = coordinator._client
         self._request_in_flight = False
 
+    def disable(self):
+        try:
+            self._client.unschedule(self)
+        except KeyError:
+            pass
+
     def reset(self):
         # start or restart the heartbeat task to be executed at the next chance
         self._heartbeat.reset_session_timeout()
diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index 515377a..d63d052 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -4,20 +4,20 @@ import copy
 import collections
 import logging
 import time
+import weakref
 
 import six
 
 from .base import BaseCoordinator
 from .assignors.range import RangePartitionAssignor
 from .assignors.roundrobin import RoundRobinPartitionAssignor
-from .protocol import (
-    ConsumerProtocolMemberMetadata, ConsumerProtocolMemberAssignment,
-    ConsumerProtocol)
+from .protocol import ConsumerProtocol
 from ..common import OffsetAndMetadata, TopicPartition
 from ..future import Future
 from ..protocol.commit import (
     OffsetCommitRequest_v2, OffsetCommitRequest_v1, OffsetCommitRequest_v0,
     OffsetFetchRequest_v0, OffsetFetchRequest_v1)
+from ..util import WeakMethod
 
 import kafka.common as Errors
 
@@ -83,7 +83,7 @@ class ConsumerCoordinator(BaseCoordinator):
         self._partitions_per_topic = {}
         self._cluster = client.cluster
         self._cluster.request_update()
-        self._cluster.add_listener(self._handle_metadata_update)
+        self._cluster.add_listener(WeakMethod(self._handle_metadata_update))
 
         self._auto_commit_task = None
         if self.config['enable_auto_commit']:
@@ -95,13 +95,18 @@ class ConsumerCoordinator(BaseCoordinator):
                 log.warning('group_id is None: disabling auto-commit.')
             else:
                 interval = self.config['auto_commit_interval_ms'] / 1000.0
-                self._auto_commit_task = AutoCommitTask(self, interval)
+                self._auto_commit_task = AutoCommitTask(weakref.proxy(self), interval)
 
         # metrics=None,
         # metric_group_prefix=None,
         # metric_tags=None,
         # self.sensors = ConsumerCoordinatorMetrics(metrics, metric_group_prefix, metric_tags)
 
+    def __del__(self):
+        if self._auto_commit_task:
+            self._auto_commit_task.disable()
+        self._cluster.remove_listener(WeakMethod(self._handle_metadata_update))
+
     def protocol_type(self):
         return ConsumerProtocol.PROTOCOL_TYPE