How to ensure Kafka messages are flushed immediately

[ndbs-tsb]

I am making some experiments in switching to FastStream instead of using aiokafka directly. In my application it is important that messages are delivered as quickly as possible, so the old implementation calls producer.flush() after every producer.send(). Is there a way to accomplish this using FastStream?

I realise this may be more of a Kafka-specific question than a FastStream question, and perhaps this is best solved by some configuration of the producer...

[Lancetnik]

@ndbs-tsb sorry for answering so late... Indeed, we missed that feature and didn't add it to public API.
You can get access to raw aiokafka producer to make it manyally smth like that

from faststream.kafka import KafkaBroker

broker = KafkaBroker()
async def main():
    msg = await broker.publish(...)
    await broker._producer._producer.flush()

But, we should make it publish for sure. What do you think about broker.flush() method?

[ndbs-tsb]

@Lancetnik no worries.

I am not sure what the best approach to this is. Exposing a broker.flush() method seems sensible. However, I am currently using the decorators, which I quite like, e.g.

@broker.subscriber(some_topic)
@publisher
async def handle_msg(msg: SomeType) -> SomeOtherType:
  .
  .
  .
  return SomeOtherType(...)

I suppose it would be easy enough to create a new decorator @autoflush that could be added to those handlers where flushing is desired? Then the above example would become:

@broker.subscriber(some_topic)
@publisher
@autoflush
async def handle_msg(msg: SomeType) -> SomeOtherType:
  .
  .
  .
  return SomeOtherType(...)

[Lancetnik]

I don't like an another decorator

Probably, we can add an option like

@broker.publisher(..., autoflush=True)

Or even

KafkaBroker(autoflush=True)

[ndbs-tsb]

I don't like an another decorator

I can understand that.

@broker.publisher(..., autoflush=True)

Or even

KafkaBroker(autoflush=True)

Something along these lines makes sense to me.

[Lancetnik]

Well, I'll create an Issue to implement this feature. Thank you for the suggestion!

[ndbs-tsb]

Cool. Thanks for the good discussion.