Skip to content

Commit ac34b28

Browse files
bgKgaryrussell
bgK
authored and
garyrussell
committed
GH-753: Close transactional producer on error
Fixes #753 Improve exception handling for producer transaction commit / rollback * Close the producer if an exception is thrown while committing / rollbacking a transaction when synchronizing the Kafka transaction with another TransactionManager. * Don't reuse transactional producers if an exception is thrown when committing / rollbacking a transaction. Some of the exceptions are fatal and mean the producer cannot be reused. * Close the transactional producer if an exception occurs when calling beginTransaction when not using DefaultKafkaProducerFactory.
1 parent 63472ce commit ac34b28

File tree

5 files changed

+212
-22
lines changed

5 files changed

+212
-22
lines changed

Diff for: ‎spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

+26-14
Original file line numberDiff line numberDiff line change
@@ -324,14 +324,6 @@ public void beginTransaction() throws ProducerFencedException {
324324
}
325325
catch (RuntimeException e) {
326326
this.txFailed = true;
327-
logger.error("Illegal transaction state; producer removed from cache; possible cause: "
328-
+ "broker restarted during transaction", e);
329-
try {
330-
this.delegate.close();
331-
}
332-
catch (Exception ee) {
333-
// empty
334-
}
335327
throw e;
336328
}
337329
}
@@ -344,20 +336,40 @@ public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offs
344336

345337
@Override
346338
public void commitTransaction() throws ProducerFencedException {
347-
this.delegate.commitTransaction();
339+
try {
340+
this.delegate.commitTransaction();
341+
}
342+
catch (RuntimeException e) {
343+
this.txFailed = true;
344+
throw e;
345+
}
348346
}
349347

350348
@Override
351349
public void abortTransaction() throws ProducerFencedException {
352-
this.delegate.abortTransaction();
350+
try {
351+
this.delegate.abortTransaction();
352+
}
353+
catch (RuntimeException e) {
354+
this.txFailed = true;
355+
throw e;
356+
}
353357
}
354358

