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 364dde5c..48c52a7e 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
@@ -42,26 +42,34 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.Serializer;
+import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
-import org.springframework.context.Lifecycle;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.ContextStoppedEvent;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
- * The {@link ProducerFactory} implementation for the {@code singleton} shared {@link Producer}
- * instance.
+ * The {@link ProducerFactory} implementation for a {@code singleton} shared
+ * {@link Producer} instance.
*
- * This implementation will produce a new {@link Producer} instance (if transactions are not enabled).
- * for provided {@link Map} {@code configs} and optional {@link Serializer} {@code keySerializer},
- * {@code valueSerializer} implementations on each {@link #createProducer()}
- * invocation.
+ * This implementation will return the same {@link Producer} instance (if transactions are
+ * not enabled) for the provided {@link Map} {@code configs} and optional {@link Serializer}
+ * {@code keySerializer}, {@code valueSerializer} implementations on each
+ * {@link #createProducer()} invocation.
*
- * The {@link Producer} instance is freed from the external {@link Producer#close()} invocation
- * with the internal wrapper. The real {@link Producer#close()} is called on the target
- * {@link Producer} during the {@link Lifecycle#stop()} or {@link DisposableBean#destroy()}.
+ * The {@link Producer} is wrapped and the underlying {@link KafkaProducer} instance is
+ * not actually closed when {@link Producer#close()} is invoked. The {@link KafkaProducer}
+ * is physically closed when {@link DisposableBean#destroy()} is invoked or when the
+ * application context publishes a {@link ContextStoppedEvent}. You can also invoke
+ * {@link #reset()}.
*
- * Setting {@link #setTransactionIdPrefix(String)} enables transactions; in which case, a cache
- * of producers is maintained; closing the producer returns it to the cache.
+ * Setting {@link #setTransactionIdPrefix(String)} enables transactions; in which case, a
+ * cache of producers is maintained; closing a producer returns it to the cache. The
+ * producers are closed and the cache is cleared when the factory is destroyed, the
+ * application context stopped, or the {@link #reset()} method is called.
*
* @param the key type.
* @param the value type.
@@ -70,7 +78,8 @@ import org.springframework.util.Assert;
* @author Murali Reddy
* @author Nakul Mishra
*/
-public class DefaultKafkaProducerFactory implements ProducerFactory, Lifecycle, DisposableBean {
+public class DefaultKafkaProducerFactory implements ProducerFactory, ApplicationContextAware,
+ ApplicationListener, DisposableBean {
private static final int DEFAULT_PHYSICAL_CLOSE_TIMEOUT = 30;
@@ -92,7 +101,7 @@ public class DefaultKafkaProducerFactory implements ProducerFactory,
private String transactionIdPrefix;
- private volatile boolean running;
+ private ApplicationContext applicationContext;
/**
* Construct a factory with the provided configuration.
@@ -116,6 +125,11 @@ public class DefaultKafkaProducerFactory implements ProducerFactory,
this.valueSerializer = valueSerializer;
}
+ @Override
+ public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+ this.applicationContext = applicationContext;
+ }
+
public void setKeySerializer(@Nullable Serializer keySerializer) {
this.keySerializer = keySerializer;
}
@@ -192,26 +206,53 @@ public class DefaultKafkaProducerFactory implements ProducerFactory,
}
@Override
- public void start() {
- this.running = true;
+ public void onApplicationEvent(ContextStoppedEvent event) {
+ if (event.getApplicationContext().equals(this.applicationContext)) {
+ reset();
+ }
}
+ /**
+ * NoOp.
+ * @deprecated {@link org.springframework.context.Lifecycle} is no longer implemented.
+ */
+ @Deprecated
+ public void start() {
+ // NOSONAR
+ }
- @Override
+ /**
+ * NoOp.
+ * @deprecated {@link org.springframework.context.Lifecycle} is no longer implemented;
+ * use {@link #reset()} to close the {@link Producer}(s).
+ */
+ @Deprecated
public void stop() {
+ reset();
+ }
+
+ /**
+ * Close the {@link Producer}(s) and clear the cache of transactional
+ * {@link Producer}(s).
+ * @since 2.2
+ */
+ public void reset() {
try {
destroy();
- this.running = false;
}
catch (Exception e) {
logger.error("Exception while closing producer", e);
}
}
-
- @Override
+ /**
+ * NoOp.
+ * @return always true.
+ * @deprecated {@link org.springframework.context.Lifecycle} is no longer implemented.
+ */
+ @Deprecated
public boolean isRunning() {
- return this.running;
+ return true;
}
@Override
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.java
index 653c7f8a..344e02cd 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.java
@@ -23,6 +23,7 @@ import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import java.util.HashMap;
+import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
@@ -31,6 +32,8 @@ import org.apache.kafka.common.KafkaException;
import org.junit.jupiter.api.Test;
import org.mockito.InOrder;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.event.ContextStoppedEvent;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import org.springframework.transaction.CannotCreateTransactionException;
@@ -98,4 +101,54 @@ public class DefaultKafkaProducerFactoryTests {
pf.destroy();
}
+ @Test
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ public void testResetSingle() throws Exception {
+ final Producer producer = mock(Producer.class);
+ DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(new HashMap<>()) {
+
+ @Override
+ protected Producer createKafkaProducer() {
+ return producer;
+ }
+
+ };
+ Producer aProducer = pf.createProducer();
+ assertThat(aProducer).isNotNull();
+ aProducer.close();
+ assertThat(KafkaTestUtils.getPropertyValue(pf, "producer")).isNotNull();
+ Queue cache = KafkaTestUtils.getPropertyValue(pf, "cache", Queue.class);
+ assertThat(cache.size()).isEqualTo(0);
+ pf.reset();
+ assertThat(KafkaTestUtils.getPropertyValue(pf, "producer")).isNull();
+ }
+
+ @Test
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ public void testResetTx() throws Exception {
+ final Producer producer = mock(Producer.class);
+ ApplicationContext ctx = mock(ApplicationContext.class);
+ DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(new HashMap<>()) {
+
+ @Override
+ protected Producer createTransactionalProducer() {
+ producer.initTransactions();
+ BlockingQueue cache = getCache();
+ Producer cached = cache.poll();
+ return cached == null ? new CloseSafeProducer(producer, cache) : cached;
+ }
+
+ };
+ pf.setApplicationContext(ctx);
+ pf.setTransactionIdPrefix("foo");
+ Producer aProducer = pf.createProducer();
+ assertThat(aProducer).isNotNull();
+ aProducer.close();
+ assertThat(KafkaTestUtils.getPropertyValue(pf, "producer")).isNull();
+ Queue cache = KafkaTestUtils.getPropertyValue(pf, "cache", Queue.class);
+ assertThat(cache.size()).isEqualTo(1);
+ pf.onApplicationEvent(new ContextStoppedEvent(ctx));
+ assertThat(cache.size()).isEqualTo(0);
+ }
+
}