GH-2001: Propagate Exception from TX Sync Commit

Resolves https://github.com/spring-projects/spring-kafka/issues/2001

Previously, synchronized transaction commits (for producer initiated
transactions) were not propagated to the caller.

**cherry-pick to all supported branches**

* Fix typo in doc
# Conflicts:
#	src/reference/asciidoc/kafka.adoc
This commit is contained in:
Gary Russell
2021-11-09 15:37:57 -05:00
committed by Artem Bilan
parent 38390e893d
commit 38ed17de81
3 changed files with 69 additions and 12 deletions

View File

@@ -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<K, V> 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();
}
}

View File

@@ -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<String, String> producer =
new MockProducer<>(true, new StringSerializer(), new StringSerializer());
producer.initTransactions();
producer.commitTransactionException = new IllegalStateException();
@SuppressWarnings("unchecked")
ProducerFactory<String, String> pf = mock(ProducerFactory.class);
given(pf.transactionCapable()).willReturn(true);
given(pf.createProducer(isNull())).willReturn(producer);
KafkaTemplate<String, String> 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 {
}
}
}

View File

@@ -3156,6 +3156,10 @@ When a listener container is configured to use a `KafkaTransactionManager` or `C
See <<ex-jdbc-sync>> 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`