Feature: add Kafka transactions support

[ManiMozaffar]

Is your feature request related to a problem? Please describe.
In batch processing, a group of messages is processed together. Transactions ensure that this entire batch is treated as a single unit of work. Although library support adding it to the producer config, but it doesn't support actualy using it.

Describe the solution you'd like
Some document and reworks regarding low level access to producer and consumer would be really nice in general. I think going forward this would be needed.

[Lancetnik]

@ManiMozaffar which broker are you talking?

[ManiMozaffar]

Kafka

[ManiMozaffar]

@Lancetnik either all messages should be produced, or none.

[davorrunje]

@ManiMozaffar we'll release a new Kafka broker based on Confluent lib this week. This is a good time to let us know what features you would like us to support.

[ManiMozaffar]

@davorrunje I haven't had a chance yet to deep dive into code, but on documentation would be really nice to give some idea about low level access to broker and other stuff, I think that would be nessecary because of many things like instrumenting the broker, fixing weird bugs sometimes happening with brokers. E.x if broker dies and it was on transaction, it would go to a state where it can't produce a message transactionality so you'd have to restart the broker. (it loses the state of producer mapping to transactional id, or aka INVALID_PRODUCER_ID_MAPPING)

[RubenRibGarcia]

Hello once again! @Lancetnik I was reading a bit about the story for this feature ( specially a closed PR #1290 ) and I would like to talk a little bit about it's approach. That is, I know there is another feature on the Roadmap for RMQ to also support publish transaction and I think it would be best that both approaches followed a same pattern.

Focusing on Kafka... does it make sense that for KafkaBroker (both AioKafka and Confluent) have the publish decorator with a transaction argument ? The idea behind this is to abstract the way to handle transactions on both libs, but I'm also thinking that we should also provide some sort of annotation for developer to have more control over the transaction using explicit "begin", "commit", "rolback" methods.

[Lancetnik]

@RubenRibGarcia It is a much complex feature we image before and I am even not sure, that we can support it in FastStream API (in Kafka case at least)

The main idea now - is Depends-based transactional for RMQ (and Kafka probably)

@broker.subscriber(...)
async def handler(
    msg,
    transaction: Transaction  # will be commited after handler reseived
): ...

Anyway, we will back to this discussion little bit further