From 41ca4d07a903990706ea2a5ccae6bcf21e32bdf7 Mon Sep 17 00:00:00 2001 From: tierno Date: Thu, 16 Jul 2020 11:22:12 +0000 Subject: [PATCH] fix 1131 Allow to get old kafka messages Change-Id: I49c43fd3b2049d32adc7be7a29be0d921624bc88 Signed-off-by: tierno --- osm_common/msgkafka.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/osm_common/msgkafka.py b/osm_common/msgkafka.py index 91775f2..4d02024 100644 --- a/osm_common/msgkafka.py +++ b/osm_common/msgkafka.py @@ -109,7 +109,8 @@ class MsgKafka(MsgBase): finally: await self.producer.stop() - async def aioread(self, topic, loop=None, callback=None, aiocallback=None, group_id=None, **kwargs): + async def aioread(self, topic, loop=None, callback=None, aiocallback=None, group_id=None, from_beginning=None, + **kwargs): """ Asyncio read from one or several topics. :param topic: can be str: single topic; or str list: several topics @@ -118,6 +119,9 @@ class MsgKafka(MsgBase): :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 from_beginning: if True, messages will be obtained from beginning instead of only new ones. + If group_id is supplied, only the not processed messages by other worker are obtained. + If group_id is None, all messages stored at kafka are obtained. :param kwargs: optional keyword arguments for callback function :return: If no callback defined, it returns (topic, key, message) """ @@ -133,8 +137,8 @@ class MsgKafka(MsgBase): topic_list = topic else: topic_list = (topic,) - - self.consumer = AIOKafkaConsumer(loop=loop, bootstrap_servers=self.broker, group_id=group_id) + self.consumer = AIOKafkaConsumer(loop=loop, bootstrap_servers=self.broker, group_id=group_id, + auto_offset_reset="earliest" if from_beginning else "latest") await self.consumer.start() self.consumer.subscribe(topic_list) -- 2.17.1