Stopping daemonized threads when references to these threads are lost

Author: chmduquesne

kafka/producer/base.py

@@ -212,3 +212,9 @@ def stop(self, timeout=1):
 
             if self.proc.is_alive():
                 self.proc.terminate()
+
+    def __del__(self):
+        try:
+            self.stop()
+        except BrokenPipeError: # Already stopped
+            pass

kafka/util.py

@@ -151,3 +151,6 @@ def stop(self):
         # noinspection PyAttributeOutsideInit
         self.timer = None
         self.fn = None
+
+    def __del__(self):
+        self.stop()