From 10602af49b01f0370761e913eacf485e77e4fde0 Mon Sep 17 00:00:00 2001 From: tierno Date: Mon, 18 Feb 2019 14:53:54 +0000 Subject: [PATCH] allow set/unset group_id at kafka message consumer Change-Id: I138cb29b012536473f6aea21975b06013e84e043 Signed-off-by: tierno --- osm_common/__init__.py | 4 ++-- osm_common/msgbase.py | 2 +- osm_common/msgkafka.py | 10 ++++++++-- osm_common/msglocal.py | 11 ++++++++--- 4 files changed, 19 insertions(+), 8 deletions(-) diff --git a/osm_common/__init__.py b/osm_common/__init__.py index edf5308..3073b42 100644 --- a/osm_common/__init__.py +++ b/osm_common/__init__.py @@ -15,6 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -version = '0.1.18' +version = '0.1.19' # TODO add package version filling commit id with 0's; e.g.: '5.0.0.post11+00000000.dirty-1' -date_version = '2019-01-28' +date_version = '2019-02-18' diff --git a/osm_common/msgbase.py b/osm_common/msgbase.py index 92b24e0..7c27f64 100644 --- a/osm_common/msgbase.py +++ b/osm_common/msgbase.py @@ -77,5 +77,5 @@ class MsgBase(object): async def aiowrite(self, topic, key, msg, loop=None): raise MsgException("Method 'aiowrite' not implemented", http_code=HTTPStatus.INTERNAL_SERVER_ERROR) - async def aioread(self, topic, loop=None, callback=None, aiocallback=None, **kwargs): + async def aioread(self, topic, loop=None, callback=None, aiocallback=None, group_id=None, **kwargs): raise MsgException("Method 'aioread' not implemented", http_code=HTTPStatus.INTERNAL_SERVER_ERROR) diff --git a/osm_common/msgkafka.py b/osm_common/msgkafka.py index 63ee69c..bc9147d 100644 --- a/osm_common/msgkafka.py +++ b/osm_common/msgkafka.py @@ -105,26 +105,32 @@ class MsgKafka(MsgBase): finally: await self.producer.stop() - async def aioread(self, topic, loop=None, callback=None, aiocallback=None, **kwargs): + async def aioread(self, topic, loop=None, callback=None, aiocallback=None, group_id=None, **kwargs): """ Asyncio read from one or several topics. :param topic: can be str: single topic; or str list: several topics :param loop: asyncio loop. To be DEPRECATED! in near future!!! loop must be provided inside config at connect :param callback: synchronous callback function that will handle the message in kafka bus :param aiocallback: async callback function that will handle the message in kafka bus + :param group_id: kafka group_id to use. Can be False (set group_id to None), None (use general group_id provided + at connect inside config), or a group_id string :param kwargs: optional keyword arguments for callback function :return: If no callback defined, it returns (topic, key, message) """ if not loop: loop = self.loop + if group_id is False: + group_id = None + elif group_id is None: + group_id = self.group_id try: if isinstance(topic, (list, tuple)): topic_list = topic else: topic_list = (topic,) - self.consumer = AIOKafkaConsumer(loop=loop, bootstrap_servers=self.broker, group_id=self.group_id) + self.consumer = AIOKafkaConsumer(loop=loop, bootstrap_servers=self.broker, group_id=group_id) await self.consumer.start() self.consumer.subscribe(topic_list) diff --git a/osm_common/msglocal.py b/osm_common/msglocal.py index eba4b97..965cb26 100644 --- a/osm_common/msglocal.py +++ b/osm_common/msglocal.py @@ -123,12 +123,17 @@ class MsgLocal(MsgBase): except Exception as e: # TODO refine raise MsgException(str(e), HTTPStatus.INTERNAL_SERVER_ERROR) - async def aioread(self, topic, loop=None, callback=None, aiocallback=None, **kwargs): + async def aioread(self, topic, loop=None, callback=None, aiocallback=None, group_id=None, **kwargs): """ Asyncio read from one or several topics. It blocks :param topic: can be str: single topic; or str list: several topics - :param loop: asyncio loop - :return: topic, key, message + :param loop: asyncio loop. To be DEPRECATED! in near future!!! loop must be provided inside config at connect + :param callback: synchronous callback function that will handle the message + :param aiocallback: async callback function that will handle the message + :param group_id: group_id to use for load balancing. Can be False (set group_id to None), None (use general + group_id provided at connect inside config), or a group_id string + :param kwargs: optional keyword arguments for callback function + :return: If no callback defined, it returns (topic, key, message) """ _loop = loop or self.loop try: -- 2.17.1