diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactoryUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactoryUtils.java index da02f657..7a5b2c19 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactoryUtils.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactoryUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -145,13 +145,15 @@ public final class ProducerFactoryUtils { return false; } + @Override + protected void processResourceAfterCommit(KafkaResourceHolder resourceHolder) { + resourceHolder.commit(); + } + @Override public void afterCompletion(int status) { try { - if (status == TransactionSynchronization.STATUS_COMMITTED) { - this.resourceHolder.commit(); - } - else { + if (status != TransactionSynchronization.STATUS_COMMITTED) { this.resourceHolder.rollback(); } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java index 6116521e..49b56a6e 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java @@ -72,10 +72,13 @@ import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.kafka.transaction.KafkaTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.TransactionException; import org.springframework.transaction.annotation.EnableTransactionManagement; import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.AbstractPlatformTransactionManager; +import org.springframework.transaction.support.DefaultTransactionStatus; import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.concurrent.SettableListenableFuture; @@ -296,14 +299,15 @@ public class KafkaTemplateTransactionTests { ResourcelessTransactionManager tm = new ResourcelessTransactionManager(); - new TransactionTemplate(tm) - .execute(s -> { - template.sendDefault("foo", "bar"); + assertThatExceptionOfType(ProducerFencedException.class).isThrownBy(() -> + new TransactionTemplate(tm) + .execute(s -> { + template.sendDefault("foo", "bar"); - // Mark the mock producer as fenced so it throws when committing the transaction - producer.fenceProducer(); - return null; - }); + // Mark the mock producer as fenced so it throws when committing the transaction + producer.fenceProducer(); + return null; + })); assertThat(producer.transactionCommitted()).isFalse(); assertThat(producer.closed()).isTrue(); @@ -573,6 +577,28 @@ public class KafkaTemplateTransactionTests { pf.destroy(); } + @Test + void syncCommitFails() { + DummyTM tm = new DummyTM(); + MockProducer producer = + new MockProducer<>(true, new StringSerializer(), new StringSerializer()); + producer.initTransactions(); + producer.commitTransactionException = new IllegalStateException(); + + @SuppressWarnings("unchecked") + ProducerFactory pf = mock(ProducerFactory.class); + given(pf.transactionCapable()).willReturn(true); + given(pf.createProducer(isNull())).willReturn(producer); + + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(STRING_KEY_TOPIC); + + assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> + new TransactionTemplate(tm).execute(status -> template.sendDefault("foo"))); + + assertThat(tm.committed).isTrue(); + } + @Configuration @EnableTransactionManagement public static class DeclarativeConfig { @@ -680,4 +706,29 @@ public class KafkaTemplateTransactionTests { } + @SuppressWarnings("serial") + private static final class DummyTM extends AbstractPlatformTransactionManager { + + boolean committed; + + @Override + protected Object doGetTransaction() throws TransactionException { + return new Object(); + } + + @Override + protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException { + } + + @Override + protected void doCommit(DefaultTransactionStatus status) throws TransactionException { + this.committed = true; + } + + @Override + protected void doRollback(DefaultTransactionStatus status) throws TransactionException { + } + + } + } diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 75482a4e..4c873af5 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -3156,6 +3156,10 @@ When a listener container is configured to use a `KafkaTransactionManager` or `C See <> for an example application that synchronizes JDBC and Kafka transactions. +NOTE: Starting with versions 2.5.17, 2.6.12, 2.7.9 and 2.8.0, if the commit fails on the synchronized transaction (after the primary transaction has committed), the exception will be thrown to the caller. +Previously, this was silently ignored (logged at debug). +Applications should take remedial action, if necessary, to compensate for the committed primary transaction. + [[chained-transaction-manager]] ===== Using `ChainedKafkaTransactionManager`