Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-646: Transaction Improvements

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/646

- add protection to the transactional producer cache to avoid multiple cache inserts
- add try/catch around `abortTransaction()`
- detect (and reject) an illegal call to `template.send()` when there is no existing transaction
- from the send callback remove the `ThreadLocal` check (will always be null) and only close the producer if it's not transactional
- remove autoFlush from template tests
This commit is contained in:
Gary Russell
2018-04-13 14:32:45 -04:00
committed by Artem Bilan
parent 9f724ea27a
commit 42e2d01026
4 changed files with 44 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 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.
@@ -302,8 +302,8 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
}
@Override
public void close() {
if (this.cache != null) {
public synchronized void close() {
if (this.cache != null && !this.cache.contains(this)) {
this.cache.offer(this);
}
}

30
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java
View File

@@ -265,10 +265,14 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
result = callback.doInOperations(this);
}
catch (Exception e) {
producer.abortTransaction();
this.producers.remove();
closeProducer(producer, false);
producer = null;
try {
producer.abortTransaction();
}
finally {
this.producers.remove();
closeProducer(producer, false);
producer = null;
}
}
if (producer != null) {
try {
@@ -328,12 +332,19 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
* @return a Future for the {@link RecordMetadata}.
*/
protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> producerRecord) {
if (this.transactional) {
Assert.state(inTransaction(),
"No transaction is in process; "
+ "possible solutions: run the template operation within the scope of a "
+ "template.executeInTransaction() operation, start a transaction with @Transactional "
+ "before invoking the template method, "
+ "run in a transaction started by a listener container when consuming a record");
}
final Producer<K, V> producer = getTheProducer();
if (this.logger.isTraceEnabled()) {
this.logger.trace("Sending: " + producerRecord);
}
final SettableListenableFuture<SendResult<K, V>> future = new SettableListenableFuture<>();
final boolean inLocalTx = inTransaction();
producer.send(producerRecord, new Callback() {
@Override
@@ -365,8 +376,8 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
}
}
finally {
if (KafkaTemplate.this.producers.get() == null) {
closeProducer(producer, inLocalTx);
if (!KafkaTemplate.this.transactional) {
closeProducer(producer, false);
}
}
}
@@ -381,8 +392,11 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
return future;
}
protected boolean inTransaction() {
return this.producers.get() != null || TransactionSynchronizationManager.isActualTransactionActive();
return this.transactional && (this.producers.get() != null
|| TransactionSynchronizationManager.getResource(this.producerFactory) != null
|| TransactionSynchronizationManager.isActualTransactionActive());
}
private Producer<K, V> getTheProducer() {

4
spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminBadContextTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 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.
@@ -41,7 +41,7 @@ public class KafkaAdminBadContextTests {
@Test
public void testContextNotLoaded() {
try {
new AnnotationConfigApplicationContext(BadConfig.class);
new AnnotationConfigApplicationContext(BadConfig.class).close();
fail("Expected Exception");
}
catch (IllegalStateException e) {

19
spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.kafka.core;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.BDDMockito.given;
@@ -83,7 +84,7 @@ public class KafkaTemplateTransactionTests {
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
pf.setKeySerializer(new StringSerializer());
pf.setTransactionIdPrefix("my.transaction.");
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf, true);
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(STRING_KEY_TOPIC);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testTxString", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
@@ -119,7 +120,7 @@ public class KafkaTemplateTransactionTests {
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
pf.setKeySerializer(new StringSerializer());
pf.setTransactionIdPrefix("my.transaction.");
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf, true);
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(STRING_KEY_TOPIC);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testTxString", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
@@ -194,6 +195,20 @@ public class KafkaTemplateTransactionTests {
.get(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)).isEqualTo(false);
}
@Test
public void testNoTx() {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
senderProps.put(ProducerConfig.RETRIES_CONFIG, 1);
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
pf.setKeySerializer(new StringSerializer());
pf.setTransactionIdPrefix("my.transaction.");
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(STRING_KEY_TOPIC);
assertThatThrownBy(() -> template.send("foo", "bar"))
.isInstanceOf(IllegalStateException.class)
.hasMessageContaining("No transaction is in process;");
}
@Configuration
@EnableTransactionManagement
public static class DeclarativeConfig {