GH-2070: Producer Factory Improvements

Resolves #2070

Update the DefaultKafkaProducerFactory to support producer configs overrides

Restored the original copyright

Added missing Java doc

Author: frosiere

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

@@ -691,17 +691,7 @@ private Producer<K, V> getOrCreateThreadBoundProducer() {
 	 * @return the producer.
 	 */
 	protected Producer<K, V> createKafkaProducer() {
-		Map<String, Object> newConfigs;
-		if (this.clientIdPrefix == null) {
-			newConfigs = new HashMap<>(this.configs);
-		}
-		else {
-			newConfigs = new HashMap<>(this.configs);
-			newConfigs.put(ProducerConfig.CLIENT_ID_CONFIG,
-					this.clientIdPrefix + "-" + this.clientIdCounter.incrementAndGet());
-		}
-		checkBootstrap(newConfigs);
-		return createRawProducer(newConfigs);
+		return createRawProducer(getProducerConfigs());
 	}
 
 	protected Producer<K, V> createTransactionalProducerForPartition() {
@@ -826,17 +816,7 @@ boolean cacheReturner(CloseSafeProducer<K, V> producerToRemove, Duration timeout
 
 	private CloseSafeProducer<K, V> doCreateTxProducer(String prefix, String suffix,
 			BiPredicate<CloseSafeProducer<K, V>, Duration> remover) {
-
-		Producer<K, V> newProducer;
-		Map<String, Object> newProducerConfigs = new HashMap<>(this.configs);
-		String txId = prefix + suffix;
-		newProducerConfigs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, txId);
-		if (this.clientIdPrefix != null) {
-			newProducerConfigs.put(ProducerConfig.CLIENT_ID_CONFIG,
-					this.clientIdPrefix + "-" + this.clientIdCounter.incrementAndGet());
-		}
-		checkBootstrap(newProducerConfigs);
-		newProducer = createRawProducer(newProducerConfigs);
+		Producer<K, V> newProducer = createRawProducer(getTxProducerConfigs(prefix + suffix));
 		try {
 			newProducer.initTransactions();
 		}
@@ -908,6 +888,35 @@ public void closeThreadBoundProducer() {
 		}
 	}
 
+	/**
+	 * Return the configuration of a producer.
+	 * @return the configuration of a producer.
+	 * @since 2.8.3
+	 * @see #createKafkaProducer()
+	 */
+	protected Map<String, Object> getProducerConfigs() {
+		final Map<String, Object> newProducerConfigs = new HashMap<>(this.configs);
+		checkBootstrap(newProducerConfigs);
+		if (this.clientIdPrefix != null) {
+			newProducerConfigs.put(ProducerConfig.CLIENT_ID_CONFIG,
+					this.clientIdPrefix + "-" + this.clientIdCounter.incrementAndGet());
+		}
+		return newProducerConfigs;
+	}
+
+	/**
+	 * Return the configuration of a transactional producer.
+	 * @param transactionId the transactionId.
+	 * @return the configuration of a transactional producer.
+	 * @since 2.8.3
+	 * @see #doCreateTxProducer(String, String, BiPredicate)
+	 */
+	protected Map<String, Object> getTxProducerConfigs(String transactionId) {
+		final Map<String, Object> newProducerConfigs = getProducerConfigs();
+		newProducerConfigs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionId);
+		return newProducerConfigs;
+	}
+
 	/**
 	 * A wrapper class for the delegate.
 	 *

spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.java

@@ -41,6 +41,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.MockProducer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.KafkaException;
@@ -499,4 +500,28 @@ void configSerializer() {
 		verify(value).configure(any(), eq(false));
 	}
 
+	@Test
+	void testConfigOverridesOfTransactionalProducer() {
+		final Map<String, Object> producerFactoryConfigs = Map.of("linger.ms", 100);
+		final Map<String, Object> producerConfigs = new HashMap<>();
+		final DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(producerFactoryConfigs) {
+			@Override
+			protected Map<String, Object> getTxProducerConfigs(String transactionId) {
+				final Map<String, Object> newProducerConfigs = super.getTxProducerConfigs(transactionId);
+				newProducerConfigs.put("linger.ms", 200);
+				return newProducerConfigs;
+			}
+
+			@Override
+			protected Producer<String, String> createRawProducer(Map<String, Object> rawConfigs) {
+				producerConfigs.putAll(rawConfigs);
+				return new MockProducer<>();
+			}
+		};
+		pf.setTransactionIdPrefix("tx-prefix");
+		pf.createProducer();
+		assertThat(producerFactoryConfigs).containsEntry("linger.ms", 100);
+		assertThat(producerConfigs).containsEntry("linger.ms", 200);
+	}
+
 }