Don't cache tx producers after reset()

Producers were incorrectly returned to the cache after a `reset()`.

**I will back-port; conflicts expected**

Author: garyrussell

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

@@ -124,8 +124,6 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
 
 	private final ThreadLocal<CloseSafeProducer<K, V>> threadBoundProducers = new ThreadLocal<>();
 
-	private final ThreadLocal<Integer> threadBoundProducerEpochs = new ThreadLocal<>();
-
 	private final AtomicInteger epoch = new AtomicInteger();
 
 	private final AtomicInteger clientIdCounter = new AtomicInteger();
@@ -543,7 +541,7 @@ private Producer<K, V> doCreateProducer(@Nullable String txIdPrefix) {
 			}
 			if (this.producer == null) {
 				this.producer = new CloseSafeProducer<>(createKafkaProducer(), this::removeProducer,
-						this.physicalCloseTimeout, this.beanName);
+						this.physicalCloseTimeout, this.beanName, this.epoch.get());
 				this.listeners.forEach(listener -> listener.producerAdded(this.producer.clientId, this.producer));
 			}
 			return this.producer;
@@ -552,22 +550,17 @@ private Producer<K, V> doCreateProducer(@Nullable String txIdPrefix) {
 
 	private Producer<K, V> getOrCreateThreadBoundProducer() {
 		CloseSafeProducer<K, V> tlProducer = this.threadBoundProducers.get();
-		if (this.threadBoundProducerEpochs.get() == null) {
-			this.threadBoundProducerEpochs.set(this.epoch.get());
-		}
-		if (tlProducer != null
-				&& (this.epoch.get() != this.threadBoundProducerEpochs.get() || expire(tlProducer))) {
+		if (tlProducer != null && (this.epoch.get() != tlProducer.epoch || expire(tlProducer))) {
 			closeThreadBoundProducer();
 			tlProducer = null;
 		}
 		if (tlProducer == null) {
 			tlProducer = new CloseSafeProducer<>(createKafkaProducer(), this::removeProducer,
-					this.physicalCloseTimeout, this.beanName);
+					this.physicalCloseTimeout, this.beanName, this.epoch.get());
 			for (Listener<K, V> listener : this.listeners) {
 				listener.producerAdded(tlProducer.clientId, tlProducer);
 			}
 			this.threadBoundProducers.set(tlProducer);
-			this.threadBoundProducerEpochs.set(this.epoch.get());
 		}
 		return tlProducer;
 	}
@@ -699,7 +692,9 @@ boolean cacheReturner(CloseSafeProducer<K, V> producerToRemove, Duration timeout
 		else {
 			synchronized (this.cache) {
 				BlockingQueue<CloseSafeProducer<K, V>> txIdCache = getCache(producerToRemove.txIdPrefix);
-				if (txIdCache != null && !txIdCache.contains(producerToRemove) && !txIdCache.offer(producerToRemove)) {
+				if (producerToRemove.epoch != this.epoch.get()
+						|| (txIdCache != null && !txIdCache.contains(producerToRemove)
+								&& !txIdCache.offer(producerToRemove))) {
 					producerToRemove.closeDelegate(timeout, this.listeners);
 					return true;
 				}
@@ -723,7 +718,8 @@ private CloseSafeProducer<K, V> doCreateTxProducer(String prefix, String suffix,
 		newProducer = createRawProducer(newProducerConfigs);
 		newProducer.initTransactions();
 		CloseSafeProducer<K, V> closeSafeProducer =
-				new CloseSafeProducer<>(newProducer, remover, prefix, this.physicalCloseTimeout, this.beanName);
+				new CloseSafeProducer<>(newProducer, remover, prefix, this.physicalCloseTimeout, this.beanName,
+						this.epoch.get());
 		this.listeners.forEach(listener -> listener.producerAdded(closeSafeProducer.clientId, closeSafeProducer));
 		return closeSafeProducer;
 	}
@@ -799,20 +795,22 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 
 		final String clientId; // NOSONAR
 
+		final int epoch; // NOSONAR
+
 		private volatile Exception producerFailed;
 
 		volatile boolean closed; // NOSONAR
 
 		CloseSafeProducer(Producer<K, V> delegate,
 				BiPredicate<CloseSafeProducer<K, V>, Duration> removeConsumerProducer, Duration closeTimeout,
-				String factoryName) {
+				String factoryName, int epoch) {
 
-			this(delegate, removeConsumerProducer, null, closeTimeout, factoryName);
+			this(delegate, removeConsumerProducer, null, closeTimeout, factoryName, epoch);
 		}
 
 		CloseSafeProducer(Producer<K, V> delegate,
 				BiPredicate<CloseSafeProducer<K, V>, Duration> removeProducer, @Nullable String txIdPrefix,
-				Duration closeTimeout, String factoryName) {
+				Duration closeTimeout, String factoryName, int epoch) {
 
 			Assert.isTrue(!(delegate instanceof CloseSafeProducer), "Cannot double-wrap a producer");
 			this.delegate = delegate;
@@ -830,6 +828,7 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 			}
 			this.clientId = factoryName + "." + id;
 			this.created = System.currentTimeMillis();
+			this.epoch = epoch;
 			LOGGER.debug(() -> "Created new Producer: " + this);
 		}
 

spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.java

@@ -186,6 +186,41 @@ protected Producer createRawProducer(Map configs) {
 		verify(producer).close(any(Duration.class));
 	}
 
+	@Test
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	void dontReturnToCacheAfterReset() {
+		final Producer producer = mock(Producer.class);
+		ApplicationContext ctx = mock(ApplicationContext.class);
+		DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(new HashMap<>()) {
+
+			@Override
+			protected Producer createRawProducer(Map configs) {
+				return producer;
+			}
+
+		};
+		pf.setApplicationContext(ctx);
+		pf.setTransactionIdPrefix("foo");
+		Producer aProducer = pf.createProducer();
+		assertThat(aProducer).isNotNull();
+		aProducer.close();
+		Producer bProducer = pf.createProducer();
+		assertThat(bProducer).isSameAs(aProducer);
+		bProducer.close();
+		assertThat(KafkaTestUtils.getPropertyValue(pf, "producer")).isNull();
+		Map<?, ?> cache = KafkaTestUtils.getPropertyValue(pf, "cache", Map.class);
+		assertThat(cache.size()).isEqualTo(1);
+		Queue queue = (Queue) cache.get("foo");
+		assertThat(queue.size()).isEqualTo(1);
+		bProducer = pf.createProducer();
+		assertThat(bProducer).isSameAs(aProducer);
+		assertThat(queue.size()).isEqualTo(0);
+		pf.reset();
+		bProducer.close();
+		assertThat(queue.size()).isEqualTo(0);
+		pf.destroy();
+	}
+
 	@Test
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	void testThreadLocal() throws InterruptedException {

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

@@ -388,7 +388,7 @@ public Producer<String, String> createProducer(String txIdPrefixArg) {
 							prod.closeDelegate(timeout, Collections.emptyList());
 							return true;
 						},
-						Duration.ofSeconds(1), "factory");
+						Duration.ofSeconds(1), "factory", 0);
 				return closeSafeProducer;
 			}
 
@@ -422,14 +422,14 @@ public Producer<String, String> createProducer(String txIdPrefixArg) {
 				BlockingQueue<CloseSafeProducer<String, String>> cache = new LinkedBlockingDeque<>(1);
 				try {
 					cache.put(new CloseSafeProducer<>(mock(Producer.class), this::removeProducer,
-							Duration.ofSeconds(1), "factory"));
+							Duration.ofSeconds(1), "factory", 0));
 				}
 				catch (@SuppressWarnings("unused") InterruptedException e) {
 					Thread.currentThread().interrupt();
 				}
 				KafkaTestUtils.getPropertyValue(this, "cache", Map.class).put("foo", cache);
 				CloseSafeProducer<String, String> closeSafeProducer = new CloseSafeProducer<>(producer,
-						this::cacheReturner, "foo", Duration.ofSeconds(1), "factory");
+						this::cacheReturner, "foo", Duration.ofSeconds(1), "factory", 0);
 				return closeSafeProducer;
 			}