Add max age configuration for Kafka transactions

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java

@@ -37,6 +37,7 @@
 import org.apache.kafka.clients.producer.ProducerConfig;
 
 import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.kafka.autoconfigure.KafkaConfigBuilder;
 import org.springframework.boot.kafka.autoconfigure.KafkaConnectionDetails;
 import org.springframework.boot.kafka.autoconfigure.KafkaProperties;
 import org.springframework.cloud.stream.binder.HeaderMode;
@@ -400,7 +401,8 @@ public void setProducerProperties(Map<String, String> producerProperties) {
 	 * @return the merged properties.
 	 */
 	public Map<String, Object> mergedConsumerConfiguration() {
-		Map<String, Object> consumerConfiguration = new HashMap<>(this.kafkaProperties.buildConsumerProperties());
+		Map<String, Object> consumerConfiguration = new HashMap<>(
+			KafkaConfigBuilder.of(this.kafkaProperties).consumer().build());
 		if (!consumerConfiguration.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG) && this.kafkaConnectionDetails != null) {
 			consumerConfiguration.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.kafkaConnectionDetails.getConsumer().getBootstrapServers());
 		}
@@ -429,7 +431,8 @@ public Map<String, Object> mergedConsumerConfiguration() {
 	 * @return the merged properties.
 	 */
 	public Map<String, Object> mergedProducerConfiguration() {
-		Map<String, Object> producerConfiguration = new HashMap<>(this.kafkaProperties.buildProducerProperties());
+		Map<String, Object> producerConfiguration = new HashMap<>(
+			KafkaConfigBuilder.of(this.kafkaProperties).producer().build());
 		if (!producerConfiguration.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG) && this.kafkaConnectionDetails != null) {
 			producerConfiguration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.kafkaConnectionDetails.getProducer().getBootstrapServers());
 		}
@@ -543,6 +546,8 @@ public static class Transaction {
 
 		private String transactionIdPrefix;
 
+		private Duration maxAge;
+
 		public String getTransactionIdPrefix() {
 			return this.transactionIdPrefix;
 		}
@@ -551,6 +556,14 @@ public void setTransactionIdPrefix(String transactionIdPrefix) {
 			this.transactionIdPrefix = transactionIdPrefix;
 		}
 
+		public Duration getMaxAge() {
+			return this.maxAge;
+		}
+
+		public void setMaxAge(Duration maxAge) {
+			this.maxAge = maxAge;
+		}
+
 		public CombinedProducerProperties getProducer() {
 			return this.producer;
 		}

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java

@@ -20,6 +20,7 @@
 import java.io.OutputStream;
 import java.net.InetSocketAddress;
 import java.nio.file.Paths;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.Map;
 
@@ -47,6 +48,17 @@ void defaultRequiredAcksIsAll() {
 		assertThat(kafkaBinderConfigurationProperties.getRequiredAcks()).isEqualTo("all");
 	}
 
+	@Test
+	void transactionMaxAgeCanBeConfigured() {
+		KafkaBinderConfigurationProperties.Transaction transaction =
+			new KafkaBinderConfigurationProperties.Transaction();
+
+		Duration maxAge = Duration.ofMinutes(5);
+		transaction.setMaxAge(maxAge);
+
+		assertThat(transaction.getMaxAge()).isEqualTo(maxAge);
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	void mergedConsumerConfigurationFiltersGroupIdFromKafkaProperties() {

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -566,6 +566,10 @@ protected DefaultKafkaProducerFactory<byte[], byte[]> getProducerFactory(
 				props);
 		if (transactionIdPrefix != null) {
 			producerFactory.setTransactionIdPrefix(transactionIdPrefix);
+			Duration maxAge = this.configurationProperties.getTransaction().getMaxAge();
+			if (maxAge != null) {
+				producerFactory.setMaxAge(maxAge);
+			}
 		}
 		if (kafkaProducerProperties.getCloseTimeout() > 0) {
 			producerFactory.setPhysicalCloseTimeout(kafkaProducerProperties.getCloseTimeout());

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationWithTransactionsTest.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.stream.binder.kafka;
 
 import java.lang.reflect.Field;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -47,6 +48,7 @@ public class KafkaBinderConfigurationWithTransactionsTest {
 		.withUserConfiguration(IntegrationAutoConfiguration.class, KafkaBinderConfiguration.class, KafkaAutoConfiguration.class)
 		.withPropertyValues(
 			"spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix=test-tx-",
+			"spring.cloud.stream.kafka.binder.transaction.max-age=5m",
 			"spring.kafka.bootstrap-servers=localhost:9092");
 
 	@Test
@@ -75,6 +77,14 @@ public void clientFactoryCustomizersAppliedToTransactionManager() {
 				ProducerFactory<?, ?> producerFactory = transactionManager.getProducerFactory();
 				assertThat(producerFactory).isNotNull();
 
+				org.springframework.kafka.core.DefaultKafkaProducerFactory<?, ?> defaultProducerFactory =
+					(org.springframework.kafka.core.DefaultKafkaProducerFactory<?, ?>) producerFactory;
+
+				assertThat(defaultProducerFactory.getClass())
+					.isEqualTo(org.springframework.kafka.core.DefaultKafkaProducerFactory.class);
+				assertThat(org.springframework.test.util.ReflectionTestUtils.getField(defaultProducerFactory, "maxAge"))
+					.isEqualTo(Duration.ofMinutes(5).toMillis());
+
 				// Verify customizer was applied - check if our flag was set
 				TransactionClientFactoryCustomizerConfig config =
 					context.getBean(TransactionClientFactoryCustomizerConfig.class);