GH-1983: No ProducerFactory copy in KafkaTemplate

Fixes #1983

It is a misdirection to use a `DefaultProducerFactory`
in the `KafkaTemplate` to create a copy of the provided `ProducerFactory`.
Now it is a target project responsibility to implement `copyWithConfigurationOverride()`
properly

Author: artembilan

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

@@ -197,57 +197,14 @@ public KafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush,
 		this.micrometerEnabled = KafkaUtils.MICROMETER_PRESENT;
 		this.customProducerFactory = configOverrides != null && configOverrides.size() > 0;
 		if (this.customProducerFactory) {
-			this.producerFactory = copyProducerFactoryWithOverrides(producerFactory, configOverrides);
+			this.producerFactory = producerFactory.copyWithConfigurationOverride(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/test/java/org/springframework/kafka/core/KafkaTemplateTests.java

@@ -454,7 +454,7 @@ void testWithCallbackFailureFunctional() throws Exception {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicReference<SendResult<Integer, String>> theResult = new AtomicReference<>();
 		AtomicReference<String> value = new AtomicReference<>();
-		future.addCallback(result -> { }, (KafkaFailureCallback<Integer, String>) ex -> {
+		future.addCallback(result -> {}, (KafkaFailureCallback<Integer, String>) ex -> {
 			ProducerRecord<Integer, String> failed = ex.getFailedProducerRecord();
 			value.set(failed.value());
 			latch.countDown();
@@ -525,65 +525,6 @@ void testConfigOverridesWithDefaultKafkaProducerFactory() {
 		assertThat(templateWTX2_2.getProducerFactory().getTransactionIdPrefix()).isEqualTo("TX2");
 	}
 
-	@Test
-	void testConfigOverridesWithCustomProducerFactory() {
-		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
-		ProducerFactory<String, String> pf = new ProducerFactory<>() {
-
-			@Override
-			public Producer<String, String> createProducer() {
-				return null;
-			}
-
-			@Override
-			public List<Listener<String, String>> getListeners() {
-				return Collections.singletonList(noopListener);
-			}
-
-			@Override
-			public List<ProducerPostProcessor<String, String>> getPostProcessors() {
-				return Collections.singletonList(noopProducerPostProcessor);
-			}
-
-			@Override
-			public Map<String, Object> getConfigurationProperties() {
-				return Collections.singletonMap(ProducerConfig.ACKS_CONFIG, "all");
-			}
-
-			@Override
-			public Duration getPhysicalCloseTimeout() {
-				return Duration.ofSeconds(6);
-			}
-
-			@Override
-			public boolean isProducerPerConsumerPartition() {
-				return true;
-			}
-
-			@Override
-			public boolean isProducerPerThread() {
-				return false;
-			}
-		};
-
-		Map<String, Object> overrides = new HashMap<>();
-		overrides.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-		overrides.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "TX");
-		KafkaTemplate<String, String> template = new KafkaTemplate<>(pf, true, overrides);
-		assertThat(template.getProducerFactory()).isOfAnyClassIn(DefaultKafkaProducerFactory.class);
-		assertThat(template.getProducerFactory().getConfigurationProperties()
-				.get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)).isEqualTo(StringSerializer.class);
-		assertThat(template.getProducerFactory().getPhysicalCloseTimeout()).isEqualTo(Duration.ofSeconds(6));
-		assertThat(template.getProducerFactory().isProducerPerConsumerPartition()).isTrue();
-		assertThat(template.getProducerFactory().isProducerPerThread()).isFalse();
-		assertThat(template.isTransactional()).isTrue();
-		assertThat(template.getProducerFactory().getListeners()).isEqualTo(pf.getListeners());
-		assertThat(template.getProducerFactory().getListeners().size()).isEqualTo(1);
-		assertThat(template.getProducerFactory().getPostProcessors()).isEqualTo(pf.getPostProcessors());
-		assertThat(template.getProducerFactory().getPostProcessors().size()).isEqualTo(1);
-		assertThat(template.getProducerFactory().getTransactionIdPrefix()).isEqualTo("TX");
-	}
-
 	@Test
 	void testConfigOverridesWithSerializers() {
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);