From 8661fd19ae4b04d55311f3a894e8db69a56c9d84 Mon Sep 17 00:00:00 2001 From: 98001yash Date: Fri, 23 Oct 2026 16:56:29 +0530 Subject: [PATCH 1/2] Add max age configuration for Kafka transactions Signed-off-by: 98001yash --- .../KafkaBinderConfigurationProperties.java | 10 ++++++++++ .../KafkaBinderConfigurationPropertiesTest.java | 12 ++++++++++++ .../binder/kafka/KafkaMessageChannelBinder.java | 4 ++++ ...KafkaBinderConfigurationWithTransactionsTest.java | 10 ++++++++++ 4 files changed, 36 insertions(+) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index 85541e5f24..ee44715999 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java @@ -543,6 +543,8 @@ public static class Transaction { private String transactionIdPrefix; + private Duration maxAge; + public String getTransactionIdPrefix() { return this.transactionIdPrefix; } @@ -551,6 +553,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; } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java index 04634a159f..5465ca30b4 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java +++ b/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() { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 9d6a95b047..f66b193d8c 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/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 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()); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationWithTransactionsTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationWithTransactionsTest.java index 44ee107489..7fa4ac631a 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationWithTransactionsTest.java +++ b/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); From d754048f57b4dce9537e0c3b9c13260cfc3b56d3 Mon Sep 17 00:00:00 2001 From: 98001yash Date: Mon, 26 Oct 2026 23:47:24 +0530 Subject: [PATCH 2/2] Replace deprecated Kafka properties builder methods Signed-off-by: 98001yash --- .../properties/KafkaBinderConfigurationProperties.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index ee44715999..7bc3593aa1 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/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 producerProperties) { * @return the merged properties. */ public Map mergedConsumerConfiguration() { - Map consumerConfiguration = new HashMap<>(this.kafkaProperties.buildConsumerProperties()); + Map 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 mergedConsumerConfiguration() { * @return the merged properties. */ public Map mergedProducerConfiguration() { - Map producerConfiguration = new HashMap<>(this.kafkaProperties.buildProducerProperties()); + Map 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()); }