Add metadata update to pattern subscribing (#915)

Author: laz2

kafka/consumer/group.py

@@ -772,6 +772,7 @@ def subscribe(self, topics=(), pattern=None, listener=None):
         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

test/test_client_async.py

@@ -17,6 +17,8 @@
 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