From 8661fd19ae4b04d55311f3a894e8db69a56c9d84 Mon Sep 17 00:00:00 2001 From: 98001yash Date: Fri, 23 Oct 2026 16:56:29 +0530 Subject: [PATCH] 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);