Bug: await broker.publish do not waits actual message sends on Kafka(aiokafka)

[zpl]

Describe the bug

await broker.publish(....) do not wait for actual message send, even with enable_idempotence=True / acks='all'

I have fix for it.

so, this happend on Aiokafka, with remote servers. the reason is in faststream/kafka/publisher/producer.py

in

    @override
    async def publish(  # type: ignore[override]
        self,
        message: "SendableMessage",
        topic: str,
        *,
        correlation_id: str,
        key: Union[bytes, Any, None] = None,
        partition: Optional[int] = None,
        timestamp_ms: Optional[int] = None,
        headers: Optional[Dict[str, str]] = None,
        reply_to: str = "",
    ) -> None:
        """Publish a message to a topic."""
        message, content_type = encode_message(message)

        headers_to_send = {
            "content-type": content_type or "",
            "correlation_id": correlation_id,
            **(headers or {}),
        }

        if reply_to:
            headers_to_send["reply_to"] = headers_to_send.get(
                "reply_to",
                reply_to,
            )

        # should be send_and_wait here
        await self._producer.send( 
            topic=topic,
            value=message,
            key=key,
            partition=partition,
            timestamp_ms=timestamp_ms,
            headers=[(i, (j or "").encode()) for i, j in headers_to_send.items()],
        ) 

there is a call to await self._producer.send and send is not actually waits for message deliver - it returns a furute (https://aiokafka.readthedocs.io/en/stable/_modules/aiokafka/producer/producer.html#AIOKafkaProducer.send)

here we should use send_and_wait instead (https://aiokafka.readthedocs.io/en/stable/_modules/aiokafka/producer/producer.html#AIOKafkaProducer.send_and_wait)

Could you please help me with pull request? thanks.

How to reproduce
Include source code:

import asyncio
from faststream import FastStream

# KAFKA_BOOTSTRAP_SERVERS - some remote cluster (not localhost)
broker = KafkaBroker( config('KAFKA_BOOTSTRAP_SERVERS'), security=security, enable_idempotence=True)

async def send_message():
    await broker.connect()
    await broker.publish([1,2,3], "msg_answer1")

    # it will work, if I add 
    # await asyncio.sleep(5) 




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

Expected behavior
Expected: Message sent and then app closes

Observed behavior
App closes before actual message sent

Additional context
Provide any other relevant context or information about the problem here.

[Lancetnik]

Thank you for the report! I found the same behavior with Confluent, but was not sure, how it works
Now I see the problem and try to fix it soon

[Lancetnik]

@zpl I created a draft PR alreat, but still not sure, which method should be used as a default one: should we wait for confirmation everytime or the current behavior should be the default one?

[zpl]

@Lancetnik , I think it should be send_and_wait for default. probably in some rare cases we can skip it for performance?

well, I think even with performance I prefer to use create_task instead of await. but I expect that send will actually send's a message.