GH-1196: Use close(Duration) instead of close()

Resolves https://github.com/spring-projects/spring-kafka/issues/1196

Add `closeTimeout` to `KafkaTemplate` and `KafkaTransactionManager` (default 5s).
Use a zero timeout if a transaction operation failed with a timeout.

Deprecate 1.3.x public APIs

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

@@ -16,6 +16,9 @@
 
 package org.springframework.kafka.core;
 
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -43,7 +46,9 @@ import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.utils.AppInfoParser;
 
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.DisposableBean;
@@ -147,8 +152,10 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 	}
 
 	/**
-	 * The time to wait when physically closing the producer (when {@link #stop()} or {@link #destroy()} is invoked).
-	 * Specified in seconds; default {@value #DEFAULT_PHYSICAL_CLOSE_TIMEOUT}.
+	 * The time to wait when physically closing the producer via the factory rather than
+	 * closing the producer itself (when {@link #reset()}, {@link #destroy() or
+	 * #closeProducerFor(String)} are invoked). Specified in seconds; default
+	 * {@link #DEFAULT_PHYSICAL_CLOSE_TIMEOUT}.
 	 * @param physicalCloseTimeout the timeout in seconds.
 	 * @since 1.0.7
 	 */
@@ -216,7 +223,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 
 	@SuppressWarnings("resource")
 	@Override
-	public void destroy() throws Exception { //NOSONAR
+	public void destroy() {
 		CloseSafeProducer<K, V> producerToClose = this.producer;
 		this.producer = null;
 		if (producerToClose != null) {
@@ -400,6 +407,25 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 	 */
 	protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 
+		private static final Duration CLOSE_TIMEOUT_AFTER_TX_TIMEOUT = Duration.ofMillis(0);
+
+		private static final Method CLOSE_WITH_DURATION;
+
+		static {
+			Method method = null;
+			String clientVersion = AppInfoParser.getVersion();
+			try {
+				if (!clientVersion.startsWith("1.") && !clientVersion.startsWith("2.0.")
+						&& !clientVersion.startsWith("2.1.")) {
+					method = KafkaProducer.class.getDeclaredMethod("close", Duration.class);
+				}
+			}
+			catch (NoSuchMethodException e) {
+				logger.error("Failed to get close(Duration) method for version: " + clientVersion, e);
+			}
+			CLOSE_WITH_DURATION = method;
+		}
+
 		private final Producer<K, V> delegate;
 
 		private final BlockingQueue<CloseSafeProducer<K, V>> cache;
@@ -408,7 +434,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 
 		private final String txId;
 
-		private volatile boolean txFailed;
+		private volatile Exception txFailed;
 
 		CloseSafeProducer(Producer<K, V> delegate) {
 			this(delegate, null, null);
@@ -476,7 +502,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 				if (logger.isErrorEnabled()) {
 					logger.error("beginTransaction failed: " + this, e);
 				}
-				this.txFailed = true;
+				this.txFailed = e;
 				throw e;
 			}
 		}
@@ -500,7 +526,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 				if (logger.isErrorEnabled()) {
 					logger.error("commitTransaction failed: " + this, e);
 				}
-				this.txFailed = true;
+				this.txFailed = e;
 				throw e;
 			}
 		}
@@ -517,7 +543,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 				if (logger.isErrorEnabled()) {
 					logger.error("Abort failed: " + this, e);
 				}
-				this.txFailed = true;
+				this.txFailed = e;
 				throw e;
 			}
 		}
@@ -530,17 +556,16 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 		@Override
 		public void close(long timeout, @Nullable TimeUnit unit) {
 			if (this.cache != null) {
-				if (this.txFailed) {
+				Duration closeTimeout = this.txFailed instanceof TimeoutException || unit == null
+						? CLOSE_TIMEOUT_AFTER_TX_TIMEOUT
+						: Duration.ofMillis(unit.toMillis(timeout));
+				if (this.txFailed != null) {
 					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);
+						logger.warn("Error during transactional operation; producer removed from cache; "
+								+ "possible cause: "
+								+ "broker restarted during transaction: " + this);
 					}
+					closeDelegate(closeTimeout);
 					if (this.removeConsumerProducer != null) {
 						this.removeConsumerProducer.accept(this);
 					}
@@ -550,12 +575,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 						synchronized (this) {
 							if (!this.cache.contains(this)
 									&& !this.cache.offer(this)) {
-								if (unit == null) {
-									this.delegate.close();
-								}
-								else {
-									this.delegate.close(timeout, unit);
-								}
+								closeDelegate(closeTimeout);
 							}
 						}
 					}
@@ -563,6 +583,20 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 			}
 		}
 
