diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java index 9917e43d9..a9c8a36b9 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java @@ -42,10 +42,6 @@ import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.ByteArrayDeserializer; -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.BeanFactoryAware; -import org.springframework.beans.factory.InitializingBean; import org.springframework.cloud.stream.binder.BindingCreatedEvent; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.context.ApplicationListener; @@ -66,7 +62,7 @@ import org.springframework.util.ObjectUtils; * @author Gary Russell */ public class KafkaBinderMetrics - implements MeterBinder, ApplicationListener, BeanFactoryAware, InitializingBean { + implements MeterBinder, ApplicationListener { private static final int DEFAULT_TIMEOUT = 5; @@ -79,7 +75,7 @@ public class KafkaBinderMetrics */ public static final String OFFSET_LAG_METRIC_NAME = "spring.cloud.stream.binder.kafka.offset"; - private KafkaMessageChannelBinder binder; + private final KafkaMessageChannelBinder binder; private final KafkaBinderConfigurationProperties binderConfigurationProperties; @@ -95,21 +91,22 @@ public class KafkaBinderMetrics Map unconsumedMessages = new ConcurrentHashMap<>(); - private BeanFactory beanFactory; - - public KafkaBinderMetrics(KafkaBinderConfigurationProperties binderConfigurationProperties, + public KafkaBinderMetrics(KafkaMessageChannelBinder binder, + KafkaBinderConfigurationProperties binderConfigurationProperties, ConsumerFactory defaultConsumerFactory, @Nullable MeterRegistry meterRegistry) { + this.binder = binder; this.binderConfigurationProperties = binderConfigurationProperties; this.defaultConsumerFactory = defaultConsumerFactory; this.meterRegistry = meterRegistry; this.metadataConsumers = new ConcurrentHashMap<>(); } - public KafkaBinderMetrics(KafkaBinderConfigurationProperties binderConfigurationProperties) { + public KafkaBinderMetrics(KafkaMessageChannelBinder binder, + KafkaBinderConfigurationProperties binderConfigurationProperties) { - this(binderConfigurationProperties, null, null); + this(binder, binderConfigurationProperties, null, null); } public void setTimeout(int timeout) { @@ -244,13 +241,4 @@ public class KafkaBinderMetrics } } - @Override - public void setBeanFactory(BeanFactory beanFactory) throws BeansException { - this.beanFactory = beanFactory; - } - - @Override - public void afterPropertiesSet() throws Exception { - this.binder = this.beanFactory.getBean(KafkaMessageChannelBinder.class); - } } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 89743f8df..ad32cb47e 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -52,11 +52,8 @@ import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.NoSuchBeanDefinitionException; -import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider; @@ -162,7 +159,7 @@ public class KafkaMessageChannelBinder extends AbstractMessageChannelBinder, ExtendedProducerProperties, KafkaTopicProvisioner> // @checkstyle:on implements - ExtendedPropertiesBinder, SmartInitializingSingleton, BeanFactoryAware { + ExtendedPropertiesBinder { /** * Kafka header for x-exception-fqcn. @@ -236,8 +233,6 @@ public class KafkaMessageChannelBinder extends private ConsumerConfigCustomizer consumerConfigCustomizer; - private BeanFactory beanFactory; - public KafkaMessageChannelBinder( KafkaBinderConfigurationProperties configurationProperties, KafkaTopicProvisioner provisioningProvider) { @@ -284,22 +279,6 @@ public class KafkaMessageChannelBinder extends this.dlqDestinationResolver = dlqDestinationResolver; } - @Override - public void afterSingletonsInstantiated() { - try { - final ClientFactoryCustomizer clientFactoryCustomizer = beanFactory.getBean(ClientFactoryCustomizer.class); - setClientFactoryCustomizer(clientFactoryCustomizer); - } - catch (Exception e) { - // nothing to do. - } - } - - @Override - public void setBeanFactory(BeanFactory beanFactory) throws BeansException { - this.beanFactory = beanFactory; - } - private static String[] headersToMap( KafkaBinderConfigurationProperties configurationProperties) { String[] headersToMap; diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index d34f33444..953bab7e1 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java @@ -18,14 +18,20 @@ package org.springframework.cloud.stream.binder.kafka.config; import java.io.IOException; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.binder.MeterBinder; + import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.annotation.StreamMessageConverter; import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics; import org.springframework.cloud.stream.binder.kafka.KafkaBindingRebalanceListener; import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder; import org.springframework.cloud.stream.binder.kafka.KafkaNullConverter; @@ -40,12 +46,20 @@ import org.springframework.cloud.stream.config.ConsumerEndpointCustomizer; import org.springframework.cloud.stream.config.ListenerContainerCustomizer; import org.springframework.cloud.stream.config.MessageSourceCustomizer; import org.springframework.cloud.stream.config.ProducerMessageHandlerCustomizer; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; import org.springframework.integration.kafka.inbound.KafkaMessageSource; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.MicrometerConsumerListener; +import org.springframework.kafka.core.MicrometerProducerListener; +import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer; import org.springframework.kafka.support.LoggingProducerListener; @@ -98,7 +112,7 @@ public class KafkaBinderConfiguration { } @SuppressWarnings("unchecked") - @Bean("kafka") + @Bean KafkaMessageChannelBinder kafkaMessageChannelBinder( KafkaBinderConfigurationProperties configurationProperties, KafkaTopicProvisioner provisioningProvider, @@ -109,6 +123,7 @@ public class KafkaBinderConfiguration { ObjectProvider rebalanceListener, ObjectProvider dlqPartitionFunction, ObjectProvider dlqDestinationResolver, + ObjectProvider clientFactoryCustomizer, ObjectProvider consumerConfigCustomizer, ObjectProvider producerConfigCustomizer ) { @@ -122,6 +137,7 @@ public class KafkaBinderConfiguration { .setExtendedBindingProperties(this.kafkaExtendedBindingProperties); kafkaMessageChannelBinder.setProducerMessageHandlerCustomizer(messageHandlerCustomizer); kafkaMessageChannelBinder.setConsumerEndpointCustomizer(consumerCustomizer); + kafkaMessageChannelBinder.setClientFactoryCustomizer(clientFactoryCustomizer.getIfUnique()); kafkaMessageChannelBinder.setConsumerConfigCustomizer(consumerConfigCustomizer.getIfUnique()); kafkaMessageChannelBinder.setProducerConfigCustomizer(producerConfigCustomizer.getIfUnique()); return kafkaMessageChannelBinder; @@ -162,6 +178,112 @@ public class KafkaBinderConfiguration { return kafkaJaasLoginModuleInitializer; } + @Configuration + @ConditionalOnMissingBean(value = KafkaBinderMetrics.class, name = "outerContext") + @ConditionalOnClass(name = "io.micrometer.core.instrument.MeterRegistry") + protected class KafkaBinderMetricsConfiguration { + + @Bean + @ConditionalOnBean(MeterRegistry.class) + @ConditionalOnMissingBean(KafkaBinderMetrics.class) + public MeterBinder kafkaBinderMetrics( + KafkaMessageChannelBinder kafkaMessageChannelBinder, + KafkaBinderConfigurationProperties configurationProperties, + MeterRegistry meterRegistry) { + + return new KafkaBinderMetrics(kafkaMessageChannelBinder, + configurationProperties, null, meterRegistry); + } + + @ConditionalOnClass(name = "org.springframework.kafka.core.MicrometerConsumerListener") + @ConditionalOnBean(MeterRegistry.class) + protected class KafkaMicrometer { + + @Bean + @ConditionalOnMissingBean(name = "binderClientFactoryCustomizer") + public ClientFactoryCustomizer binderClientFactoryCustomizer(MeterRegistry meterRegistry) { + + return new ClientFactoryCustomizer() { + + @Override + public void configure(ProducerFactory pf) { + if (pf instanceof DefaultKafkaProducerFactory) { + ((DefaultKafkaProducerFactory) pf) + .addListener(new MicrometerProducerListener<>(meterRegistry)); + } + } + + @Override + public void configure(ConsumerFactory cf) { + if (cf instanceof DefaultKafkaConsumerFactory) { + ((DefaultKafkaConsumerFactory) cf) + .addListener(new MicrometerConsumerListener<>(meterRegistry)); + } + } + + }; + + } + + } + + } + + @Configuration + @ConditionalOnBean(name = "outerContext") + @ConditionalOnMissingBean(KafkaBinderMetrics.class) + @ConditionalOnClass(name = "io.micrometer.core.instrument.MeterRegistry") + protected class KafkaBinderMetricsConfigurationWithMultiBinder { + + @Bean + public MeterBinder kafkaBinderMetrics( + KafkaMessageChannelBinder kafkaMessageChannelBinder, + KafkaBinderConfigurationProperties configurationProperties, + ConfigurableApplicationContext context) { + + MeterRegistry meterRegistry = context.getBean("outerContext", ApplicationContext.class) + .getBean(MeterRegistry.class); + return new KafkaBinderMetrics(kafkaMessageChannelBinder, + configurationProperties, null, meterRegistry); + } + + @ConditionalOnClass(name = "org.springframework.kafka.core.MicrometerConsumerListener") + protected class KafkaMicrometer { + + @Bean + @ConditionalOnMissingBean(name = "binderClientFactoryCustomizer") + public ClientFactoryCustomizer binderClientFactoryCustomizer(ConfigurableApplicationContext context) { + + + return new ClientFactoryCustomizer() { + + MeterRegistry meterRegistry = context.getBean("outerContext", ApplicationContext.class) + .getBean(MeterRegistry.class); + + @Override + public void configure(ProducerFactory pf) { + if (pf instanceof DefaultKafkaProducerFactory) { + ((DefaultKafkaProducerFactory) pf) + .addListener(new MicrometerProducerListener<>(this.meterRegistry)); + } + } + + @Override + public void configure(ConsumerFactory cf) { + if (cf instanceof DefaultKafkaConsumerFactory) { + ((DefaultKafkaConsumerFactory) cf) + .addListener(new MicrometerConsumerListener<>(this.meterRegistry)); + } + } + + }; + + } + + } + + } + /** * Properties configuration for Jaas. */ diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderMetricsConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderMetricsConfiguration.java deleted file mode 100644 index 36df906e2..000000000 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderMetricsConfiguration.java +++ /dev/null @@ -1,83 +0,0 @@ -/* - * Copyright 2018-2021 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.binder.kafka.config; - -import io.micrometer.core.instrument.MeterRegistry; -import io.micrometer.core.instrument.binder.MeterBinder; - -import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; -import org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.kafka.core.ConsumerFactory; -import org.springframework.kafka.core.DefaultKafkaConsumerFactory; -import org.springframework.kafka.core.DefaultKafkaProducerFactory; -import org.springframework.kafka.core.MicrometerConsumerListener; -import org.springframework.kafka.core.MicrometerProducerListener; -import org.springframework.kafka.core.ProducerFactory; - -@Configuration -@ConditionalOnMissingBean(value = KafkaBinderMetrics.class, name = "outerContext") -@ConditionalOnClass(name = "io.micrometer.core.instrument.MeterRegistry") -public class KafkaBinderMetricsConfiguration { - - @Bean - @ConditionalOnBean(MeterRegistry.class) - @ConditionalOnMissingBean(KafkaBinderMetrics.class) - public MeterBinder kafkaBinderMetrics( - KafkaBinderConfigurationProperties configurationProperties, - MeterRegistry meterRegistry) { - - return new KafkaBinderMetrics(configurationProperties, null, meterRegistry); - } - - @ConditionalOnClass(name = "org.springframework.kafka.core.MicrometerConsumerListener") - @ConditionalOnBean(MeterRegistry.class) - public class KafkaMicrometer { - - @Bean - @ConditionalOnMissingBean(name = "binderClientFactoryCustomizer") - public ClientFactoryCustomizer binderClientFactoryCustomizer(MeterRegistry meterRegistry) { - - return new ClientFactoryCustomizer() { - - @Override - public void configure(ProducerFactory pf) { - if (pf instanceof DefaultKafkaProducerFactory) { - ((DefaultKafkaProducerFactory) pf) - .addListener(new MicrometerProducerListener<>(meterRegistry)); - } - } - - @Override - public void configure(ConsumerFactory cf) { - if (cf instanceof DefaultKafkaConsumerFactory) { - ((DefaultKafkaConsumerFactory) cf) - .addListener(new MicrometerConsumerListener<>(meterRegistry)); - } - } - - }; - - } - - } - -} diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderMetricsConfigurationWithMultiBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderMetricsConfigurationWithMultiBinder.java deleted file mode 100644 index 686e6e66d..000000000 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderMetricsConfigurationWithMultiBinder.java +++ /dev/null @@ -1,89 +0,0 @@ -/* - * Copyright 2018-2021 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.binder.kafka.config; - -import io.micrometer.core.instrument.MeterRegistry; -import io.micrometer.core.instrument.binder.MeterBinder; - -import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; -import org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.kafka.core.ConsumerFactory; -import org.springframework.kafka.core.DefaultKafkaConsumerFactory; -import org.springframework.kafka.core.DefaultKafkaProducerFactory; -import org.springframework.kafka.core.MicrometerConsumerListener; -import org.springframework.kafka.core.MicrometerProducerListener; -import org.springframework.kafka.core.ProducerFactory; - -@Configuration -@ConditionalOnBean(name = "outerContext") -@ConditionalOnMissingBean(KafkaBinderMetrics.class) -@ConditionalOnClass(name = "io.micrometer.core.instrument.MeterRegistry") -public class KafkaBinderMetricsConfigurationWithMultiBinder { - - @Bean - public MeterBinder kafkaBinderMetrics( - KafkaBinderConfigurationProperties configurationProperties, - ConfigurableApplicationContext context) { - - MeterRegistry meterRegistry = context.getBean("outerContext", ApplicationContext.class) - .getBean(MeterRegistry.class); - return new KafkaBinderMetrics(configurationProperties, null, meterRegistry); - } - - @ConditionalOnClass(name = "org.springframework.kafka.core.MicrometerConsumerListener") - protected class KafkaMicrometer { - - @Bean - @ConditionalOnMissingBean(name = "binderClientFactoryCustomizer") - public ClientFactoryCustomizer binderClientFactoryCustomizer(ConfigurableApplicationContext context) { - - - return new ClientFactoryCustomizer() { - - MeterRegistry meterRegistry = context.getBean("outerContext", ApplicationContext.class) - .getBean(MeterRegistry.class); - - @Override - public void configure(ProducerFactory pf) { - if (pf instanceof DefaultKafkaProducerFactory) { - ((DefaultKafkaProducerFactory) pf) - .addListener(new MicrometerProducerListener<>(this.meterRegistry)); - } - } - - @Override - public void configure(ConsumerFactory cf) { - if (cf instanceof DefaultKafkaConsumerFactory) { - ((DefaultKafkaConsumerFactory) cf) - .addListener(new MicrometerConsumerListener<>(this.meterRegistry)); - } - } - - }; - - } - - } - -} diff --git a/spring-cloud-stream-binder-kafka/src/main/resources/spring.binders b/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.binders similarity index 100% rename from spring-cloud-stream-binder-kafka/src/main/resources/spring.binders rename to spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.binders diff --git a/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories index 84921eb5e..1d2e61ded 100644 --- a/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories @@ -1,8 +1,4 @@ org.springframework.boot.env.EnvironmentPostProcessor:\ org.springframework.cloud.stream.binder.kafka.KafkaBinderEnvironmentPostProcessor org.springframework.boot.autoconfigure.EnableAutoConfiguration:\ -org.springframework.cloud.stream.binder.kafka.config.ExtendedBindingHandlerMappingsProviderConfiguration,\ -org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration,\ -org.springframework.cloud.stream.binder.kafka.config.KafkaBinderMetricsConfiguration,\ -org.springframework.cloud.stream.binder.kafka.config.KafkaBinderMetricsConfigurationWithMultiBinder - +org.springframework.cloud.stream.binder.kafka.config.ExtendedBindingHandlerMappingsProviderConfiguration diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java index 9c0c0471c..480d7795c 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.binder.kafka; -import java.lang.reflect.Field; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -42,7 +41,6 @@ import org.mockito.MockitoAnnotations; import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.TopicInformation; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; -import org.springframework.util.ReflectionUtils; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; @@ -81,14 +79,8 @@ public class KafkaBinderMetricsTest { .createConsumer(ArgumentMatchers.any(), ArgumentMatchers.any())) .willReturn(consumer); org.mockito.BDDMockito.given(binder.getTopicsInUse()).willReturn(topicsInUse); - metrics = new KafkaBinderMetrics(kafkaBinderConfigurationProperties, + metrics = new KafkaBinderMetrics(binder, kafkaBinderConfigurationProperties, consumerFactory, null); - - Field binderField = ReflectionUtils - .findField(KafkaBinderMetrics.class, "binder", KafkaMessageChannelBinder.class); - ReflectionUtils.makeAccessible(binderField); - ReflectionUtils.setField(binderField, metrics, this.binder); - org.mockito.BDDMockito .given(consumer.endOffsets(ArgumentMatchers.anyCollection())) .willReturn(java.util.Collections diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderMeterRegistryTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderMeterRegistryTest.java index 3cf70311d..5e45deff5 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderMeterRegistryTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderMeterRegistryTest.java @@ -20,7 +20,6 @@ import java.util.function.Function; import io.micrometer.core.instrument.MeterRegistry; import org.junit.ClassRule; -import org.junit.Ignore; import org.junit.Test; import org.springframework.boot.WebApplicationType; @@ -57,7 +56,6 @@ public class KafkaBinderMeterRegistryTest { } @Test - @Ignore("temporarily") public void testMetricsWithMultiBinders() { ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder(SimpleApplication.class) .web(WebApplicationType.NONE) @@ -89,7 +87,7 @@ public class KafkaBinderMeterRegistryTest { .tag("topic", "inputTopic").gauge().value()).isNotNull(); // assert consumer metrics - //assertThatCode(() -> meterRegistry.get("kafka.consumer.connection.count").meter()).doesNotThrowAnyException(); + assertThatCode(() -> meterRegistry.get("kafka.consumer.connection.count").meter()).doesNotThrowAnyException(); // assert producer metrics assertThatCode(() -> meterRegistry.get("kafka.producer.connection.count").meter()).doesNotThrowAnyException();