Minor test cleanups (#1343)

Author: dpkp

test/test_consumer_group.py

@@ -68,8 +68,8 @@ def consumer_thread(i):
             for tp, records in six.itervalues(consumers[i].poll(100)):
                 messages[i][tp].extend(records)
         consumers[i].close()
-        del consumers[i]
-        del stop[i]
+        consumers[i] = None
+        stop[i] = None
 
     num_consumers = 4
     for i in range(num_consumers):
@@ -134,6 +134,7 @@ def consumer_thread(i):
             logging.info('Stopping consumer %s', c)
             stop[c].set()
             threads[c].join()
+            threads[c] = None
 
 
 @pytest.mark.skipif(not version(), reason="No KAFKA_VERSION set")

test/test_coordinator.py

@@ -400,6 +400,7 @@ def patched_coord(mocker, coordinator):
                         return_value=1)
     mocker.patch.object(coordinator._client, 'ready', return_value=True)
     mocker.patch.object(coordinator._client, 'send')
+    mocker.patch.object(coordinator, '_heartbeat_thread')
     mocker.spy(coordinator, '_failed_request')
     mocker.spy(coordinator, '_handle_offset_commit_response')
     mocker.spy(coordinator, '_handle_offset_fetch_response')

test/test_producer.py

@@ -72,6 +72,7 @@ def test_end_to_end(kafka_broker, compression):
 @pytest.mark.skipif(platform.python_implementation() != 'CPython',
                     reason='Test relies on CPython-specific gc policies')
 def test_kafka_producer_gc_cleanup():
+    gc.collect()
     threads = threading.active_count()
     producer = KafkaProducer(api_version='0.9') # set api_version explicitly to avoid auto-detection
     assert threading.active_count() == threads + 1