diff options
| author | Dana Powers <dana.powers@gmail.com> | 2018-02-16 11:54:44 -0800 |
|---|---|---|
| committer | Dana Powers <dana.powers@rd.io> | 2018-03-07 15:51:26 -0800 |
| commit | 3ccd6638f03d64d30ecc4e17014f51387804d974 (patch) | |
| tree | 2dc0ad56e45c0c855c590bb49ff0dcd9d423b7e2 | |
| parent | ff13f872f4a517c341cd84db89111dcbdf642b60 (diff) | |
| download | kafka-python-3ccd6638f03d64d30ecc4e17014f51387804d974.tar.gz | |
Close KafkaConsumers during tests
| -rw-r--r-- | test/test_consumer_group.py | 3 | ||||
| -rw-r--r-- | test/test_consumer_integration.py | 7 |
2 files changed, 10 insertions, 0 deletions
diff --git a/test/test_consumer_group.py b/test/test_consumer_group.py index b930748..f9a41a4 100644 --- a/test/test_consumer_group.py +++ b/test/test_consumer_group.py @@ -44,6 +44,7 @@ def test_consumer(kafka_broker, version): assert len(consumer._client._conns) > 0 node_id = list(consumer._client._conns.keys())[0] assert consumer._client._conns[node_id].state is ConnectionStates.CONNECTED + consumer.close() @pytest.mark.skipif(version() < (0, 9), reason='Unsupported Kafka Version') @@ -153,6 +154,7 @@ def test_paused(kafka_broker, topic): consumer.unsubscribe() assert set() == consumer.paused() + consumer.close() @pytest.mark.skipif(version() < (0, 9), reason='Unsupported Kafka Version') @@ -183,3 +185,4 @@ def test_heartbeat_thread(kafka_broker, topic): assert consumer._coordinator.heartbeat.last_poll == last_poll consumer.poll(timeout_ms=100) assert consumer._coordinator.heartbeat.last_poll > last_poll + consumer.close() diff --git a/test/test_consumer_integration.py b/test/test_consumer_integration.py index fe4e454..6429b77 100644 --- a/test/test_consumer_integration.py +++ b/test/test_consumer_integration.py @@ -44,6 +44,7 @@ def test_kafka_consumer(simple_client, topic, kafka_consumer_factory): assert len(messages[0]) == 100 assert len(messages[1]) == 100 + consumer.close() class TestConsumerIntegration(KafkaIntegrationTestCase): @@ -558,6 +559,7 @@ class TestConsumerIntegration(KafkaIntegrationTestCase): messages.add((msg.partition, msg.offset)) self.assertEqual(len(messages), 5) self.assertGreaterEqual(t.interval, TIMEOUT_MS / 1000.0 ) + consumer.close() @kafka_versions('>=0.8.1') def test_kafka_consumer__offset_commit_resume(self): @@ -597,6 +599,7 @@ class TestConsumerIntegration(KafkaIntegrationTestCase): output_msgs2.append(m) self.assert_message_count(output_msgs2, 20) self.assertEqual(len(set(output_msgs1) | set(output_msgs2)), 200) + consumer2.close() @kafka_versions('>=0.10.1') def test_kafka_consumer_max_bytes_simple(self): @@ -617,6 +620,7 @@ class TestConsumerIntegration(KafkaIntegrationTestCase): self.assertEqual( seen_partitions, set([ TopicPartition(self.topic, 0), TopicPartition(self.topic, 1)])) + consumer.close() @kafka_versions('>=0.10.1') def test_kafka_consumer_max_bytes_one_msg(self): @@ -642,6 +646,7 @@ class TestConsumerIntegration(KafkaIntegrationTestCase): fetched_msgs = [next(consumer) for i in range(10)] self.assertEqual(len(fetched_msgs), 10) + consumer.close() @kafka_versions('>=0.10.1') def test_kafka_consumer_offsets_for_time(self): @@ -695,6 +700,7 @@ class TestConsumerIntegration(KafkaIntegrationTestCase): self.assertEqual(offsets, { tp: late_msg.offset + 1 }) + consumer.close() @kafka_versions('>=0.10.1') def test_kafka_consumer_offsets_search_many_partitions(self): @@ -733,6 +739,7 @@ class TestConsumerIntegration(KafkaIntegrationTestCase): tp0: p0msg.offset + 1, tp1: p1msg.offset + 1 }) + consumer.close() @kafka_versions('<0.10.1') def test_kafka_consumer_offsets_for_time_old(self): |