From e16d5b9e6eb24e325e315c349231eaa40bf2df15 Mon Sep 17 00:00:00 2001 From: Alpha <2748109647@qq.com> Date: Mon, 26 Oct 2026 14:01:20 +0800 Subject: [PATCH 1/2] Apply consumer customizers to binder health indicator Resolves #2895 Signed-off-by: Alpha <2748109647@qq.com> --- ...fkaBinderHealthIndicatorConfiguration.java | 7 +++++- .../kafka/KafkaBinderPropertiesTest.java | 24 ++++++++++++++++++- .../ROOT/pages/kafka/kafka_overview.adoc | 1 + 3 files changed, 30 insertions(+), 2 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java index 45b727467e..788e936a42 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java @@ -22,6 +22,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.health.autoconfigure.contributor.ConditionalOnEnabledHealthIndicator; @@ -29,6 +30,7 @@ import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealthIndicator; import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.core.ConsumerFactory; @@ -53,7 +55,8 @@ public class KafkaBinderHealthIndicatorConfiguration { @Bean public KafkaBinderHealthIndicator kafkaBinderHealthIndicator( KafkaMessageChannelBinder kafkaMessageChannelBinder, - KafkaBinderConfigurationProperties configurationProperties) { + KafkaBinderConfigurationProperties configurationProperties, + ObjectProvider consumerConfigCustomizers) { Map props = new HashMap<>(); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); @@ -72,6 +75,8 @@ public KafkaBinderHealthIndicator kafkaBinderHealthIndicator( if (StringUtils.hasText(healthIndicatorConsumerGroup)) { props.put(ConsumerConfig.GROUP_ID_CONFIG, healthIndicatorConsumerGroup); } + consumerConfigCustomizers.orderedStream().forEach(customizer -> + customizer.configure(props, "kafkaBinderHealthIndicator", "kafkaBinderHealthIndicator")); ConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>(props); KafkaBinderHealthIndicator indicator = new KafkaBinderHealthIndicator( kafkaMessageChannelBinder, consumerFactory); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderPropertiesTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderPropertiesTest.java index dacb0459c4..72513eff07 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderPropertiesTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderPropertiesTest.java @@ -24,6 +24,7 @@ import java.util.Map; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerGroupMetadata; import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.ByteArraySerializer; @@ -32,6 +33,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.kafka.autoconfigure.KafkaAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.TestConfiguration; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.kafka.common.AbstractKafkaBinderHealthIndicator; @@ -39,6 +41,8 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; +import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer; +import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; @@ -52,7 +56,8 @@ * @author Soby Chacko */ @SpringBootTest(classes = { KafkaBinderConfiguration.class, KafkaAutoConfiguration.class, - KafkaBinderPropertiesTest.class }) + KafkaBinderPropertiesTest.class, + KafkaBinderPropertiesTest.HealthIndicatorCustomizerConfiguration.class }) @TestPropertySource(locations = "classpath:binder-config.properties", properties = "spring.cloud.stream.kafka.binder.healthIndicatorConsumerGroup=health-consumer-group") class KafkaBinderPropertiesTest { @@ -81,6 +86,12 @@ void kafkaBinderConfigurationProperties() throws Exception { ConsumerFactory healthIndicatorConsumerFactory = (ConsumerFactory) consumerFactoryField.get(this.kafkaBinderHealthIndicator); assertThat(healthIndicatorConsumerFactory).isNotNull(); + Field healthConsumerFactoryConfigField = ReflectionUtils + .findField(DefaultKafkaConsumerFactory.class, "configs", Map.class); + ReflectionUtils.makeAccessible(healthConsumerFactoryConfigField); + Map healthConsumerConfigs = (Map) ReflectionUtils + .getField(healthConsumerFactoryConfigField, healthIndicatorConsumerFactory); + assertThat(healthConsumerConfigs).containsEntry(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); Consumer consumer = healthIndicatorConsumerFactory.createConsumer(); ConsumerGroupMetadata consumerGroupMetadata = consumer.groupMetadata(); assertThat(consumerGroupMetadata.groupId()).isEqualTo("health-consumer-group"); @@ -144,4 +155,15 @@ void kafkaBinderConfigurationProperties() throws Exception { .contains("10.98.09.199:9082"))).isTrue(); } + @TestConfiguration(proxyBeanMethods = false) + static class HealthIndicatorCustomizerConfiguration { + + @Bean + ConsumerConfigCustomizer healthIndicatorConsumerConfigCustomizer() { + return (consumerProperties, bindingName, destination) -> + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + } + + } + } diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview.adoc b/docs/modules/ROOT/pages/kafka/kafka_overview.adoc index e943b9db4a..a8868787ea 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka_overview.adoc @@ -961,6 +961,7 @@ For example, if you want to gain access to a bean that is defined at the applica When the binder discovers that these customizers are available as beans, it will invoke the `configure` method right before creating the consumer and producer factories. Both of these interfaces also provide access to both the binding and destination names so that they can be accessed while customizing producer and consumer properties. +`ConsumerConfigCustomizer` is also applied to the Kafka binder health indicator's consumer. Since that consumer is not associated with a binding, both the binding and destination names are `kafkaBinderHealthIndicator`. [[admin-client-config-customization]] === Customizing AdminClient Configuration From 2480ab2c44d6a80e63b976cd075d4be48402855a Mon Sep 17 00:00:00 2001 From: Alpha <2748109647@qq.com> Date: Tue, 3 Nov 2026 06:41:32 +0800 Subject: [PATCH 2/2] test: accept health indicator customizer binding Signed-off-by: Alpha <2748109647@qq.com> --- .../kafka/integration/KafkaConfigCustomizationTests.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java index d0e00bebe4..5f730258fa 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java @@ -99,8 +99,8 @@ public Function process() { @Bean public ConsumerConfigCustomizer consumerConfigCustomizer() { return (consumerProperties, binding, destination) -> { - assertThat(binding).isEqualTo("process-in-0"); - assertThat(destination).isEqualTo("process-in-0"); + assertThat(binding).isIn("process-in-0", "kafkaBinderHealthIndicator"); + assertThat(destination).isEqualTo(binding); consumerProperties.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MyConsumerInterceptor.class.getName()); consumerProperties.put("foo.bean", foo()); };