squash finalise warnign

Author: Martin Durant

streamz/sources.py

@@ -447,7 +447,7 @@ def start(self):
             self.stopped = False
             self.consumer = ck.Consumer(self.cpars)
             self.consumer.subscribe(self.topics)
-            weakref.finalize(self, self.consumer.close)
+            weakref.finalize(self, lambda consumer=self.consumer: _close_consumer(consumer))
             tp = ck.TopicPartition(self.topics[0], 0, 0)
 
             # blocks for consumer thread to come up
@@ -463,6 +463,13 @@ def _close_consumer(self):
         self.stopped = True
 
 
+def _close_consumer(consumer):
+    try:
+        consumer.close()
+    except RuntimeError:
+        pass
+
+
 class FromKafkaBatched(Source):
     """Base class for both local and cluster-based batched kafka processing"""
     def __init__(self, topic, consumer_params, poll_interval='1s',
@@ -574,7 +581,7 @@ def start(self):
                 self.consumer = kafka.Consumer(self.consumer_params)
             else:
                 self.consumer = ck.Consumer(self.consumer_params)
-            weakref.finalize(self, self.consumer.close)
+            weakref.finalize(self, lambda consumer=self.consumer: _close_consumer(consumer))
             self.stopped = False
             tp = ck.TopicPartition(self.topic, 0, 0)
 

streamz/tests/test_sources.py

@@ -13,8 +13,8 @@ def test_periodic():
     l = s.sink_to_list()
     assert s.stopped
     s.start()
-    wait_for(lambda: l, 0.11, period=0.01)
-    wait_for(lambda: len(l) > 1, 0.11, period=0.01)
+    wait_for(lambda: l, 0.3, period=0.01)
+    wait_for(lambda: len(l) > 1, 0.3, period=0.01)
     assert all(l)