+		private void closeDelegate(Duration closeTimeout) {
+			if (CLOSE_WITH_DURATION != null) {
+				try {
+					CLOSE_WITH_DURATION.invoke(this.delegate, closeTimeout);
+				}
+				catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
+					logger.error("Failed to invoke close(Duration) with reflection", e);
+				}
+			}
+			else {
+				this.delegate.close(closeTimeout.toMillis(), TimeUnit.MILLISECONDS);
+			}
+		}
+
 		@Override
 		public String toString() {
 			return "CloseSafeProducer [delegate=" + this.delegate + ""

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

@@ -16,9 +16,13 @@
 
 package org.springframework.kafka.core;
 
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
 import org.apache.kafka.clients.producer.Producer;
 
 import org.springframework.transaction.support.ResourceHolderSupport;
+import org.springframework.util.Assert;
 
 /**
  * Kafka resource holder, wrapping a Kafka producer. KafkaTransactionManager binds instances of this
@@ -33,12 +37,41 @@ public class KafkaResourceHolder<K, V> extends ResourceHolderSupport {
 
 	private final Producer<K, V> producer;
 
+	private final Duration closeTimeout;
+
 	/**
 	 * Construct an instance for the producer.
 	 * @param producer the producer.
 	 */
 	public KafkaResourceHolder(Producer<K, V> producer) {
+		this(producer, ProducerFactoryUtils.DEFAULT_CLOSE_TIMEOUT);
+	}
+
+	/**
+	 * Construct an instance for the producer.
+	 * @param producer the producer.
+	 * @param closeTimeout the close timeout.
+	 * @deprecated in favor of {@link #KafkaResourceHolder(Producer, Duration)}
+	 * @since 1.3.11
+	 */
+	@Deprecated
+	public KafkaResourceHolder(Producer<K, V> producer, long closeTimeout) {
+		Assert.notNull(producer, "'producer' cannot be null");
+		Assert.notNull(closeTimeout, "'closeTimeout' cannot be null");
 		this.producer = producer;
+		this.closeTimeout = Duration.ofMillis(closeTimeout);
+	}
+
+	/**
+	 * Construct an instance for the producer.
+	 * @param producer the producer.
+	 * @param closeTimeout the close timeout.
+	 */
+	public KafkaResourceHolder(Producer<K, V> producer, Duration closeTimeout) {
+		Assert.notNull(producer, "'producer' cannot be null");
+		Assert.notNull(closeTimeout, "'closeTimeout' cannot be null");
+		this.producer = producer;
+		this.closeTimeout = closeTimeout;
 	}
 
 	public Producer<K, V> getProducer() {
@@ -50,7 +83,7 @@ public class KafkaResourceHolder<K, V> extends ResourceHolderSupport {
 	}
 
 	public void close() {
-		this.producer.close();
+		this.producer.close(this.closeTimeout.toMillis(), TimeUnit.MILLISECONDS);
 	}
 
 	public void rollback() {

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

@@ -16,8 +16,10 @@
 
 package org.springframework.kafka.core;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -77,6 +79,7 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 
 	private volatile ProducerListener<K, V> producerListener = new LoggingProducerListener<K, V>();
 
+	private Duration closeTimeout = ProducerFactoryUtils.DEFAULT_CLOSE_TIMEOUT;
 
 	/**
 	 * Create an instance using the supplied producer factory and autoFlush false.
@@ -158,6 +161,27 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 		return this.transactional;
 	}
 
+	/**
+	 * Set the maximum time to wait when closing a producer; default 5 seconds.
+	 * @param closeTimeout the close timeout.
+	 * @deprecated in favor of {@link #setCloseTimeout(Duration)}.
+	 * @since 1.3.11
+	 */
+	@Deprecated
+	public void setCloseTimeout(long closeTimeout) {
+		setCloseTimeout(Duration.ofMillis(closeTimeout));
+	}
+
+	/**
+	 * Set the maximum time to wait when closing a producer; default 5 seconds.
+	 * @param closeTimeout the close timeout.
+	 * @since 2.1.14
+	 */
+	public void setCloseTimeout(Duration closeTimeout) {
+		Assert.notNull(closeTimeout, "'closeTimeout' cannot be null");
+		this.closeTimeout = closeTimeout;
+	}
+
 	/**
 	 * Return the producer factory used by this template.
 	 * @return the factory.
@@ -354,9 +378,9 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 		producer.sendOffsetsToTransaction(offsets, consumerGroupId);
 	}
 
-	protected void closeProducer(Producer<K, V> producer, boolean inLocalTx) {
-		if (!inLocalTx) {
-			producer.close();
+	protected void closeProducer(Producer<K, V> producer, boolean inTx) {
+		if (!inTx) {
+			producer.close(this.closeTimeout.toMillis(), TimeUnit.MILLISECONDS);
 		}
 	}
 
@@ -441,7 +465,7 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 				return producer;
 			}
 			KafkaResourceHolder<K, V> holder = ProducerFactoryUtils
-					.getTransactionalResourceHolder(this.producerFactory);
+					.getTransactionalResourceHolder(this.producerFactory, this.closeTimeout);
 			return holder.getProducer();
 		}
 		else {

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

@@ -16,6 +16,9 @@
 
 package org.springframework.kafka.core;
 
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
 import org.apache.kafka.clients.producer.Producer;
 
 import org.springframework.lang.Nullable;
@@ -35,6 +38,11 @@ import org.springframework.util.Assert;
  */
 public final class ProducerFactoryUtils {
 
+	/**
+	 * The default close timeout (5 seconds).
+	 */
+	public static final Duration DEFAULT_CLOSE_TIMEOUT = Duration.ofSeconds(5);
+
 	private static ThreadLocal<String> groupIds = new ThreadLocal<>();
 
 	private ProducerFactoryUtils() {
@@ -51,6 +59,38 @@ public final class ProducerFactoryUtils {
 	public static <K, V> KafkaResourceHolder<K, V> getTransactionalResourceHolder(
 			final ProducerFactory<K, V> producerFactory) {
 
+		return getTransactionalResourceHolder(producerFactory, DEFAULT_CLOSE_TIMEOUT);
+	}
+
+	/**
+	 * Obtain a Producer that is synchronized with the current transaction, if any.
+	 * @param producerFactory the ProducerFactory to obtain a Channel for
+	 * @param closeTimeout the producer close timeout.
+	 * @param <K> the key type.
+	 * @param <V> the value type.
+	 * @return the resource holder.
+	 * @deprecated in favor of {@link #getTransactionalResourceHolder(ProducerFactory, Duration)}
+	 * @since 1.3.11
+	 */
+	@Deprecated
+	public static <K, V> KafkaResourceHolder<K, V> getTransactionalResourceHolder(
+			final ProducerFactory<K, V> producerFactory, long closeTimeout) {
+
+		return getTransactionalResourceHolder(producerFactory, Duration.ofMillis(closeTimeout));
+	}
+
+	/**
+	 * Obtain a Producer that is synchronized with the current transaction, if any.
+	 * @param producerFactory the ProducerFactory to obtain a Channel for
+	 * @param closeTimeout the producer close timeout.
+	 * @param <K> the key type.
+	 * @param <V> the value type.
+	 * @return the resource holder.
+	 * @since 2.1.14
+	 */
+	public static <K, V> KafkaResourceHolder<K, V> getTransactionalResourceHolder(
+			final ProducerFactory<K, V> producerFactory, Duration closeTimeout) {
+
 		Assert.notNull(producerFactory, "ProducerFactory must not be null");
 
 		@SuppressWarnings("unchecked")
@@ -63,11 +103,11 @@ public final class ProducerFactoryUtils {
 				producer.beginTransaction();
 			}
 			catch (RuntimeException e) {
-				producer.close();
+				producer.close(closeTimeout.toMillis(), TimeUnit.MILLISECONDS);
 				throw e;
 			}
 
-			resourceHolder = new KafkaResourceHolder<K, V>(producer);
+			resourceHolder = new KafkaResourceHolder<K, V>(producer, closeTimeout);
 			bindResourceToTransaction(resourceHolder, producerFactory);
 		}
 		return resourceHolder;
@@ -75,7 +115,7 @@ public final class ProducerFactoryUtils {
 
 	public static <K, V> void releaseResources(@Nullable KafkaResourceHolder<K, V> resourceHolder) {
 		if (resourceHolder != null) {
-			resourceHolder.getProducer().close();
+			resourceHolder.close();
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/transaction/KafkaTransactionManager.java

@@ -16,6 +16,8 @@
 
 package org.springframework.kafka.transaction;
 
+import java.time.Duration;
+
 import org.springframework.kafka.core.KafkaResourceHolder;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.core.ProducerFactoryUtils;
@@ -42,9 +44,9 @@ import org.springframework.util.Assert;
  *
  * <p>
  * Application code is required to retrieve the transactional Kafka resources via
- * {@link ProducerFactoryUtils#getTransactionalResourceHolder(ProducerFactory)}. Spring's
- * {@link org.springframework.kafka.core.KafkaTemplate KafkaTemplate} will auto detect a
- * thread-bound Producer and automatically participate in it.
+ * {@link ProducerFactoryUtils#getTransactionalResourceHolder(ProducerFactory, java.time.Duration)}.
+ * Spring's {@link org.springframework.kafka.core.KafkaTemplate KafkaTemplate} will auto
+ * detect a thread-bound Producer and automatically participate in it.
  *
  * <p>
  * <b>The use of {@link org.springframework.kafka.core.DefaultKafkaProducerFactory
@@ -70,6 +72,8 @@ public class KafkaTransactionManager<K, V> extends AbstractPlatformTransactionMa
 
 	private final ProducerFactory<K, V> producerFactory;
 
+	private Duration closeTimeout = ProducerFactoryUtils.DEFAULT_CLOSE_TIMEOUT;
+
 	/**
 	 * Create a new KafkaTransactionManager, given a ProducerFactory.
 	 * Transaction synchronization is turned off by default, as this manager might be used alongside a datastore-based
@@ -93,6 +97,27 @@ public class KafkaTransactionManager<K, V> extends AbstractPlatformTransactionMa
 		return this.producerFactory;
 	}
 
+	/**
+	 * Set the maximum time to wait when closing a producer; default 5 seconds.
+	 * @param closeTimeout the close timeout.
+	 * @deprecated in favor of {@link #setCloseTimeout(Duration)}.
+	 * @since 1.3.11
+	 */
+	@Deprecated
+	public void setCloseTimeout(long closeTimeout) {
+		setCloseTimeout(Duration.ofMillis(closeTimeout));
+	}
+
+	/**
+	 * Set the maximum time to wait when closing a producer; default 5 seconds.
+	 * @param closeTimeout the close timeout.
+	 * @since 2.1.14
+	 */
+	public void setCloseTimeout(Duration closeTimeout) {
+		Assert.notNull(closeTimeout, "'closeTimeout' cannot be null");
+		this.closeTimeout = closeTimeout;
+	}
+
 	/**
 	 * Return the producer factory.
 	 * @return the producer factory.
@@ -132,7 +157,8 @@ public class KafkaTransactionManager<K, V> extends AbstractPlatformTransactionMa
 		KafkaTransactionObject<K, V> txObject = (KafkaTransactionObject<K, V>) transaction;
 		KafkaResourceHolder<K, V> resourceHolder = null;
 		try {
-			resourceHolder = ProducerFactoryUtils.getTransactionalResourceHolder(getProducerFactory());
+			resourceHolder = ProducerFactoryUtils.getTransactionalResourceHolder(getProducerFactory(),
+					this.closeTimeout);
 			if (logger.isDebugEnabled()) {
 				logger.debug("Created Kafka transaction on producer [" + resourceHolder.getProducer() + "]");
 			}

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

@@ -25,6 +25,7 @@ import static org.mockito.Mockito.mock;
 import java.util.HashMap;
 import java.util.Queue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.kafka.clients.producer.Producer;
@@ -96,14 +97,14 @@ public class DefaultKafkaProducerFactoryTests {
 		inOrder.verify(producer).send(any(), any());
 		inOrder.verify(producer).commitTransaction();
 		inOrder.verify(producer).beginTransaction();
-		inOrder.verify(producer).close();
+		inOrder.verify(producer).close(ProducerFactoryUtils.DEFAULT_CLOSE_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
 		inOrder.verifyNoMoreInteractions();
 		pf.destroy();
 	}
 
 	@Test
 	@SuppressWarnings({ "rawtypes", "unchecked" })
-	public void testResetSingle() throws Exception {
+	public void testResetSingle() {
 		final Producer producer = mock(Producer.class);
 		DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(new HashMap<>()) {
 
@@ -115,7 +116,7 @@ public class DefaultKafkaProducerFactoryTests {
 		};
 		Producer aProducer = pf.createProducer();
 		assertThat(aProducer).isNotNull();
-		aProducer.close();
+		aProducer.close(ProducerFactoryUtils.DEFAULT_CLOSE_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
 		assertThat(KafkaTestUtils.getPropertyValue(pf, "producer")).isNotNull();
 		Queue cache = KafkaTestUtils.getPropertyValue(pf, "cache", Queue.class);
 		assertThat(cache.size()).isEqualTo(0);

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

@@ -17,10 +17,13 @@
 package org.springframework.kafka.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willThrow;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -30,10 +33,13 @@ import static org.mockito.Mockito.verify;
 import static org.springframework.kafka.test.assertj.KafkaConditions.key;
 import static org.springframework.kafka.test.assertj.KafkaConditions.value;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.kafka.clients.consumer.Consumer;
@@ -48,6 +54,7 @@ import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.assertj.core.api.Assertions;
@@ -316,7 +323,7 @@ public class KafkaTemplateTransactionTests {
 
 		verify(producer1).beginTransaction();
 		verify(producer1).commitTransaction();
-		verify(producer1).close();
+		verify(producer1).close(anyLong(), any());
 		verify(producer2, never()).beginTransaction();
 		verify(template, never()).executeInTransaction(any());
 	}
@@ -343,6 +350,69 @@ public class KafkaTemplateTransactionTests {
 		assertThat(producer.transactionAborted()).isFalse();
 		assertThat(producer.closed()).isTrue();
 		verify(producer, never()).abortTransaction();
+		verify(producer).close(ProducerFactoryUtils.DEFAULT_CLOSE_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+	}
+
+	@Test
+	public void testQuickCloseAfterCommitTimeout() {
+		@SuppressWarnings("unchecked")
+		Producer<String, String> producer = mock(Producer.class);
+
+		DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<String, String>(Collections.emptyMap()) {
+
+			@Override
+			public Producer<String, String> createProducer() {
+				CloseSafeProducer<String, String> closeSafeProducer = new CloseSafeProducer<>(producer, getCache());
+				return closeSafeProducer;
+			}
+
+		};
+		pf.setTransactionIdPrefix("foo");
+
+		KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);
+		template.setDefaultTopic(STRING_KEY_TOPIC);
+
+		willThrow(new TimeoutException()).given(producer).commitTransaction();
+		assertThatExceptionOfType(TimeoutException.class)
+			.isThrownBy(() ->
+				template.executeInTransaction(t -> {
+					return null;
+				}));
+		verify(producer, never()).abortTransaction();
+		verify(producer).close(Duration.ofMillis(0).toMillis(), TimeUnit.MILLISECONDS);
+	}
+
+	@Test
+	public void testNormalCloseAfterCommitCacheFull() {
+		@SuppressWarnings("unchecked")
+		Producer<String, String> producer = mock(Producer.class);
+
+		DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<String, String>(Collections.emptyMap()) {
+
+			@SuppressWarnings("unchecked")
+			@Override
+			public Producer<String, String> createProducer() {
+				BlockingQueue<CloseSafeProducer<String, String>> cache = new LinkedBlockingQueue<>(1);
+				try {
+					cache.put(new CloseSafeProducer<>(mock(Producer.class)));
+				}
+				catch (@SuppressWarnings("unused") InterruptedException e) {
+					Thread.currentThread().interrupt();
+				}
+				CloseSafeProducer<String, String> closeSafeProducer = new CloseSafeProducer<>(producer, cache);
+				return closeSafeProducer;
+			}
+
+		};
+		pf.setTransactionIdPrefix("foo");
+
+		KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);
+		template.setDefaultTopic(STRING_KEY_TOPIC);
+
+		template.executeInTransaction(t -> {
+			return null;
+		});
+		verify(producer).close(ProducerFactoryUtils.DEFAULT_CLOSE_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
 	}
 
 	@Test
@@ -435,9 +505,9 @@ public class KafkaTemplateTransactionTests {
 			inOrder.verify(producer1).beginTransaction();
 			inOrder.verify(producer2).beginTransaction();
 			inOrder.verify(producer2).commitTransaction();
-			inOrder.verify(producer2).close();
+			inOrder.verify(producer2).close(anyLong(), any());
 			inOrder.verify(producer1).commitTransaction();
-			inOrder.verify(producer1).close();
+			inOrder.verify(producer1).close(anyLong(), any());
 		}
 		finally {
 			TransactionSupport.clearTransactionIdSuffix();

spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java

@@ -161,7 +161,7 @@ public class TransactionalContainerTests {
 		willAnswer(i -> {
 			closeLatch.countDown();
 			return null;
-		}).given(producer).close();
+		}).given(producer).close(anyLong(), any());
 		ProducerFactory pf = mock(ProducerFactory.class);
 		given(pf.transactionCapable()).willReturn(true);
 		final List<String> transactionalIds = new ArrayList<>();
@@ -196,7 +196,7 @@ public class TransactionalContainerTests {
 		inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
 				new OffsetAndMetadata(0)), "group");
 		inOrder.verify(producer).commitTransaction();
-		inOrder.verify(producer).close();
+		inOrder.verify(producer).close(anyLong(), any());
 		inOrder.verify(producer).beginTransaction();
 		ArgumentCaptor<ProducerRecord> captor = ArgumentCaptor.forClass(ProducerRecord.class);
 		inOrder.verify(producer).send(captor.capture(), any(Callback.class));
@@ -204,7 +204,7 @@ public class TransactionalContainerTests {
 		inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
 				new OffsetAndMetadata(1)), "group");
 		inOrder.verify(producer).commitTransaction();
-		inOrder.verify(producer).close();
+		inOrder.verify(producer).close(anyLong(), any());
 		container.stop();
 		verify(pf, times(2)).createProducer();
 		verifyNoMoreInteractions(producer);
@@ -244,7 +244,7 @@ public class TransactionalContainerTests {
 		willAnswer(i -> {
 			closeLatch.countDown();
 			return null;
-		}).given(producer).close();
+		}).given(producer).close(anyLong(), any());
 		ProducerFactory pf = mock(ProducerFactory.class);
 		given(pf.transactionCapable()).willReturn(true);
 		given(pf.createProducer()).willReturn(producer);
@@ -271,7 +271,7 @@ public class TransactionalContainerTests {
 		inOrder.verify(producer, never()).sendOffsetsToTransaction(anyMap(), anyString());
 		inOrder.verify(producer, never()).commitTransaction();
 		inOrder.verify(producer).abortTransaction();
-		inOrder.verify(producer).close();
+		inOrder.verify(producer).close(anyLong(), any());
 		verify(consumer).seek(topicPartition0, 0);
 		verify(consumer).seek(topicPartition1, 0);
 		verify(consumer, never()).commitSync(anyMap());
@@ -311,7 +311,7 @@ public class TransactionalContainerTests {
 		willAnswer(i -> {
 			closeLatch.countDown();
 			return null;
-		}).given(producer).close();
+		}).given(producer).close(anyLong(), any());
 		ProducerFactory pf = mock(ProducerFactory.class);
 		given(pf.transactionCapable()).willReturn(true);
 		given(pf.createProducer()).willReturn(producer);
@@ -338,7 +338,7 @@ public class TransactionalContainerTests {
 		inOrder.verify(producer, never()).sendOffsetsToTransaction(anyMap(), anyString());
 		inOrder.verify(producer, never()).commitTransaction();
 		inOrder.verify(producer).abortTransaction();
-		inOrder.verify(producer).close();
+		inOrder.verify(producer).close(anyLong(), any());
 		verify(consumer).seek(topicPartition0, 0);
 		verify(consumer).seek(topicPartition1, 0);
 		verify(consumer, never()).commitSync(anyMap());
@@ -377,7 +377,7 @@ public class TransactionalContainerTests {
 		willAnswer(i -> {
 			closeLatch.countDown();
 			return null;
-		}).given(producer).close();
+		}).given(producer).close(anyLong(), any());
 
 		final ProducerFactory pf = mock(ProducerFactory.class);
 		given(pf.transactionCapable()).willReturn(true);
@@ -405,7 +405,7 @@ public class TransactionalContainerTests {
 		inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
 				new OffsetAndMetadata(1)), "group");
 		inOrder.verify(producer).commitTransaction();
-		inOrder.verify(producer).close();
+		inOrder.verify(producer).close(anyLong(), any());
 		container.stop();
 		verify(pf).createProducer();
 	}