Synchronized kafka transaction is not commited nor rolled back when code from custom "afterCommit" throws exception

[ltorzynski]

Affects Version(s):

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

Problem description

We are using transaction synchronization mechanism for some business logic in afterCommit method (hibernate transactions).

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

java.lang.IllegalStateException: TransactionalId tx-tenants-RgIAuTvu9J-1: Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION
	at org.apache.kafka.clients.producer.internals.TransactionManager.transitionTo(TransactionManager.java:1089)
	at org.apache.kafka.clients.producer.internals.TransactionManager.transitionTo(TransactionManager.java:1082)
	at org.apache.kafka.clients.producer.internals.TransactionManager.beginTransaction(TransactionManager.java:368)
	at org.apache.kafka.clients.producer.KafkaProducer.beginTransaction(KafkaProducer.java:628)
	at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.beginTransaction(DefaultKafkaProducerFactory.java:1126)
	at org.springframework.kafka.core.ProducerFactoryUtils.getTransactionalResourceHolder(ProducerFactoryUtils.java:99)
	at org.springframework.kafka.core.KafkaTemplate.getTheProducer(KafkaTemplate.java:751)
	at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:645)
	at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:430)

The problem is that when our "afterCommit" method throws exception - "Kafka synchronized" producer initiated transaction is never commited ( ProducerFactoryUtils.processResourceAfterCommit is never invoked). ProducerFactoryUtils.releaseResource method is then called in the end of the main transaction and this results in returning invalid KafkaProducer (transactionState = IN_TRANSACTION) to the DefaultKafkaProducerFactory cache.

Subsequent calls end with above error because KafkaProducer returned from cache has invalid transaction state ((transactionState = IN_TRANSACTION)

Sample method:

@Transactional
public void sampleTransactionalMethod() {
  ...
   kafkaTemplate.send(..)
  TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                         if(true){
                            throw new RuntimeException("Any exception");
                        }
                }
                @Override
                public int getOrder() {
                    return Ordered.HIGHEST_PRECEDENCE;
                }

   });

}

The problem was probably introduced in #2005 - in ProducerFactoryUtils commit on resourceHolder was moved from afterCompletion method to processResourceAfterCommit method.

[garyrussell]

Please provide much more context - transaction manager and synchronizations hierarchy.

If possible, provide an MCRE so we can reproduce locally.

[garyrussell]

I just looked at the framework code and it looks like it stops calling other synchronizations when any of them throw an exception; that is probably the cause (if the Kafka synchronization is after the one you are throwing the exception from).

	public static void invokeAfterCommit(@Nullable List<TransactionSynchronization> synchronizations) {
		if (synchronizations != null) {
			for (TransactionSynchronization synchronization : synchronizations) {
				synchronization.afterCommit();
			}
		}
	}

Can you reorder the synchronizations so that Kafka is called first?

[ltorzynski]

This is spring boot application with almost default kafka config. Main transactionManager is HibernateTransactionManager (but it does not really matter). It is "producer initiated" transaction - no listener container.

Yes, KafkaResourceSynchronization has LOWEST_PRECEDENCE (the default one) and it cannot be changed so it is always called as the last one.

This problem is quite tricky because it results in some inconsistent state in DefaultKafkaProducerFactory producers cache (as I mentioned - KafkaProducers with IN_TRANSACTION state)

I will try to prepare MCRE - but the problem is general and easilly reproducible.

[garyrussell]

I think we can add some logic in KafkaResourceHolder to track whether commit or rollback was called before close() and do something there.

It's not clear, though, whether it should commit or rollback. As I understand it, the primary transaction has already committed so maybe commit() is the right thing to do.

[ltorzynski]

Yes, but I am not sure why previous "KafkaResourceSynchronization" solution (fixed in: #2005) was wrong - it commited of rolled back kafka transaction based on main transaction status.
We could also send some additional kafka messages in our "aferCommit" method without creating separate transactions.
It was also much safer (commit or rollback was always invoked)

[ltorzynski]

We can of course change our code and not use "afterCommit" mechanism at all - but I believe current spring-kafka behaviour should be somehow fixed

[garyrussell]

The problem fixed by #2005 was that a failure to commit (in afterCompletion) was eaten by the framework code and not propagated to the caller, so the application had no idea that there was a problem with the commit.

		if (synchronizations != null) {
			for (TransactionSynchronization synchronization : synchronizations) {
				try {
					synchronization.afterCompletion(completionStatus);
				}
				catch (Throwable ex) {
					logger.debug("TransactionSynchronization.afterCompletion threw exception", ex);
				}
			}
		}

[garyrussell]

Another option is we could restore the logic in. afterCompletion() and ignore it in the resource holder if the transaction was already committed.

[ltorzynski]

But the main transaction is already commited (in afterCompletion()) so the caller does not really need to catch it. The exception is also always logged so it is not eaten (hidden)

[garyrussell]

Right, but throwing the exception to the caller allows him to create a compensating transaction (to back out the primary commit). Without the exception propagation, the application has no way to do that.

My last suggestion should satisfy both needs.