GH-2852: API for customizing transactionIdSuffix

Fixes: #2852 

* Provide a way to customize the transactionIdSuffix by providing a new strategy API - `TransactionIdSuffixStrategy`
* PR adds a default implementation for this new API
* Related changes in classes that deal with transactions

Author: stillya

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/transactions.adoc

@@ -102,11 +102,43 @@ NOTE: If there is a `KafkaTransactionManager` (or synchronized) transaction in p
 Instead, a new "nested" transaction is used.
 
 [[transaction-id-prefix]]
-== `transactionIdPrefix`
+== `TransactionIdPrefix`
 
 With `EOSMode.V2` (aka `BETA`), the only supported mode, it is no longer necessary to use the same `transactional.id`, even for consumer-initiated transactions; in fact, it must be unique on each instance the same as for producer-initiated transactions.
 This property must have a different value on each application instance.
 
+[[transaction-id-suffix-fixed]]
+== `TransactionIdSuffix Fixed`
+
+Since 3.2, a new `TransactionIdSuffixStrategy` interface was introduced to manage `transactional.id` suffix.
+The default implementation is `DefaultTransactionIdSuffixStrategy` when setting `maxCache` greater than zero can reuse `transactional.id` within a specific range, otherwise suffixes will be generated on the fly by incrementing a counter.
+When a transaction producer is requested and `transactional.id` all in use, throw a `NoProducerAvailableException`.
+User can then use a `RetryTemplate` configured to retry that exception, with a suitably configured back off.
+
+[source,java]
+----
+public static class Config {
+
+    @Bean
+    public ProducerFactory<String, String> myProducerFactory() {
+        Map<String, Object> configs = producerConfigs();
+        configs.put(ProducerConfig.CLIENT_ID_CONFIG, "myClientId");
+        ...
+        DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(configs);
+        ...
+        TransactionIdSuffixStrategy ss = new DefaultTransactionIdSuffixStrategy(5);
+        pf.setTransactionIdSuffixStrategy(ss);
+        return pf;
+    }
+
+}
+----
+When setting `maxCache` to 5, `transactional.id` is `my.txid.`++`{0-4}`+.
+
+IMPORTANT: When using `KafkaTransactionManager` with the `ConcurrentMessageListenerContainer` and enabling `maxCache`, it is necessary to set `maxCache` to a value greater than or equal to `concurrency`.
+If a `MessageListenerContainer` is unable to acquire a `transactional.id` suffix, it will throw a `NoProducerAvailableException`.
+When using nested transactions in the `ConcurrentMessageListenerContainer`, it is necessary to adjust the maxCache setting to handle the increased number of nested transactions.
+
 [[tx-template-mixed]]
 == `KafkaTemplate` Transactional and non-Transactional Publishing
 

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

@@ -7,3 +7,9 @@
 This section covers the changes made from version 3.1 to version 3.2.
 For changes in earlier version, see xref:appendix/change-history.adoc[Change History].
 
+[[x32-tiss]]
+=== TransactionIdSuffixStrategy
+
+A new `TransactionIdSuffixStrategy` interface was introduced to manage `transactional.id` suffix.
+The default implementation is `DefaultTransactionIdSuffixStrategy` when setting `maxCache` greater than zero can reuse `transactional.id` within a specific range, otherwise suffixes will be generated on the fly by incrementing a counter.
+See xref:kafka/transactions.adoc#transaction-id-suffix-fixed[Fixed TransactionIdSuffix] for more information.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2023 the original author or authors.
+ * Copyright 2016-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -121,8 +121,6 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
 
 	private final Map<String, Object> configs;
 
-	private final AtomicInteger transactionIdSuffix = new AtomicInteger();
-
 	private final Map<String, BlockingQueue<CloseSafeProducer<K, V>>> cache = new ConcurrentHashMap<>();
 
 	private final Map<Thread, CloseSafeProducer<K, V>> threadBoundProducers = new ConcurrentHashMap<>();
@@ -137,6 +135,8 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
 
 	private final AtomicBoolean running = new AtomicBoolean();
 
