broker = KafkaBroker("localhost:9092")

Условие

broker = KafkaBroker("localhost:9092") app = FastStream(broker)

@broker.subscriber("raw_texts_topic", group_id="text_parsers") async def handle_text(text_data: dict): # Парсинг и сохранение в БД занимает ~10 секунд parse_text_and_save_to_postgres(text_data)

Ответ

Синхронный parse_text_and_save_to_postgres блокирует event loop внутри async-обработчика. Для блокирующего I/O вынесите вызов в ограниченный пул потоков; тяжёлый CPU-парсинг требует другого решения.

Минимальный вариант для преимущественно I/O-нагрузки:

import asyncio

@broker.subscriber("raw_texts_topic", group_id="text_parsers")
async def handle_text(text_data: dict):
    await asyncio.to_thread(parse_text_and_save_to_postgres, text_data)

Соединение или сессия БД должны создаваться и использоваться безопасно внутри рабочего потока. Для тяжёлого Python-парсинга рассмотрите процессы или отдельные workers. Ограничьте параллелизм и согласуйте подтверждение Kafka с успешным commit, предусмотрев идемпотентность. Одно увеличение числа обработчиков не устраняет блокирующий вызов.

Источник: faststream.ag2.ai.