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/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()); }; diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview.adoc b/docs/modules/ROOT/pages/kafka/kafka_overview.adoc index 67a3ddf9d0..5bf0250a02 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