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

GH-493: set enable.idempotence to true by default

Browse Source 
Fixes: spring-projects/spring-kafka#493

* Polishing `if()` statement in the `DefaultKafkaProducerFactory`
* Add author name to the test class
This commit is contained in:
Nakul Mishra
2017-11-28 02:19:48 +01:00
committed by Artem Bilan
parent 8c65d62a24
commit ebd01beba8
2 changed files with 38 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -67,6 +67,7 @@ import org.springframework.util.Assert;
*
* @author Gary Russell
* @author Murali Reddy
* @author Nakul Mishra
*/
public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>, Lifecycle, DisposableBean {
@@ -129,6 +130,18 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
public void setTransactionIdPrefix(String transactionIdPrefix) {
Assert.notNull(transactionIdPrefix, "'transactionIdPrefix' cannot be null");
this.transactionIdPrefix = transactionIdPrefix;
enableIdempotentBehaviour();
}
/**
* When set to 'true', the producer will ensure that exactly one copy of each message is written in the stream.
*/
private void enableIdempotentBehaviour() {
Object previousValue = this.configs.putIfAbsent(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
if (logger.isDebugEnabled() && Boolean.FALSE.equals(previousValue)) {
logger.debug("The '" + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG +
"' is set to false, may result in duplicate messages");
}
}
/**

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

@@ -60,6 +60,8 @@ import org.springframework.transaction.support.TransactionTemplate;
/**
* @author Gary Russell
* @author Nakul Mishra
*
* @since 1.3
*
*/
@@ -165,6 +167,29 @@ public class KafkaTemplateTransactionTests {
ctx.close();
}
@Test
public void testDefaultProducerIdempotentConfig() throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(
senderProps);
pf.setTransactionIdPrefix("my.transaction.");
pf.destroy();
assertThat(pf.getConfigurationProperties()
.get(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)).isEqualTo(true);
}
@Test
public void testOverrideProducerIdempotentConfig() throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
senderProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false);
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(
senderProps);
pf.setTransactionIdPrefix("my.transaction.");
pf.destroy();
assertThat(pf.getConfigurationProperties()
.get(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)).isEqualTo(false);
}
@Configuration
@EnableTransactionManagement
public static class DeclarativeConfig {