Kafka transaction not being commited when initiated in afterCommit of other transaction

[dmkaw]

Affects Version(s):

2.9.0 (an earlier - starting 2.5.17, 2.6.12, 2.7.9 and 2.8.0)

We are using transaction synchronization mechanism for scheduling some business logic beforeCommit, afterCommit etc. (hibernate transaction), in this logic we sometimes are sending transactional kafka messages
Everything was working fine with kafka 2.7.1 + spring-kafka 2.7.6 (Spring boot 2.5.4)

After upgrading to kafka 3.2.0 + spring-kafka 2.9.0 we started to receive errors:

beginTransaction failed: CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@238e30ae]

java.lang.IllegalStateException: TransactionalId tx-fsm-core-jIAGrMnP58-0: Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION

I believe, problem was introduced in #2005 - in ProducerFactoryUtils commit on resourceHolder was moved from afterCompletion method to processResourceAfterCommit method.
Problem now is, that in TransactionSynchronization.afterCommit() kafka transaction is being attached to main transaction (hibernate) which is still active, but already commited, and because of that it is not being commited afterwards (main transaction afterCommit method was already fired, and resourceHolder commit method will not be executed)

Problem exist only when sending kafka messages in afterCommit.
beforeCommit, and synchronous processing (without transaction synchronization) is working fine.

To Reproduce

Explained above

Expected behavior

Kafka transaction properly commited when invoked in other transaction synchronization afterCommit method.

Sample

https://github.com/dmkaw/kafka-transaction-issue

[garyrussell]

Can you provide a concrete example MCRE - this sounds like an abuse of transaction synchronization, but I need to see the code.

[dmkaw]

I have provided sample in main post.

[garyrussell]

Your KafkaSender is using the hibernate TM, not the KafkaTransactionManager.

This works fine.

@Component
@Transactional(transactionManager = "kafkaTransactionManager")
public class KafkaSender {

    @Autowired
    private KafkaTemplate<Object, Object> template;

    public void execute() {
        template.send("test", "test_msg");
        template.send("test", "test_msg1");
        template.send("test", "test_msg2");
        template.send("test", "test_msg3");
    }

}

Or you can remove the annotation and use template.executeInTransaction().

[garyrussell]

You are adding a new transaction status after triggerAfterCommit has been called; hence it won't work. With the above change, we get a new kafka transaction which will commit on method exit; providing similar functionality to transaction synchronization.

The previous logic of doing the commit in afterCompletion was clearly wrong, because exceptions were being swallowed.

[dmkaw]

Thank you for comments.
I will use slightly different approach as our implementation is more generic, and i cant predict if only thing that will be done in after commit will be kafka related stuff.
Instead of creating new kafka transaction i will create new hibernate transaction, this way it is working properly, and in case of rollback both kafka and jdbc should be rollbacked correctly.