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

Resolves #1196

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

Author: garyrussell

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

@@ -42,6 +42,7 @@
 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.springframework.beans.factory.DisposableBean;
@@ -123,8 +124,10 @@ public void setValueSerializer(Serializer<V> valueSerializer) {
 	}
 
 	/**
-	 * 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
 	 */
@@ -179,7 +182,7 @@ public boolean transactionCapable() {
 
 	@SuppressWarnings("resource")
 	@Override
-	public void destroy() throws Exception { //NOSONAR
+	public void destroy() {
 		CloseSafeProducer<K, V> producer = this.producer;
 		this.producer = null;
 		if (producer != null) {
@@ -334,7 +337,7 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 
 		private final String txId;
 
-		private volatile boolean txFailed;
+		private volatile Exception txFailed;
 
 		CloseSafeProducer(Producer<K, V> delegate) {
 			this(delegate, null, null);
@@ -402,7 +405,7 @@ public void beginTransaction() throws ProducerFencedException {
 				if (logger.isErrorEnabled()) {
 					logger.error("beginTransaction failed: " + this, e);
 				}
-				this.txFailed = true;
+				this.txFailed = e;
 				throw e;
 			}
 		}
@@ -426,7 +429,7 @@ public void commitTransaction() throws ProducerFencedException {
 				if (logger.isErrorEnabled()) {
 					logger.error("commitTransaction failed: " + this, e);
 				}
-				this.txFailed = true;
+				this.txFailed = e;
 				throw e;
 			}
 		}
@@ -443,7 +446,7 @@ public void abortTransaction() throws ProducerFencedException {
 				if (logger.isErrorEnabled()) {
 					logger.error("Abort failed: " + this, e);
 				}
-				this.txFailed = true;
+				this.txFailed = e;
 				throw e;
 			}
 		}
@@ -456,17 +459,16 @@ public void close() {
 		@Override
 		public void close(long timeout, TimeUnit unit) {
 			if (this.cache != null) {
-				if (this.txFailed) {
+				long closeTimeout = this.txFailed instanceof TimeoutException || unit == null
+						? 0L
+						: 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);
 					}
+					this.delegate.close(closeTimeout, unit);
 					if (this.consumerProducers != null) {
 						removeConsumerProducer();
 					}
@@ -476,12 +478,7 @@ public void close(long timeout, TimeUnit unit) {
 						synchronized (this) {
 							if (!this.cache.contains(this)
 									&& !this.cache.offer(this)) {
-								if (unit == null) {
-									this.delegate.close();
-								}
-								else {
-									this.delegate.close(timeout, unit);
-								}
+								this.delegate.close(closeTimeout, unit);
 							}
 						}
 					}

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

@@ -16,9 +16,12 @@
 
 package org.springframework.kafka.core;
 
+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 +36,27 @@ public class KafkaResourceHolder<K, V> extends ResourceHolderSupport {
 
 	private final Producer<K, V> producer;
 
+	private final long 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.
+	 * @since 1.3.11
+	 */
+	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 = closeTimeout;
 	}
 
 	public Producer<K, V> getProducer() {
@@ -50,7 +68,7 @@ public void commit() {
 	}
 
 	public void close() {
-		this.producer.close();
+		this.producer.close(this.closeTimeout, TimeUnit.MILLISECONDS);
 	}
 
 	public void rollback() {

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

@@ -18,6 +18,7 @@
 
 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 +78,7 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 
 	private volatile ProducerListener<K, V> producerListener = new LoggingProducerListener<K, V>();
 
+	private final long closeTimeout = ProducerFactoryUtils.DEFAULT_CLOSE_TIMEOUT;
 
 	/**
 	 * Create an instance using the supplied producer factory and autoFlush false.
@@ -149,6 +151,15 @@ public void setMessageConverter(RecordMessageConverter messageConverter) {
 		this.messageConverter = messageConverter;
 	}
 
+	/**
+	 * Set the maximum time to wait when closing a producer; default 5 seconds.
+	 * @param closeTimeout the close timeout.
+	 * @since 1.3.11
+	 */
+	public void setCloseTimeout(long closeTimeout) {
+		setCloseTimeout(closeTimeout);
+	}
+
 	@Override
 	public ListenableFuture<SendResult<K, V>> sendDefault(V data) {
 		return send(this.defaultTopic, data);
@@ -329,9 +340,9 @@ public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offs
 		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, TimeUnit.MILLISECONDS);
 		}
 	}
 
@@ -409,7 +420,7 @@ private Producer<K, V> getTheProducer() {
 				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,8 @@
 
 package org.springframework.kafka.core;
 
+import java.util.concurrent.TimeUnit;
+
 import org.apache.kafka.clients.producer.Producer;
 
 import org.springframework.transaction.support.ResourceHolderSynchronization;
@@ -34,6 +36,11 @@
  */
 public final class ProducerFactoryUtils {
 
+	/**
+	 * The default close timeout (5 seconds) - package private to avoid later deprecation.
+	 */
+	static final long DEFAULT_CLOSE_TIMEOUT = 5000L;
+
 	private static ThreadLocal<String> groupIds = new ThreadLocal<>();
 
 	private ProducerFactoryUtils() {
@@ -50,6 +57,21 @@ private 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.
+	 * @since 1.3.11
+	 */
+	public static <K, V> KafkaResourceHolder<K, V> getTransactionalResourceHolder(
+			final ProducerFactory<K, V> producerFactory, long closeTimeout) {
+
 		Assert.notNull(producerFactory, "ProducerFactory must not be null");
 
 		@SuppressWarnings("unchecked")
@@ -62,19 +84,19 @@ public static <K, V> KafkaResourceHolder<K, V> getTransactionalResourceHolder(
 				producer.beginTransaction();
 			}
 			catch (RuntimeException e) {
-				producer.close();
+				producer.close(closeTimeout, TimeUnit.MILLISECONDS);
 				throw e;
 			}
 
-			resourceHolder = new KafkaResourceHolder<K, V>(producer);
+			resourceHolder = new KafkaResourceHolder<K, V>(producer, closeTimeout);
 			bindResourceToTransaction(resourceHolder, producerFactory);
 		}
 		return resourceHolder;
 	}
 
 	public static <K, V> void releaseResources(KafkaResourceHolder<K, V> resourceHolder) {
 		if (resourceHolder != null) {
-			resourceHolder.getProducer().close();
+			resourceHolder.close();
 		}
 	}
 

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

@@ -18,7 +18,6 @@
 
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaResourceHolder;
-import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.core.ProducerFactoryUtils;
 import org.springframework.transaction.CannotCreateTransactionException;
@@ -45,9 +44,9 @@
  *
  * <p>
  * Application code is required to retrieve the transactional Kafka resources via
- * {@link ProducerFactoryUtils#getTransactionalResourceHolder(ProducerFactory)}.
- * Spring's {@link 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 DefaultKafkaProducerFactory} as a target for this transaction
@@ -68,8 +67,12 @@
 public class KafkaTransactionManager<K, V> extends AbstractPlatformTransactionManager
 		implements ResourceTransactionManager {
 
+	private static final long DEFAULT_CLOSE_TIMEOUT = 5000L;
+
 	private final ProducerFactory<K, V> producerFactory;
 
+	private final long closeTimeout = DEFAULT_CLOSE_TIMEOUT;
+
 	/**
 	 * Create a new KafkaTransactionManager, given a ConnectionFactory.
 	 * Transaction synchronization is turned off by default, as this manager might be used alongside a datastore-based
@@ -92,6 +95,15 @@ public ProducerFactory<K, V> getProducerFactory() {
 		return this.producerFactory;
 	}
 
+	/**
+	 * Set the maximum time to wait when closing a producer; default 5 seconds.
+	 * @param closeTimeout the close timeout.
+	 * @since 1.3.11
+	 */
+	public void setCloseTimeout(long closeTimeout) {
+		setCloseTimeout(closeTimeout);
+	}
+
 	@Override
 	public Object getResourceFactory() {
 		return getProducerFactory();
@@ -122,7 +134,8 @@ protected void doBegin(Object transaction, TransactionDefinition definition) {
 		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

@@ -24,6 +24,7 @@
 
 import java.util.HashMap;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.kafka.clients.producer.Producer;
@@ -93,7 +94,7 @@ protected Producer createTransactionalProducer() {
 		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, TimeUnit.MILLISECONDS);
 		inOrder.verifyNoMoreInteractions();
 		pf.destroy();
 	}