kafka client update cluster

tests/test_unikafka.py

@@ -62,6 +62,9 @@ def __init__(self, *args, **kw):
     def topics(self):
         return self._topic_dict
 
+    def update_cluster(self):
+        pass
+
 
 @pytest.fixture(scope="function")
 def topic_consumer(request, monkeypatch):

xpaw/fetcher.py

@@ -313,6 +313,7 @@ def push_request(self, topic, req):
         with self._lock:
             if topic not in self._set:
                 self._set.add(topic)
+                self._kafka_client.update_cluster()
                 self._producers[topic] = self._kafka_client.topics[topic.encode("utf-8")].get_producer()
             self._producers[topic].produce(r)
 

xpaw/unikafka.py

@@ -137,7 +137,6 @@ async def _poll_forever(self):
                         except ConsumerStoppedException:
                             log.warn("Consumer of topic '{0}' stopped".format(t))
                             self._remove_consumer(t)
-                            self._kafka_client.update_cluster()
                             self._create_consumer(t)
                             log.info("Reset consumer of topic '{0}'".format(t))
                             msg = self._consumers[t].consume(block=True)
@@ -157,6 +156,7 @@ async def _poll_forever(self):
     def _create_consumer(self, topic):
         q = deque(maxlen=self._queue_size)
         self._mq[topic] = q
+        self._kafka_client.update_cluster()
         self._consumers[topic] = self._kafka_client.topics[topic.encode("utf-8")].get_balanced_consumer(
             consumer_group=self._group.encode("utf-8"),
             auto_commit_enable=True,
@@ -168,6 +168,7 @@ def _remove_consumer(self, topic):
         if len(q) > 0:
             producer = None
             try:
+                self._kafka_client.update_cluster()
                 producer = self._kafka_client.topics[topic.encode("utf-8")].get_producer(linger_ms=100)
                 while len(q) > 0:
                     b = q.popleft()

xpaw/version.py

@@ -1,3 +1,3 @@
 # coding=utf-8
 
-__version__ = "0.5.5a7"
+__version__ = "0.5.5a8"