diff options
| -rw-r--r-- | kafka/consumer/group.py | 1 | ||||
| -rw-r--r-- | test/test_client_async.py | 28 |
2 files changed, 27 insertions, 2 deletions
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py index 9ebf604..2562cfb 100644 --- a/kafka/consumer/group.py +++ b/kafka/consumer/group.py @@ -772,6 +772,7 @@ class KafkaConsumer(six.Iterator): if pattern is not None: self._client.cluster.need_all_topic_metadata = True self._client.set_topics([]) + self._client.cluster.request_update() log.debug("Subscribed to topic pattern: %s", pattern) else: self._client.cluster.need_all_topic_metadata = False diff --git a/test/test_client_async.py b/test/test_client_async.py index b165f93..8661e90 100644 --- a/test/test_client_async.py +++ b/test/test_client_async.py @@ -17,6 +17,8 @@ from kafka.future import Future from kafka.protocol.metadata import MetadataResponse, MetadataRequest from kafka.protocol.produce import ProduceRequest from kafka.structs import BrokerMetadata +from kafka.cluster import ClusterMetadata +from kafka.future import Future @pytest.fixture @@ -285,8 +287,30 @@ def test_least_loaded_node(): pass -def test_set_topics(): - pass +def test_set_topics(mocker): + request_update = mocker.patch.object(ClusterMetadata, 'request_update') + request_update.side_effect = lambda: Future() + cli = KafkaClient(api_version=(0, 10)) + + # replace 'empty' with 'non empty' + request_update.reset_mock() + fut = cli.set_topics(['t1', 't2']) + assert not fut.is_done + request_update.assert_called_with() + + # replace 'non empty' with 'same' + request_update.reset_mock() + fut = cli.set_topics(['t1', 't2']) + assert fut.is_done + assert fut.value == set(['t1', 't2']) + request_update.assert_not_called() + + # replace 'non empty' with 'empty' + request_update.reset_mock() + fut = cli.set_topics([]) + assert fut.is_done + assert fut.value == set() + request_update.assert_not_called() @pytest.fixture |