+	private TransactionIdSuffixStrategy transactionIdSuffixStrategy = new DefaultTransactionIdSuffixStrategy(0);
+
 	private Supplier<Serializer<K>> keySerializerSupplier;
 
 	private Supplier<Serializer<V>> valueSerializerSupplier;
@@ -350,6 +350,16 @@ public void setValueSerializerSupplier(Supplier<Serializer<V>> valueSerializerSu
 		this.valueSerializerSupplier = valueSerializerSupplier(valueSerializerSupplier);
 	}
 
+	/**
+	 * Set the transaction suffix strategy.
+	 * @param transactionIdSuffixStrategy the strategy.
+	 * @since 3.2
+	 */
+	public void setTransactionIdSuffixStrategy(TransactionIdSuffixStrategy transactionIdSuffixStrategy) {
+		Assert.notNull(transactionIdSuffixStrategy, "'transactionIdSuffixStrategy' cannot be null");
+		this.transactionIdSuffixStrategy = transactionIdSuffixStrategy;
+	}
+
 	/**
 	 * If true (default), programmatically provided serializers (via constructor or
 	 * setters) will be configured using the producer configuration. Set to false if the
@@ -404,7 +414,7 @@ public Duration getPhysicalCloseTimeout() {
 
 	/**
 	 * Set a prefix for the {@link ProducerConfig#TRANSACTIONAL_ID_CONFIG} config. By
-	 * default a {@link ProducerConfig#TRANSACTIONAL_ID_CONFIG} value from configs is used
+	 * default, a {@link ProducerConfig#TRANSACTIONAL_ID_CONFIG} value from configs is used
 	 * as a prefix in the target producer configs.
 	 * @param transactionIdPrefix the prefix.
 	 * @since 1.3
@@ -536,9 +546,9 @@ public ProducerFactory<K, V> copyWithConfigurationOverride(Map<String, Object> o
 		producerProperties.putAll(overrideProperties);
 		producerProperties = ensureExistingTransactionIdPrefixInProperties(producerProperties);
 		DefaultKafkaProducerFactory<K, V> newFactory = new DefaultKafkaProducerFactory<>(producerProperties,
-						getKeySerializerSupplier(),
-						getValueSerializerSupplier(),
-						isConfigureSerializers());
+				getKeySerializerSupplier(),
+				getValueSerializerSupplier(),
+				isConfigureSerializers());
 		newFactory.setPhysicalCloseTimeout((int) getPhysicalCloseTimeout().getSeconds());
 		newFactory.setProducerPerThread(isProducerPerThread());
 		for (ProducerPostProcessor<K, V> templatePostProcessor : getPostProcessors()) {
@@ -690,7 +700,7 @@ public void destroy() {
 		}
 		if (producerToClose != null) {
 			try {
-				producerToClose.closeDelegate(this.physicalCloseTimeout, this.listeners);
+				producerToClose.closeDelegate(this.physicalCloseTimeout);
 			}
 			catch (Exception e) {
 				LOGGER.error(e, "Exception while closing producer");
@@ -700,7 +710,7 @@ public void destroy() {
 			CloseSafeProducer<K, V> next = queue.poll();
 			while (next != null) {
 				try {
-					next.closeDelegate(this.physicalCloseTimeout, this.listeners);
+					next.closeDelegate(this.physicalCloseTimeout);
 				}
 				catch (Exception e) {
 					LOGGER.error(e, "Exception while closing producer");
@@ -711,7 +721,7 @@ public void destroy() {
 		this.cache.clear();
 		this.threadBoundProducers.values().forEach(prod -> {
 			try {
-				prod.closeDelegate(this.physicalCloseTimeout, this.listeners);
+				prod.closeDelegate(this.physicalCloseTimeout);
 			}
 			catch (Exception e) {
 				LOGGER.error(e, "Exception while closing producer");
@@ -769,7 +779,7 @@ private Producer<K, V> doCreateProducer(@Nullable String txIdPrefix) {
 		this.globalLock.lock();
 		try {
 			if (this.producer != null && this.producer.closed) {
-				this.producer.closeDelegate(this.physicalCloseTimeout, this.listeners);
+				this.producer.closeDelegate(this.physicalCloseTimeout);
 				this.producer = null;
 			}
 			if (this.producer != null && expire(this.producer)) {
@@ -820,6 +830,9 @@ protected Producer<K, V> createKafkaProducer() {
 	 * @since 2.2.13
 	 */
 	protected final boolean removeProducer(CloseSafeProducer<K, V> producerToRemove, Duration timeout) {
+		if (producerToRemove.closed) {
+			this.listeners.forEach(listener -> listener.producerRemoved(producerToRemove.clientId, producerToRemove));
+		}
 		return producerToRemove.closed;
 	}
 
