Kafka consumer records - processing

Question

For kafka I'm using getmany to read the consumer messages. In the total of 650 messages(which will take around 3days to process), processing happens for around 100-150 records(sometimes 12hrs or sometimes 24hrs) and then there is no further processing happening. But the consumer stream is not closed, when I put a new message it is processing, but couldn't figure out why it happens.

I thought that may be some messages are getting skipped so to check that I tried to produce only 25 messages - all processed(3hrs) then produced 100 messages - all processed(12hrs), in these situations messages are not getting skipped. Only when many no of messages(650) are put this issue happens.

Code snippet :

async def consume(loop,lock):
    logger.info('Inside Consume')

    consumer = AIOKafkaConsumer(KAFKA_TOPIC,
                        loop=loop,
                        bootstrap_servers=bootstrap_servers,
                        group_id=group_id,           
                        enable_auto_commit=enable_auto_commit,       
                        auto_commit_interval_ms=auto_commit_interval_ms,  
                        auto_offset_reset=auto_offset_reset,  
                        max_poll_records= 1,
                        max_poll_interval_ms=1500000,
                        rebalance_timeout_ms=1500000)
              
    await consumer.start()
    while True:
        result = await consumer.getmany(timeout_ms=1500000, max_records=1)
        for tp, msg in result.items(): 
            if msg:
                message = msg[0].value.decode()
                message_json = json.loads(message)
                data_encoded = message_json['payload']

                if data_encoded.get('content') is not None:
                    msg = data_encoded['content']
                    message_dict_dum = base64.b64decode(msg)
                    message_dict = json.loads(message_dict_dum)
                    
                    if message_dict.get("key1") is not None:
                        if message_dict["key1"] == "something":
                            logger.info('Hurray new message to process')
                            logger.info(message_dict["id"])
                            await process(data_encoded['content'],lock)
                            

async def process(msg, lock):
    async with lock:
        logger.info('processing... ->{}'.format(msg))
        await processmessage(msg)
        logger.info('done processing')        


if __name__ == "__main__":
    logger.info('Inside main')

    loop = asyncio.new_event_loop()
    lock = asyncio.Lock(loop=loop)

    loop.create_task(consume(loop, lock))
    loop.run_forever()