GH-1615: Add maxAge to Producers

Resolves #1615

**cherry-pick to 2.5.x**

# Conflicts:
#	src/reference/asciidoc/kafka.adoc
#	src/reference/asciidoc/whats-new.adoc

Author: garyrussell

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

@@ -152,6 +152,8 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
 
 	private String clientIdPrefix;
 
+	private long maxAge;
+
 	private volatile CloseSafeProducer<K, V> producer;
 
 	/**
@@ -349,6 +351,16 @@ public List<ProducerPostProcessor<K, V>> getPostProcessors() {
 		return Collections.unmodifiableList(this.postProcessors);
 	}
 
+	/**
+	 * Set the maximum age for a producer; useful when using transactions and the broker
+	 * might expire a {@code transactional.id} due to inactivity.
+	 * @param maxAge the maxAge to set
+	 * @since 2.5.8
+	 */
+	public void setMaxAge(Duration maxAge) {
+		this.maxAge = maxAge.toMillis();
+	}
+
 	/**
 	 * Add a listener.
 	 * @param listener the listener.
@@ -498,7 +510,8 @@ private Producer<K, V> doCreateProducer(@Nullable String txIdPrefix) {
 			if (this.threadBoundProducerEpochs.get() == null) {
 				this.threadBoundProducerEpochs.set(this.epoch.get());
 			}
-			if (tlProducer != null && this.epoch.get() != this.threadBoundProducerEpochs.get()) {
+			if (tlProducer != null
+					&& (this.epoch.get() != this.threadBoundProducerEpochs.get() || expire(tlProducer))) {
 				closeThreadBoundProducer();
 				tlProducer = null;
 			}
@@ -514,6 +527,9 @@ private Producer<K, V> doCreateProducer(@Nullable String txIdPrefix) {
 			return tlProducer;
 		}
 		synchronized (this) {
+			if (this.producer != null && expire(this.producer)) {
+				this.producer = null;
+			}
 			if (this.producer == null) {
 				this.producer = new CloseSafeProducer<>(createKafkaProducer(), this::removeProducer,
 						this.physicalCloseTimeout, this.beanName);
@@ -552,14 +568,15 @@ protected Producer<K, V> createTransactionalProducerForPartition(String txIdPref
 		}
 		else {
 			synchronized (this.consumerProducers) {
-				if (!this.consumerProducers.containsKey(suffix)) {
+				CloseSafeProducer<K, V> consumerProducer = this.consumerProducers.get(suffix);
+				if (consumerProducer == null || expire(consumerProducer)) {
 					CloseSafeProducer<K, V> newProducer = doCreateTxProducer(txIdPrefix, suffix,
 							this::removeConsumerProducer);
 					this.consumerProducers.put(suffix, newProducer);
 					return newProducer;
 				}
 				else {
-					return this.consumerProducers.get(suffix);
+					return consumerProducer;
 				}
 			}
 		}
@@ -616,7 +633,15 @@ protected Producer<K, V> createTransactionalProducer() {
 	protected Producer<K, V> createTransactionalProducer(String txIdPrefix) {
 		BlockingQueue<CloseSafeProducer<K, V>> queue = getCache(txIdPrefix);
 		Assert.notNull(queue, () -> "No cache found for " + txIdPrefix);
-		Producer<K, V> cachedProducer = queue.poll();
+		CloseSafeProducer<K, V> cachedProducer = queue.poll();
+		while (cachedProducer != null) {
+			if (expire(cachedProducer)) {
+				cachedProducer = queue.poll();
+			}
+			else {
+				break;
+			}
+		}
 		if (cachedProducer == null) {
 			return doCreateTxProducer(txIdPrefix, "" + this.transactionIdSuffix.getAndIncrement(), this::cacheReturner);
 		}
@@ -625,6 +650,14 @@ protected Producer<K, V> createTransactionalProducer(String txIdPrefix) {
 		}
 	}
 
+	private boolean expire(CloseSafeProducer<K, V> producer) {
+		boolean expired = this.maxAge > 0 && System.currentTimeMillis() - producer.created > this.maxAge;
+		if (expired) {
+			producer.closeDelegate(this.physicalCloseTimeout, this.listeners);
+		}
+		return expired;
+	}
+
 	boolean cacheReturner(CloseSafeProducer<K, V> producerToRemove, Duration timeout) {
 		if (producerToRemove.closed) {
 			producerToRemove.closeDelegate(timeout, this.listeners);
@@ -727,6 +760,8 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 
 		final String txIdPrefix; // NOSONAR
 
+		final long created;
+
 		private final Duration closeTimeout;
 
 		final String clientId; // NOSONAR
@@ -761,6 +796,7 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 				id = "unknown";
 			}
 			this.clientId = factoryName + "." + id;
+			this.created = System.currentTimeMillis();
 			LOGGER.debug(() -> "Created new Producer: " + this);
 		}
 

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

@@ -25,6 +25,7 @@
 import static org.mockito.BDDMockito.willThrow;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import java.time.Duration;
@@ -122,9 +123,9 @@ protected Producer createRawProducer(Map configs) {
 
 	@Test
 	@SuppressWarnings({ "rawtypes", "unchecked" })
-	void testResetSingle() {
+	void testResetSingle() throws InterruptedException {
 		final Producer producer = mock(Producer.class);
-		ProducerFactory pf = new DefaultKafkaProducerFactory(new HashMap<>()) {
+		DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(new HashMap<>()) {
 
 			@Override
 			protected Producer createRawProducer(Map configs) {
@@ -134,13 +135,19 @@ protected Producer createRawProducer(Map configs) {
 		};
 		Producer aProducer = pf.createProducer();
 		assertThat(aProducer).isNotNull();
+		Producer bProducer = pf.createProducer();
+		assertThat(bProducer).isSameAs(aProducer);
 		aProducer.close(ProducerFactoryUtils.DEFAULT_CLOSE_TIMEOUT);
 		assertThat(KafkaTestUtils.getPropertyValue(pf, "producer")).isNotNull();
+		pf.setMaxAge(Duration.ofMillis(10));
+		Thread.sleep(50);
+		aProducer = pf.createProducer();
+		assertThat(aProducer).isNotSameAs(bProducer);
 		Map<?, ?> cache = KafkaTestUtils.getPropertyValue(pf, "cache", Map.class);
 		assertThat(cache.size()).isEqualTo(0);
 		pf.reset();
 		assertThat(KafkaTestUtils.getPropertyValue(pf, "producer")).isNull();
-		verify(producer).close(any(Duration.class));
+		verify(producer, times(2)).close(any(Duration.class));
 	}
 
 	@Test
@@ -161,28 +168,34 @@ protected Producer createRawProducer(Map configs) {
 		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);
+		pf.setMaxAge(Duration.ofMillis(10));
+		Thread.sleep(50);
+		aProducer = pf.createProducer();
+		assertThat(aProducer).isNotSameAs(bProducer);
 		pf.onApplicationEvent(new ContextStoppedEvent(ctx));
 		assertThat(queue.size()).isEqualTo(0);
 		verify(producer).close(any(Duration.class));
 	}
 
 	@Test
 	@SuppressWarnings({ "rawtypes", "unchecked" })
-	void testThreadLocal() {
+	void testThreadLocal() throws InterruptedException {
 		final Producer producer = mock(Producer.class);
+		AtomicBoolean created = new AtomicBoolean();
 		DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(new HashMap<>()) {
 
-			boolean created;
-
 			@Override
 			protected Producer createKafkaProducer() {
-				assertThat(this.created).isFalse();
-				this.created = true;
+				assertThat(created.get()).isFalse();
+				created.set(true);
 				return producer;
 			}
 
@@ -196,9 +209,14 @@ protected Producer createKafkaProducer() {
 		bProducer.close();
 		assertThat(KafkaTestUtils.getPropertyValue(pf, "producer")).isNull();
 		assertThat(KafkaTestUtils.getPropertyValue(pf, "threadBoundProducers", ThreadLocal.class).get()).isNotNull();
+		pf.setMaxAge(Duration.ofMillis(10));
+		Thread.sleep(50);
+		created.set(false);
+		aProducer = pf.createProducer();
+		assertThat(aProducer).isNotSameAs(bProducer);
 		pf.closeThreadBoundProducer();
 		assertThat(KafkaTestUtils.getPropertyValue(pf, "threadBoundProducers", ThreadLocal.class).get()).isNull();
-		verify(producer).close(any(Duration.class));
+		verify(producer, times(3)).close(any(Duration.class));
 	}
 
 	@Test

src/reference/asciidoc/kafka.adoc

@@ -3030,6 +3030,11 @@ This is `true` by default since version 2.5 when transactions are enabled with `
 
 Also see <<transaction-id-prefix>>.
 
+IMPORTANT: Starting with version 2.5.8, you can now configure the `maxAge` property on the producer factory.
+This is useful when using transactional producers that might lay idle for the broker's `transactional.id.expiration.ms`.
+With current `kafka-clients`, this can cause a `ProducerFencedException` without a rebalance.
+By setting the `maxAge` to less than `transactional.id.expiration.ms`, the factory will refresh the producer if it is past it's max age.
+
 ===== Using `KafkaTransactionManager`
 
 The `KafkaTransactionManager` is an implementation of Spring Framework's `PlatformTransactionManager`.