Skip to content

Apply consumer config customizers to the Kafka health indicator #3315

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
2748109647 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Apply consumer config customizers to the Kafka health indicator #3315

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e16d5b9
Apply consumer customizers to binder health indicator
2748109647 Oct 26, 2026
0338cfb
Merge remote-tracking branch 'origin/main' into followup/3315
2748109647 Nov 2, 2026
2480ab2
test: accept health indicator customizer binding
2748109647 Nov 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion ...ngframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,13 +22,15 @@
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;
import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealth;
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;
Expand All @@ -53,7 +55,8 @@ public class KafkaBinderHealthIndicatorConfiguration {
@Bean
public KafkaBinderHealthIndicator kafkaBinderHealthIndicator(
KafkaMessageChannelBinder kafkaMessageChannelBinder,
KafkaBinderConfigurationProperties configurationProperties) {
KafkaBinderConfigurationProperties configurationProperties,
ObjectProvider<ConsumerConfigCustomizer> consumerConfigCustomizers) {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class);
Expand All @@ -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);
Expand Down
24 changes: 23 additions & 1 deletion ...rc/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderPropertiesTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -32,13 +33,16 @@
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;
import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration;
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;
Expand All @@ -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 {
Expand Down Expand Up @@ -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<String, Object> healthConsumerConfigs = (Map<String, Object>) 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");
Expand Down Expand Up @@ -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");
}

}

}
4 changes: 2 additions & 2 deletions .../springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,8 +99,8 @@ public Function<String, String> 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());
};
Expand Down
1 change: 1 addition & 0 deletions docs/modules/ROOT/pages/kafka/kafka_overview.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down