@@ -846,7 +859,8 @@ protected Producer<K, V> createTransactionalProducer(String txIdPrefix) {
 			}
 		}
 		if (cachedProducer == null) {
-			return doCreateTxProducer(txIdPrefix, "" + this.transactionIdSuffix.getAndIncrement(), this::cacheReturner);
+			String suffix = this.transactionIdSuffixStrategy.acquireSuffix(txIdPrefix);
+			return doCreateTxProducer(txIdPrefix, suffix, this::cacheReturner);
 		}
 		else {
 			return cachedProducer;
@@ -856,24 +870,28 @@ 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);
+			producer.closeDelegate(this.physicalCloseTimeout);
 		}
 		return expired;
 	}
 
 	boolean cacheReturner(CloseSafeProducer<K, V> producerToRemove, Duration timeout) {
 		if (producerToRemove.closed) {
-			producerToRemove.closeDelegate(timeout, this.listeners);
+			this.removeTransactionProducer(producerToRemove, timeout, this.listeners);
 			return true;
 		}
 		else {
 			this.globalLock.lock();
 			try {
+				if (producerToRemove.epoch != this.epoch.get()) {
+					this.removeTransactionProducer(producerToRemove, timeout, this.listeners);
+					return true;
+				}
 				BlockingQueue<CloseSafeProducer<K, V>> txIdCache = getCache(producerToRemove.txIdPrefix);
 				if (producerToRemove.epoch != this.epoch.get()
 						|| (txIdCache != null && !txIdCache.contains(producerToRemove)
 						&& !txIdCache.offer(producerToRemove))) {
-					producerToRemove.closeDelegate(timeout, this.listeners);
+					this.removeTransactionProducer(producerToRemove, timeout, this.listeners);
 					return true;
 				}
 			}
@@ -884,6 +902,12 @@ boolean cacheReturner(CloseSafeProducer<K, V> producerToRemove, Duration timeout
 		}
 	}
 
+	private void removeTransactionProducer(CloseSafeProducer<K, V> producer, Duration timeout,
+			List<Listener<K, V>> listeners) {
+		this.transactionIdSuffixStrategy.releaseSuffix(producer.txIdPrefix, producer.txIdSuffix);
+		listeners.forEach(listener -> listener.producerRemoved(producer.clientId, producer));
+	}
+
 	private CloseSafeProducer<K, V> doCreateTxProducer(String prefix, String suffix,
 			BiPredicate<CloseSafeProducer<K, V>, Duration> remover) {
 		Producer<K, V> newProducer = createRawProducer(getTxProducerConfigs(prefix + suffix));
@@ -899,10 +923,13 @@ private CloseSafeProducer<K, V> doCreateTxProducer(String prefix, String suffix,
 				newEx.addSuppressed(ex2);
 				throw newEx; // NOSONAR - lost stack trace
 			}
+			finally {
+				this.transactionIdSuffixStrategy.releaseSuffix(prefix, suffix);
+			}
 			throw new KafkaException("initTransactions() failed", ex);
 		}
 		CloseSafeProducer<K, V> closeSafeProducer =
-				new CloseSafeProducer<>(newProducer, remover, prefix, this.physicalCloseTimeout, this.beanName,
+				new CloseSafeProducer<>(newProducer, remover, prefix, suffix, this.physicalCloseTimeout, this.beanName,
 						this.epoch.get());
 		this.listeners.forEach(listener -> listener.producerAdded(closeSafeProducer.clientId, closeSafeProducer));
 		return closeSafeProducer;
@@ -923,7 +950,7 @@ protected BlockingQueue<CloseSafeProducer<K, V>> getCache() {
 	}
 
 	@Nullable
-	protected BlockingQueue<CloseSafeProducer<K, V>> getCache(String txIdPrefix) {
+	protected BlockingQueue<CloseSafeProducer<K, V>> getCache(@Nullable String txIdPrefix) {
 		if (txIdPrefix == null) {
 			return null;
 		}
@@ -941,7 +968,7 @@ protected BlockingQueue<CloseSafeProducer<K, V>> getCache(String txIdPrefix) {
 	public void closeThreadBoundProducer() {
 		CloseSafeProducer<K, V> tlProducer = this.threadBoundProducers.remove(Thread.currentThread());
 		if (tlProducer != null) {
-			tlProducer.closeDelegate(this.physicalCloseTimeout, this.listeners);
+			tlProducer.closeDelegate(this.physicalCloseTimeout);
 		}
 	}
 
@@ -991,6 +1018,8 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 
 		final String txIdPrefix; // NOSONAR
 
+		final String txIdSuffix;  // NOSONAR
+
 		final long created; // NOSONAR
 
 		private final Duration closeTimeout;
@@ -1010,14 +1039,21 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 			this(delegate, removeConsumerProducer, null, closeTimeout, factoryName, epoch);
 		}
 
+		CloseSafeProducer(Producer<K, V> delegate, BiPredicate<CloseSafeProducer<K, V>, Duration> removeProducer,
+				@Nullable String txIdPrefix, Duration closeTimeout, String factoryName, int epoch) {
+
+			this(delegate, removeProducer, txIdPrefix, null, closeTimeout, factoryName, epoch);
+		}
+
 		CloseSafeProducer(Producer<K, V> delegate,
 				BiPredicate<CloseSafeProducer<K, V>, Duration> removeProducer, @Nullable String txIdPrefix,
-				Duration closeTimeout, String factoryName, int epoch) {
+				@Nullable String txIdSuffix, Duration closeTimeout, String factoryName, int epoch) {
 
 			Assert.isTrue(!(delegate instanceof CloseSafeProducer), "Cannot double-wrap a producer");
 			this.delegate = delegate;
 			this.removeProducer = removeProducer;
 			this.txIdPrefix = txIdPrefix;
+			this.txIdSuffix = txIdSuffix;
 			this.closeTimeout = closeTimeout;
 			Map<MetricName, ? extends Metric> metrics = delegate.metrics();
 			Iterator<MetricName> metricIterator = metrics.keySet().iterator();
@@ -1057,7 +1093,6 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
 					}
 					callback.onCompletion(metadata, exception);
 				}
-
 			});
 		}
 
