SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Native compilation changes

Browse Source 
This commit effectively reverts 4e6881830a

Remove spring.factories changes introduced in the commit mentiond above.
Revert the binder to use spring.binders mechanism to register binder configurations.
This commit is contained in:
Soby Chacko
2021-04-29 19:07:33 -04:00
parent 7e48afa005
commit 79497ad264
9 changed files with 135 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java
View File

@@ -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<BindingCreatedEvent>, BeanFactoryAware, InitializingBean {
implements MeterBinder, ApplicationListener<BindingCreatedEvent> {
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<String, Long> 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);
}
}

23
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -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<ExtendedConsumerProperties<KafkaConsumerProperties>, ExtendedProducerProperties<KafkaProducerProperties>, KafkaTopicProvisioner>
// @checkstyle:on
implements
ExtendedPropertiesBinder<MessageChannel, KafkaConsumerProperties, KafkaProducerProperties>, SmartInitializingSingleton, BeanFactoryAware {
ExtendedPropertiesBinder<MessageChannel, KafkaConsumerProperties, KafkaProducerProperties> {
/**
* 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;

124
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java
View File

@@ -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<KafkaBindingRebalanceListener> rebalanceListener,
ObjectProvider<DlqPartitionFunction> dlqPartitionFunction,
ObjectProvider<DlqDestinationResolver> dlqDestinationResolver,
ObjectProvider<ClientFactoryCustomizer> clientFactoryCustomizer,
ObjectProvider<ConsumerConfigCustomizer> consumerConfigCustomizer,
ObjectProvider<ProducerConfigCustomizer> 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.
*/

83
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderMetricsConfiguration.java
View File

@@ -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));
}
}
};
}
}
}

89
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderMetricsConfigurationWithMultiBinder.java
View File

@@ -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));
}
}
};
}
}
}

0
spring-cloud-stream-binder-kafka/src/main/resources/spring.binders → spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.binders
View File

6
spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories
View File

@@ -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

10
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java
View File

@@ -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

4
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderMeterRegistryTest.java
View File

@@ -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();