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.

Deprecate 1.3.x public APIs

Author: garyrussell

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.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.factory.DisposableBean;
 import org.springframework.context.Lifecycle;
@@ -126,8 +131,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
 	 */
@@ -195,7 +202,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) {
@@ -352,6 +359,25 @@ public void closeProducerFor(String transactionIdSuffix) {
 	 */
 	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;
@@ -360,7 +386,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);
@@ -428,7 +454,7 @@ public void beginTransaction() throws ProducerFencedException {
 				if (logger.isErrorEnabled()) {
 					logger.error("beginTransaction failed: " + this, e);
 				}
-				this.txFailed = true;
+				this.txFailed = e;
 				throw e;
 			}
 		}
@@ -452,7 +478,7 @@ public void commitTransaction() throws ProducerFencedException {
 				if (logger.isErrorEnabled()) {
 					logger.error("commitTransaction failed: " + this, e);
 				}
-				this.txFailed = true;
+				this.txFailed = e;
 				throw e;
 			}
 		}
@@ -469,7 +495,7 @@ public void abortTransaction() throws ProducerFencedException {
 				if (logger.isErrorEnabled()) {
 					logger.error("Abort failed: " + this, e);
 				}
-				this.txFailed = true;
+				this.txFailed = e;
 				throw e;
 			}
 		}
@@ -482,17 +508,16 @@ public void close() {
 		@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);
 					}
@@ -502,19 +527,28 @@ public void close(long timeout, @Nullable 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);
-								}
+								closeDelegate(closeTimeout);
 							}
 						}
 					}
 				}
 			}
 		}
 
+		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 void commit() {
 	}
 
 	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.
@@ -157,6 +160,27 @@ public boolean isTransactional() {
 		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;
+	}
+
 	@Override
 	public ListenableFuture<SendResult<K, V>> sendDefault(V data) {
 		return send(this.defaultTopic, data);
@@ -342,9 +366,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.toMillis(), TimeUnit.MILLISECONDS);
 		}
 	}
 
@@ -422,7 +446,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,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.transaction.support.ResourceHolderSynchronization;
@@ -34,6 +37,11 @@
  */
 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() {
@@ -50,6 +58,38 @@ 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.
+	 * @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")
@@ -62,19 +102,19 @@ public static <K, V> KafkaResourceHolder<K, V> getTransactionalResourceHolder(
 				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;
 	}
 
 	public static <K, V> void releaseResources(KafkaResourceHolder<K, V> resourceHolder) {
 		if (resourceHolder != null) {
-			resourceHolder.getProducer().close();
+			resourceHolder.close();
 		}
 	}