@@ -1159,22 +1194,30 @@ public void close(@Nullable Duration timeout) {
 					this.removeProducer.test(this, this.producerFailed instanceof TimeoutException
 							? CLOSE_TIMEOUT_AFTER_TX_TIMEOUT
 							: timeout);
+					this.delegate.close(timeout == null ? this.closeTimeout : this.producerFailed instanceof TimeoutException
+							? CLOSE_TIMEOUT_AFTER_TX_TIMEOUT
+							: timeout);
 				}
 				else {
 					this.closed = this.removeProducer.test(this, timeout);
+					if (this.closed) {
+						this.delegate.close(timeout == null ? this.closeTimeout : timeout);
+					}
 				}
 			}
 		}
 
-		void closeDelegate(Duration timeout, List<Listener<K, V>> listeners) {
+		void closeDelegate(Duration timeout) {
 			try {
-				this.delegate.close(timeout == null ? this.closeTimeout : timeout);
+				if (!this.closed) {
+					this.delegate.close(timeout == null ? this.closeTimeout : timeout);
+					this.closed = true;
+					this.removeProducer.test(this, timeout == null ? this.closeTimeout : timeout);
+				}
 			}
 			catch (Exception ex) {
 				LOGGER.warn(ex, () -> "Failed to close " + this.delegate);
 			}
-			listeners.forEach(listener -> listener.producerRemoved(this.clientId, this));
-			this.closed = true;
 		}
 
 		@Override