355359
@Override
356360
public void close() {
357-
if (this.cache != null && !this.txFailed) {
358-
synchronized (this) {
359-
if (!this.cache.contains(this)) {
360-
this.cache.offer(this);
361+
if (this.cache != null) {
362+
if (this.txFailed) {
363+
logger.warn("Error during transactional operation; producer removed from cache; possible cause: "
364+
+ "broker restarted during transaction");
365+
366+
this.delegate.close();
367+
}
368+
else {
369+
synchronized (this) {
370+
if (!this.cache.contains(this)) {
371+
this.cache.offer(this);
372+
}
361373
}
362374
}
363375
}

Diff for: ‎spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

+9-1
Original file line numberDiff line numberDiff line change
@@ -259,7 +259,15 @@ public <T> T executeInTransaction(OperationsCallback<K, V, T> callback) {
259259
Producer<K, V> producer = this.producers.get();
260260
Assert.state(producer == null, "Nested calls to 'executeInTransaction' are not allowed");
261261
producer = this.producerFactory.createProducer();
262-
producer.beginTransaction();
262+
263+
try {
264+
producer.beginTransaction();
265+
}
266+
catch (Exception e) {
267+
closeProducer(producer, false);
268+
throw e;
269+
}
270+
263271
this.producers.set(producer);
264272
T result = null;
265273
try {

Diff for: ‎spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactoryUtils.java

+18-7
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,15 @@ public static <K, V> KafkaResourceHolder<K, V> getTransactionalResourceHolder(
5757
.getResource(producerFactory);
5858
if (resourceHolder == null) {
5959
Producer<K, V> producer = producerFactory.createProducer();
60-
producer.beginTransaction();
60+
61+
try {
62+
producer.beginTransaction();
63+
}
64+
catch (RuntimeException e) {
65+
producer.close();
66+
throw e;
67+
}
68+
6169
resourceHolder = new KafkaResourceHolder<K, V>(producer);
6270
bindResourceToTransaction(resourceHolder, producerFactory);
6371
}
@@ -128,14 +136,17 @@ protected boolean shouldReleaseBeforeCompletion() {
128136

129137
@Override
130138
public void afterCompletion(int status) {
131-
if (status == TransactionSynchronization.STATUS_COMMITTED) {
132-
this.resourceHolder.commit();
139+
try {
140+
if (status == TransactionSynchronization.STATUS_COMMITTED) {
141+
this.resourceHolder.commit();
142+
}
143+
else {
144+
this.resourceHolder.rollback();
145+
}
133146
}
134-
else {
135-
this.resourceHolder.rollback();
147+
finally {
148+
super.afterCompletion(status);
136149
}
137-
138-
super.afterCompletion(status);
139150
}
140151

141152
@Override

Diff for: ‎spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java

+54
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import org.apache.kafka.clients.consumer.ConsumerRecord;
3838
import org.apache.kafka.clients.consumer.ConsumerRecords;
3939
import org.apache.kafka.clients.producer.Callback;
40+
import org.apache.kafka.clients.producer.MockProducer;
4041
import org.apache.kafka.clients.producer.Producer;
4142
import org.apache.kafka.clients.producer.ProducerConfig;
4243
import org.apache.kafka.clients.producer.ProducerRecord;
@@ -50,6 +51,7 @@
5051
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
5152
import org.springframework.context.annotation.Bean;
5253
import org.springframework.context.annotation.Configuration;
54+
import org.springframework.kafka.support.transaction.ResourcelessTransactionManager;
5355
import org.springframework.kafka.test.EmbeddedKafkaBroker;
5456
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
5557
import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -212,6 +214,58 @@ public void testNoTx() {
212214
.hasMessageContaining("No transaction is in process;");
213215
}
214216

217+
@Test
218+
public void testTransactionSynchronization() {
219+
MockProducer<String, String> producer = new MockProducer<>();
220+
producer.initTransactions();
221+
222+
@SuppressWarnings("unchecked")
223+
ProducerFactory<String, String> pf = mock(ProducerFactory.class);
224+
given(pf.transactionCapable()).willReturn(true);
225+
given(pf.createProducer()).willReturn(producer);
226+
227+
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);
228+
template.setDefaultTopic(STRING_KEY_TOPIC);
229+
230+
ResourcelessTransactionManager tm = new ResourcelessTransactionManager();
231+
232+
new TransactionTemplate(tm).execute(s -> {
233+
template.sendDefault("foo", "bar");
234+
return null;
235+
});
236+
237+
assertThat(producer.history()).containsExactly(new ProducerRecord<>(STRING_KEY_TOPIC, "foo", "bar"));
238+
assertThat(producer.transactionCommitted()).isTrue();
239+
assertThat(producer.closed()).isTrue();
240+
}
241+
242+
@Test
243+
public void testTransactionSynchronizationExceptionOnCommit() {
244+
MockProducer<String, String> producer = new MockProducer<>();
245+
producer.initTransactions();
246+
247+
@SuppressWarnings("unchecked")
248+
ProducerFactory<String, String> pf = mock(ProducerFactory.class);
249+
given(pf.transactionCapable()).willReturn(true);
250+
given(pf.createProducer()).willReturn(producer);
251+
252+
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);
253+
template.setDefaultTopic(STRING_KEY_TOPIC);
254+
255+
ResourcelessTransactionManager tm = new ResourcelessTransactionManager();
256+
257+
new TransactionTemplate(tm).execute(s -> {
258+
template.sendDefault("foo", "bar");
259+
260+
// Mark the mock producer as fenced so it throws when committing the transaction
261+
producer.fenceProducer();
262+
return null;
263+
});
264+
265+
assertThat(producer.transactionCommitted()).isFalse();
266+
assertThat(producer.closed()).isTrue();
267+
}
268+
215269
@Configuration
216270
@EnableTransactionManagement
217271
public static class DeclarativeConfig {

Diff for: ‎spring-kafka/src/test/java/org/springframework/kafka/support/transaction/ResourcelessTransactionManager.java

+105
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
/*
2+
* Copyright 2017-2018 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.kafka.support.transaction;
18+
19+
import java.util.ArrayList;
20+
import java.util.List;
21+
22+
import org.springframework.transaction.TransactionDefinition;
23+
import org.springframework.transaction.TransactionException;
24+
import org.springframework.transaction.support.AbstractPlatformTransactionManager;
25+
import org.springframework.transaction.support.DefaultTransactionStatus;
26+
import org.springframework.transaction.support.TransactionSynchronizationManager;
27+
28+
@SuppressWarnings("serial")
29+
public class ResourcelessTransactionManager extends AbstractPlatformTransactionManager {
30+
31+
@Override
32+
protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException {
33+
((ResourcelessTransaction) transaction).begin();
34+
}
35+
36+
@Override
37+
protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
38+
if (logger.isDebugEnabled()) {
39+
logger.debug("Committing resourceless transaction on [" + status.getTransaction() + "]");
40+
}
41+
}
42+
43+
@Override
44+
protected Object doGetTransaction() throws TransactionException {
45+
Object transaction = new ResourcelessTransaction();
46+
List<Object> resources;
47+
if (!TransactionSynchronizationManager.hasResource(this)) {
48+
resources = new ArrayList<>();
49+
TransactionSynchronizationManager.bindResource(this, resources);
50+
}
51+
else {
52+
@SuppressWarnings("unchecked")
53+
List<Object> stack = (List<Object>) TransactionSynchronizationManager.getResource(this);
54+
resources = stack;
55+
}
56+
resources.add(transaction);
57+
return transaction;
58+
}
59+
60+
@Override
61+
protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
62+
if (logger.isDebugEnabled()) {
63+
logger.debug("Rolling back resourceless transaction on [" + status.getTransaction() + "]");
64+
}
65+
}
66+
67+
@Override
68+
protected boolean isExistingTransaction(Object transaction) throws TransactionException {
69+
if (TransactionSynchronizationManager.hasResource(this)) {
70+
List<?> stack = (List<?>) TransactionSynchronizationManager.getResource(this);
71+
return stack.size() > 1;
72+
}
73+
return ((ResourcelessTransaction) transaction).isActive();
74+
}
75+
76+
@Override
77+
protected void doSetRollbackOnly(DefaultTransactionStatus status) throws TransactionException {
78+
}
79+
80+
@Override
81+
protected void doCleanupAfterCompletion(Object transaction) {
82+
List<?> resources = (List<?>) TransactionSynchronizationManager.getResource(this);
83+
resources.clear();
84+
TransactionSynchronizationManager.unbindResource(this);
85+
((ResourcelessTransaction) transaction).clear();
86+
}
87+
88+
private static class ResourcelessTransaction {
89+
90+
private boolean active = false;
91+
92+
public boolean isActive() {
93+
return active;
94+
}
95+
96+
public void begin() {
97+
active = true;
98+
}
99+
100+
public void clear() {
101+
active = false;
102+
}
103+
104+
}
105+
}

0 commit comments

Comments
 (0)