GH-1983: Support copy of producer factory

Fixes #1983

Preserves post processors for of custom factories

Instrumentations attached to ProducerFactory instances are destroyed if the user adds configOverrides.
This behavior is unexpected for the developer, for example it will undo sleuth instrumentation
that is otherwise kept if the new KafkaTemplate(Map) constructor would be used.

* adds copy for all visible factory properties and tests
* moves the copy feature to ProducerFactory
* fixing review remarks
* removes a left over TODO
* implements a generic copy in the KafkaTemplate to avoid breaking changes
* wraps javadoc at col 90
* Clean up for code style

**Cherry-pick to `2.5.x`, `2.6.x` & `2.7.x`**

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
#	spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java
#	spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java

Author: thst71

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

@@ -108,10 +108,11 @@
  * @author Nakul Mishra
  * @author Artem Bilan
  * @author Chris Gilbert
+ * @author Thomas Strauß
  */
 public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
 		implements ProducerFactory<K, V>, ApplicationContextAware,
-			BeanNameAware, ApplicationListener<ContextStoppedEvent>, DisposableBean {
+		BeanNameAware, ApplicationListener<ContextStoppedEvent>, DisposableBean {
 
 	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(DefaultKafkaProducerFactory.class));
 
@@ -361,6 +362,63 @@ public void setMaxAge(Duration maxAge) {
 		this.maxAge = maxAge.toMillis();
 	}
 
+	/**
+	 * Copy properties of the instance and the given properties to create a new producer factory.
+	 * <p>If the {@link org.springframework.kafka.core.DefaultKafkaProducerFactory} makes a
+	 * copy of itself, the transaction id prefix is recovered from the properties. If
+	 * you want to change the ID config, add a new
+	 * {@link org.apache.kafka.clients.producer.ProducerConfig#TRANSACTIONAL_ID_CONFIG}
+	 * key to the override config.</p>
+	 * @param overrideProperties the properties to be applied to the new factory
+	 * @return {@link org.springframework.kafka.core.DefaultKafkaProducerFactory} with
+	 *  properties applied
+	 */
+	@Override
+	public ProducerFactory<K, V> copyWithConfigurationOverride(Map<String, Object> overrideProperties) {
+		Map<String, Object> producerProperties = new HashMap<>(getConfigurationProperties());
+		producerProperties.putAll(overrideProperties);
+		producerProperties = ensureExistingTransactionIdPrefixInProperties(producerProperties);
+		DefaultKafkaProducerFactory<K, V> newFactory =
+				new DefaultKafkaProducerFactory<>(producerProperties,
+						getKeySerializerSupplier(),
+						getValueSerializerSupplier());
+		newFactory.setPhysicalCloseTimeout((int) getPhysicalCloseTimeout().getSeconds());
+		newFactory.setProducerPerConsumerPartition(isProducerPerConsumerPartition());
+		newFactory.setProducerPerThread(isProducerPerThread());
+		for (ProducerPostProcessor<K, V> templatePostProcessor : getPostProcessors()) {
+			newFactory.addPostProcessor(templatePostProcessor);
+		}
+		for (ProducerFactory.Listener<K, V> templateListener : getListeners()) {
+			newFactory.addListener(templateListener);
+		}
+		return newFactory;
+	}
+
+
+	/**
+	 * Ensures that the returned properties map contains a transaction id prefix.
+	 * The {@link org.springframework.kafka.core.DefaultKafkaProducerFactory}
+	 * modifies the local properties copy, the txn key is removed and
+	 * stored locally in a property. To make a proper copy of the properties in a
+	 * new factory, the transactionId has to be reinserted prior use.
+	 * The incoming properties are checked for a transactionId key. If none is
+	 * there, the one existing in the factory is added.
+	 * @param producerProperties the properties to be used for the new factory
+	 * @return the producerProperties or a copy with the transaction ID set
+	 */
+	private Map<String, Object> ensureExistingTransactionIdPrefixInProperties(Map<String, Object> producerProperties) {
+		String transactionIdPrefix = getTransactionIdPrefix();
+		if (StringUtils.hasText(transactionIdPrefix)) {
+			if (!producerProperties.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG)) {
+				Map<String, Object> producerPropertiesWithTxnId = new HashMap<>(producerProperties);
+				producerPropertiesWithTxnId.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionIdPrefix);
+				return producerPropertiesWithTxnId;
+			}
+		}
+
+		return producerProperties;
+	}
+
 	/**
 	 * Add a listener.
 	 * @param listener the listener.
@@ -418,8 +476,8 @@ public void updateConfigs(Map<String, Object> updates) {
 				Assert.isTrue(entry.getValue() instanceof String, () -> "'" + ProducerConfig.TRANSACTIONAL_ID_CONFIG
 						+ "' must be a String, not a " + entry.getClass().getName());
 				Assert.isTrue(this.transactionIdPrefix != null
-							? entry.getValue() != null
-							: entry.getValue() == null,
+								? entry.getValue() != null
+								: entry.getValue() == null,
 						"Cannot change transactional capability");
 				this.transactionIdPrefix = (String) entry.getValue();
 			}
@@ -695,7 +753,7 @@ boolean cacheReturner(CloseSafeProducer<K, V> producerToRemove, Duration timeout
 				BlockingQueue<CloseSafeProducer<K, V>> txIdCache = getCache(producerToRemove.txIdPrefix);
 				if (producerToRemove.epoch != this.epoch.get()
 						|| (txIdCache != null && !txIdCache.contains(producerToRemove)
-								&& !txIdCache.offer(producerToRemove))) {
+						&& !txIdCache.offer(producerToRemove))) {
 					producerToRemove.closeDelegate(timeout, this.listeners);
 					return true;
 				}
@@ -943,7 +1001,7 @@ public void abortTransaction() throws ProducerFencedException {
 			LOGGER.debug(() -> toString() + " abortTransaction()");
 			if (this.producerFailed != null) {
 				LOGGER.debug(() -> "abortTransaction ignored - previous txFailed: " + this.producerFailed.getMessage()
-					+ ": " + this);
+						+ ": " + this);
 			}
 			else {
 				try {

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

@@ -75,7 +75,8 @@
  * @author Igor Stepanov
  * @author Artem Bilan
  * @author Biju Kunjummen
- * @author Endika Guti?rrez
+ * @author Endika Gutierrez
+ * @author Thomas Strauß
  */
 public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationContextAware, BeanNameAware,
 		ApplicationListener<ContextStoppedEvent>, DisposableBean {
@@ -158,8 +159,16 @@ public KafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush) {
 	 * to occur immediately, regardless of that setting, or if you wish to block until the
 	 * broker has acknowledged receipt according to the producer's {@code acks} property.
 	 * If the configOverrides is not null or empty, a new
-	 * {@link DefaultKafkaProducerFactory} will be created with merged producer properties
-	 * with the overrides being applied after the supplied factory's properties.
+	 * {@link ProducerFactory} will be created using
+	 * {@link org.springframework.kafka.core.ProducerFactory#copyWithConfigurationOverride(java.util.Map)}
+	 * The factory shall apply the overrides after the supplied factory's properties.
+	 * The {@link org.springframework.kafka.core.ProducerPostProcessor}s from the
+	 * original factory are copied over to keep instrumentation alive.
+	 * Registered {@link org.springframework.kafka.core.ProducerFactory.Listener}s are
+	 * also added to the new factory. If the factory implementation does not support
+	 * the copy operation, a generic copy of the ProducerFactory is created which will
+	 * be of type
+	 * DefaultKafkaProducerFactory.
 	 * @param producerFactory the producer factory.
 	 * @param autoFlush true to flush after each send.
 	 * @param configOverrides producer configuration properties to override.
@@ -174,21 +183,57 @@ public KafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush,
 		this.micrometerEnabled = KafkaUtils.MICROMETER_PRESENT;
 		this.customProducerFactory = configOverrides != null && configOverrides.size() > 0;
 		if (this.customProducerFactory) {
-			Map<String, Object> configs = new HashMap<>(producerFactory.getConfigurationProperties());
-			configs.putAll(configOverrides);
-			DefaultKafkaProducerFactory<K, V> newFactory = new DefaultKafkaProducerFactory<>(configs,
-					producerFactory.getKeySerializerSupplier(), producerFactory.getValueSerializerSupplier());
-			newFactory.setPhysicalCloseTimeout((int) producerFactory.getPhysicalCloseTimeout().getSeconds());
-			newFactory.setProducerPerConsumerPartition(producerFactory.isProducerPerConsumerPartition());
-			newFactory.setProducerPerThread(producerFactory.isProducerPerThread());
-			this.producerFactory = newFactory;
+			this.producerFactory = copyProducerFactoryWithOverrides(producerFactory, configOverrides);
 		}
 		else {
 			this.producerFactory = producerFactory;
 		}
 		this.transactional = this.producerFactory.transactionCapable();
 	}
 
+	private ProducerFactory<K, V> copyProducerFactoryWithOverrides(ProducerFactory<K, V> templateFactory,
+			Map<String, Object> configOverrides) {
+
+		ProducerFactory<K, V> newFactory;
+		try {
+			newFactory = templateFactory.copyWithConfigurationOverride(configOverrides);
+		}
+		catch (UnsupportedOperationException e) {
+			newFactory = handleNonCopyableProducerFactory(templateFactory, configOverrides);
+		}
+
+		return newFactory;
+	}
+
+	/**
+	 * This method copies a ProducerFactory that misses the implementation of
+	 * {@link org.springframework.kafka.core.ProducerFactory#copyWithConfigurationOverride(java.util.Map)}.
+	 *
+	 * @param templateFactory the ProducerFactory to copy from
+	 * @param configOverrides new properties to be applied onto the templateFactory properties
+	 * @return a DefaultKafkaProducerFactory configured with configOverrides and all
+	 * public reachable settings of ProducerFactory
+	 */
+	private DefaultKafkaProducerFactory<K, V> handleNonCopyableProducerFactory(ProducerFactory<K, V> templateFactory,
+			Map<String, Object> configOverrides) {
+
+		Map<String, Object> producerProperties = new HashMap<>(templateFactory.getConfigurationProperties());
+		producerProperties.putAll(configOverrides);
+		DefaultKafkaProducerFactory<K, V> defaultFactory = new DefaultKafkaProducerFactory<>(producerProperties,
+				templateFactory.getKeySerializerSupplier(),
+				templateFactory.getValueSerializerSupplier());
+		defaultFactory.setPhysicalCloseTimeout((int) templateFactory.getPhysicalCloseTimeout().getSeconds());
+		defaultFactory.setProducerPerConsumerPartition(templateFactory.isProducerPerConsumerPartition());
+		defaultFactory.setProducerPerThread(templateFactory.isProducerPerThread());
+		for (ProducerPostProcessor<K, V> templatePostProcessor : templateFactory.getPostProcessors()) {
+			defaultFactory.addPostProcessor(templatePostProcessor);
+		}
+		for (ProducerFactory.Listener<K, V> templateListener : templateFactory.getListeners()) {
+			defaultFactory.addListener(templateListener);
+		}
+		return defaultFactory;
+	}
+
 	@Override
 	public void setBeanName(String name) {
 		this.beanName = name;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -34,6 +34,7 @@
  * @param <V> the value type.
  *
  * @author Gary Russell
+ * @author Thomas Strauß
  */
 public interface ProducerFactory<K, V> {
 
@@ -141,7 +142,7 @@ default Supplier<Serializer<K>> getKeySerializerSupplier() {
 
 	/**
 	 * Return true when there is a producer per thread.
-	 * @return the produver per thread.
+	 * @return the producer per thread.
 	 * @since 2.5
 	 */
 	default boolean isProducerPerThread() {
@@ -250,6 +251,26 @@ default void updateConfigs(Map<String, Object> updates) {
 	default void removeConfig(String configKey) {
 	}
 
+	/**
+	 * Copy the properties of the instance and the given properties to create a new producer factory.
+	 * <p>The copy shall prioritize the override properties over the configured values.
+	 * It is in the responsibility of the factory implementation to make sure the
+	 * configuration of the new factory is identical, complete and correct.</p>
+	 * <p>ProducerPostProcessor and Listeners must stay intact.</p>
+	 * <p>If the factory does not implement this method, an exception will be thrown.</p>
+	 * <p>Note: see
+	 * {@link org.springframework.kafka.core.DefaultKafkaProducerFactory#copyWithConfigurationOverride}</p>
+	 * @param overrideProperties the properties to be applied to the new factory
+	 * @return {@link org.springframework.kafka.core.ProducerFactory} with properties
+	 * applied
+	 * @since 2.5.17
+	 * @see org.springframework.kafka.core.KafkaTemplate#KafkaTemplate(ProducerFactory, java.util.Map)
+	 */
+	default ProducerFactory<K, V> copyWithConfigurationOverride(Map<String, Object> overrideProperties) {
+		throw new UnsupportedOperationException(
+				"This factory implementation doesn't support creating reconfigured copies.");
+	}
+
 	/**
 	 * Called whenever a producer is added or removed.
 	 *
@@ -271,9 +292,8 @@ default void producerAdded(String id, Producer<K, V> producer) {
 		}
 
 		/**
-		 * An exsting producer was removed.
-		 * @param id the producer id (factory bean name and client.id separated by a
-		 * period).
+		 * An existing producer was removed.
+		 * @param id the producer id (factory bean name and client.id separated by a period).
 		 * @param producer the producer.
 		 */
 		default void producerRemoved(String id, Producer<K, V> producer) {