diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java index 87eee40d..6f532e30 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java @@ -48,6 +48,7 @@ import org.apache.kafka.common.serialization.Serializer; import org.springframework.beans.factory.DisposableBean; import org.springframework.context.Lifecycle; import org.springframework.kafka.support.TransactionSupport; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** @@ -304,12 +305,13 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, } private CloseSafeProducer doCreateTxProducer(String suffix, Consumer> remover) { - Producer producer; - Map configs = new HashMap<>(this.configs); - configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, this.transactionIdPrefix + suffix); - producer = new KafkaProducer(configs, this.keySerializer, this.valueSerializer); - producer.initTransactions(); - return new CloseSafeProducer(producer, this.cache, remover); + Producer newProducer; + Map newProducerConfigs = new HashMap<>(this.configs); + newProducerConfigs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, this.transactionIdPrefix + suffix); + newProducer = new KafkaProducer(newProducerConfigs, this.keySerializer, this.valueSerializer); + newProducer.initTransactions(); + return new CloseSafeProducer(newProducer, this.cache, remover, + (String) newProducerConfigs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG)); } protected BlockingQueue> getCache() { @@ -343,6 +345,8 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, private final Consumer> removeConsumerProducer; + private final String txId; + private volatile boolean txFailed; CloseSafeProducer(Producer delegate) { @@ -356,9 +360,17 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, CloseSafeProducer(Producer delegate, BlockingQueue> cache, Consumer> removeConsumerProducer) { + + this(delegate, cache, removeConsumerProducer, null); + } + + CloseSafeProducer(Producer delegate, @Nullable BlockingQueue> cache, + @Nullable Consumer> removeConsumerProducer, @Nullable String txId) { + this.delegate = delegate; this.cache = cache; this.removeConsumerProducer = removeConsumerProducer; + this.txId = txId; } @Override @@ -393,10 +405,16 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, @Override public void beginTransaction() throws ProducerFencedException { + if (logger.isDebugEnabled()) { + logger.debug("beginTransaction: " + this); + } try { this.delegate.beginTransaction(); } catch (RuntimeException e) { + if (logger.isErrorEnabled()) { + logger.error("beginTransaction failed: " + this, e); + } this.txFailed = true; throw e; } @@ -405,15 +423,22 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, @Override public void sendOffsetsToTransaction(Map offsets, String consumerGroupId) throws ProducerFencedException { + this.delegate.sendOffsetsToTransaction(offsets, consumerGroupId); } @Override public void commitTransaction() throws ProducerFencedException { + if (logger.isDebugEnabled()) { + logger.debug("commitTransaction: " + this); + } try { this.delegate.commitTransaction(); } catch (RuntimeException e) { + if (logger.isErrorEnabled()) { + logger.error("commitTransaction failed: " + this, e); + } this.txFailed = true; throw e; } @@ -421,10 +446,16 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, @Override public void abortTransaction() throws ProducerFencedException { + if (logger.isDebugEnabled()) { + logger.debug("abortTransaction: " + this); + } try { this.delegate.abortTransaction(); } catch (RuntimeException e) { + if (logger.isErrorEnabled()) { + logger.error("Abort failed: " + this, e); + } this.txFailed = true; throw e; } @@ -432,34 +463,50 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, @Override public void close() { + close(0, null); + } + + @Override + public void close(long timeout, @Nullable TimeUnit unit) { if (this.cache != null) { if (this.txFailed) { - logger.warn("Error during transactional operation; producer removed from cache; possible cause: " - + "broker restarted during transaction"); - - this.delegate.close(); + if (logger.isWarnEnabled()) { + logger.warn("Error during transactional operation; producer removed from cache; possible cause: " + + "broker restarted during transaction: " + this); + } + if (unit == null) { + this.delegate.close(); + } + else { + this.delegate.close(timeout, unit); + } if (this.removeConsumerProducer != null) { this.removeConsumerProducer.accept(this); } } else { - synchronized (this) { - if (!this.cache.contains(this)) { - this.cache.offer(this); + if (this.removeConsumerProducer == null) { // dedicated consumer producers are not cached + synchronized (this) { + if (!this.cache.contains(this) + && !this.cache.offer(this)) { + if (unit == null) { + this.delegate.close(); + } + else { + this.delegate.close(timeout, unit); + } + } } } } } } - @Override - public void close(long timeout, TimeUnit unit) { - close(); - } - @Override public String toString() { - return "CloseSafeProducer [delegate=" + this.delegate + "]"; + return "CloseSafeProducer [delegate=" + this.delegate + "" + + (this.txId != null ? ", txId=" + this.txId : "") + + "]"; } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java index b1cee31c..22d5eb55 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java @@ -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. @@ -20,22 +20,50 @@ import static org.assertj.core.api.Assertions.assertThat; import java.util.Collections; import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.producer.ProducerConfig; import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.listener.KafkaMessageListenerContainer; +import org.springframework.kafka.listener.MessageListener; +import org.springframework.kafka.listener.config.ContainerProperties; +import org.springframework.kafka.support.SendResult; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.kafka.transaction.KafkaTransactionManager; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.util.concurrent.ListenableFuture; /** * @author Gary Russell * @since 1.0.6 * */ +@EmbeddedKafka(topics = { "txCache1", "txCache2", "txCacheSendFromListener" }, + brokerProperties = { + "transaction.state.log.replication.factor=1", + "transaction.state.log.min.isr=1" } +) +@RunWith(SpringRunner.class) public class DefaultKafkaConsumerFactoryTests { + @Autowired + private KafkaEmbedded embeddedKafka; + @Test public void testClientId() { Map configs = Collections.singletonMap(ConsumerConfig.CLIENT_ID_CONFIG, "foo"); - DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory(configs) { + DefaultKafkaConsumerFactory factory = + new DefaultKafkaConsumerFactory(configs) { @Override protected KafkaConsumer createKafkaConsumer(Map configs) { @@ -47,4 +75,87 @@ public class DefaultKafkaConsumerFactoryTests { factory.createConsumer("-1"); } + @SuppressWarnings("unchecked") + @Test + public void testNestedTxProducerIsCached() throws Exception { + Map producerProps = KafkaTestUtils.producerProps(this.embeddedKafka); + producerProps.put(ProducerConfig.RETRIES_CONFIG, 1); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(producerProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + DefaultKafkaProducerFactory pfTx = new DefaultKafkaProducerFactory<>(producerProps); + pfTx.setTransactionIdPrefix("fooTx."); + KafkaTemplate templateTx = new KafkaTemplate<>(pfTx); + Map consumerProps = KafkaTestUtils.consumerProps("txCache1Group", "false", this.embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + ContainerProperties containerProps = new ContainerProperties("txCache1"); + CountDownLatch latch = new CountDownLatch(1); + containerProps.setMessageListener((MessageListener) r -> { + templateTx.executeInTransaction(t -> t.send("txCacheSendFromListener", "bar")); + templateTx.executeInTransaction(t -> t.send("txCacheSendFromListener", "baz")); + latch.countDown(); + }); + KafkaTransactionManager tm = new KafkaTransactionManager<>(pfTx); + containerProps.setTransactionManager(tm); + KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, + containerProps); + container.start(); + try { + ListenableFuture> future = template.send("txCache1", "foo"); + future.get(); + assertThat(KafkaTestUtils.getPropertyValue(pf, "cache", BlockingQueue.class)).hasSize(0); + assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue(); + assertThat(KafkaTestUtils.getPropertyValue(pfTx, "cache", BlockingQueue.class)).hasSize(1); + } + finally { + container.stop(); + pf.destroy(); + pfTx.destroy(); + } + } + + @SuppressWarnings("unchecked") + @Test + public void testContainerTxProducerIsNotCached() throws Exception { + Map producerProps = KafkaTestUtils.producerProps(this.embeddedKafka); + producerProps.put(ProducerConfig.RETRIES_CONFIG, 1); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(producerProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + DefaultKafkaProducerFactory pfTx = new DefaultKafkaProducerFactory<>(producerProps); + pfTx.setTransactionIdPrefix("fooTx."); + KafkaTemplate templateTx = new KafkaTemplate<>(pfTx); + Map consumerProps = KafkaTestUtils.consumerProps("txCache2Group", "false", this.embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + ContainerProperties containerProps = new ContainerProperties("txCache2"); + CountDownLatch latch = new CountDownLatch(1); + containerProps.setMessageListener((MessageListener) r -> { + templateTx.send("txCacheSendFromListener", "bar"); + templateTx.send("txCacheSendFromListener", "baz"); + latch.countDown(); + }); + KafkaTransactionManager tm = new KafkaTransactionManager<>(pfTx); + containerProps.setTransactionManager(tm); + KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, + containerProps); + container.start(); + try { + ListenableFuture> future = template.send("txCache2", "foo"); + future.get(); + assertThat(KafkaTestUtils.getPropertyValue(pf, "cache", BlockingQueue.class)).hasSize(0); + assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue(); + assertThat(KafkaTestUtils.getPropertyValue(pfTx, "cache", BlockingQueue.class)).hasSize(0); + } + finally { + container.stop(); + pf.destroy(); + pfTx.destroy(); + } + } + + @Configuration + public static class Config { + + } + }