Close thread bound producers after reset

**cherry-pick to 2.4.x, 2.3.x**

Author: garyrussell

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

@@ -126,6 +126,10 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 
 	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();
 
 	private Supplier<Serializer<K>> keySerializerSupplier;
@@ -331,6 +335,7 @@ public void destroy() {
 					(k, v) -> v.getDelegate().close(this.physicalCloseTimeout));
 			this.consumerProducers.clear();
 		}
+		this.epoch.incrementAndGet();
 	}
 
 	@Override
@@ -382,10 +387,18 @@ public Producer<K, V> createProducer(@Nullable String txIdPrefixArg) {
 		}
 		if (this.producerPerThread) {
 			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()) {
+				closeThreadBoundProducer();
+				tlProducer = null;
+			}
 			if (tlProducer == null) {
 				tlProducer = new CloseSafeProducer<>(createKafkaProducer(), this::removeProducer,
 						this.physicalCloseTimeout);
 				this.threadBoundProducers.set(tlProducer);
+				this.threadBoundProducerEpochs.set(this.epoch.get());
 			}
 			return tlProducer;
 		}

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

@@ -202,6 +202,35 @@ protected Producer createKafkaProducer() {
 		verify(producer).close(any(Duration.class));
 	}
 
+	@Test
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	void testThreadLocalReset() {
+		Producer producer1 = mock(Producer.class);
+		Producer producer2 = mock(Producer.class);
+		ProducerFactory mockPf = mock(ProducerFactory.class);
+		given(mockPf.createProducer()).willReturn(producer1, producer2);
+		DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(new HashMap<>()) {
+
+			@Override
+			protected Producer createKafkaProducer() {
+				return mockPf.createProducer();
+			}
+
+		};
+		pf.setProducerPerThread(true);
+		Producer aProducer = pf.createProducer();
+		assertThat(aProducer).isNotNull();
+		aProducer.close();
+		Producer bProducer = pf.createProducer();
+		assertThat(bProducer).isSameAs(aProducer);
+		bProducer.close();
+		pf.reset();
+		bProducer = pf.createProducer();
+		assertThat(bProducer).isNotSameAs(aProducer);
+		bProducer.close();
+		verify(producer1).close(any(Duration.class));
+	}
+
 	@Test
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	void testCleanUpAfterTxFence() {