python fastapi, aiokafka starvation problem

Question

Problem : aiokafka consumer starving the fastapi endpoint, due to which our kubernetes liveness probes are failing and any other service calling the exposed endpoints are getting timed out.

Details :

There is a kafka consumer, which starts during fastapi startup event, and keep on listening to the particular topic.

And then there is fastapi endpoint which serves the request.

When there are lot messages in kafka topic partion, kafka consumer starving the eventloop and the requests served by fastapi endpoints are timing out.

How can we solve this problem?

#all the imports



consumer = None
consumer_task = None

log = None

def get_application():
    #initialize fastapi app and with different routes and do some stuff
    return app

app = get_application()

@app.on_event("startup")
async def startup_event():
    #initialize consumer
    await initialize()
    # start consuming
    await consume()

@app.on_event("shutdown")
async def shutdown_event():
    #close consumer

async def initialize():
    #initilize
    # get cluster layout and join group
    await consumer.start()
    await consumer.seek_to_committed()

async def consume():
    global consumer_task
    loop = asyncio.get_event_loop()
    consumer_task = loop.create_task(send_consumer_message(consumer))


async def send_consumer_message(consumer):
    try:
        # consume messages
        async for msg in consumer:
            #do message processing
    except Exception as e:
        log.info(f"message consuming failed withe error: {repr(e)}")
    finally:
        log.warning("stopping consumer")
        await consumer.stop()