Use weakref when registering a producer.close atexit to fix normal gc (#728)

* Use weakref when registering a producer.close atexit to fix normal gc
* Test that del(producer) terminates async thread

1 files changed, 14 insertions, 0 deletions

diff --git a/test/test_producer.py b/test/test_producer.py
index f11bb05..125737b 100644
--- a/test/test_producer.py
+++ b/test/test_producer.py
@@ -1,4 +1,7 @@
+import gc
+import platform
 import sys
+import threading
 
 import pytest
 
@@ -64,3 +67,14 @@ def test_end_to_end(kafka_broker, compression):
             break
 
     assert msgs == set(['msg %d' % i for i in range(messages)])
+
+
+@pytest.mark.skipif(platform.python_implementation() != 'CPython',
+                    reason='Test relies on CPython-specific gc policies')
+def test_kafka_producer_gc_cleanup():
+    threads = threading.active_count()
+    producer = KafkaProducer(api_version='0.9') # set api_version explicitly to avoid auto-detection
+    assert threading.active_count() == threads + 1
+    del(producer)
+    gc.collect()
+    assert threading.active_count() == threads