Python Faust await agent.ask () не возвращает ответ и зависает вызывающая его функция

Question

Я новичок в Python, играю с вещами, пытаюсь общаться с Python через Kafka с помощью Faust.Итак, у меня есть небольшой проект PoC.Определение приложения Faust:

# app.py

import faust as f
from models import ReadRequest, ReadResponse

app = f.App("faust-app", broker="kafka://localhost:9092", store="rocksdb://")

topics = {"read-request": app.topic("read-request", value_type=ReadRequest)}


def get_app() -> f.types.AppT:
    return app


def get_topic(name: str) -> f.types.TopicT:
    return topics[name]

Агент чтения моей БД:

# reader.py

import pandas as pd
from pymongo import MongoClient
from app import get_app, get_topic

client = MongoClient()

app = get_app()
req_topic = get_topic("read-request")


@app.agent(req_topic)
async def read_request(requests):
    async for request in requests:
        db = client.test
        coll = db[request["collection"]]
        result = coll.find(request["query"])
        df = pd.DataFrame(result)
        response = {
            "id": request["id"],
            "data": list(df.loc[:, df.columns != "_id"].to_dict(orient="records")),
        }
        print(response) # debug <1>
        yield response

Определения модели:

# models.py

import faust as f


class ReadRequest(f.Record):
    id: int
    collection: str
    query: dict

Тест agent.ask()

# test.py

import asyncio
from reader import read_request
from models import ReadRequest


async def run():
    result = await read_request.ask(ReadRequest(id=1, collection="test", query={}))
    print(result) # debug <2>


if __name__ == "__main__":
    asyncio.run(run())

Итак, у меня работает zookeeper, kafka server, mongodb и faust reader.Все использует готовые конфигурации.

Когда я запускаю python3 test.py, я вижу debug <1> вывод на печать, как и ожидалось, но debug <2> никогда не выключается и выполнение зависает там.

FaustДокументы говорят

Так что я предполагаю, что я все делаю правильно.У кого-нибудь есть подсказки?