Transaction support

[jonas-grgt]

I've been looking into how spring-pulsar could support transactions.

The main hurdle to overcome will be sharing the Transaction details between the PulsarMessageListenerContainer and PulsarTemplate as both the pulsar Consumer (through acknowledgeAsync) and Producer (through newMessage(txn)) need access to the Transaction.

To give a more concrete example, imagine the consume, process, and produce messages in one atomic operation use-case.
In such a scenario there is no way to access the Transaction from within an applications listener method.
One way to solve this would be to pass in the Transaction as a parameter in the listener method together with a transaction parameter in the @PulsarListener annotation.

@PulsarListener(topic = A_TOPIC, transactional = true)
public void consumer(SampleMessage message, Transaction transaction) {
    pulsarTemplate.send(transform(message), transaction);
}

This way the PulsarMessageListenerContainer would know when to initiate a transaction, get the transaction from the PulsarClient. Pass the Transaction in to the listener method. And at the end commit the transaction and acknowledge the consumption or rollback (through Consumer.negativeAcknowledge() and Transaction.abort())

The same would go for the @ReactivePulsarListener

What do you think?

[onobc]

Thanks for starting the discussion @jonasgeiregat .

We have done some preliminary design which IIRC lines up w/ your suggestion. I was planning on continuing w/ this in the next 1-2 weeks. Let me get back into the design and then we can sync up.

[onobc]

Doing a bit of GH discussions housekeeping...

Since this has been implemented starting in 1.1.0 - going to close the discussion as resolved.

Thanks @jonas-grgt for the discussion.