From 9a4e86a7505336da73937023606831e02daec63a Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 20 Feb 2018 16:52:50 -0500 Subject: [PATCH] KafkaBinderConfigurationProperties duplicate bean ConfigurationProperties bean provided by Kafka Streams binder extends from `KafkaBinderConfigurationProperties` used by Kafka binder. It creates a conflict when autowiring this bean from Kafka binder configuration. This prevents an application to have both binders in the classpath. Change the creation of this ConfigurationProperties bean so that it avoids creating bean using EnbaleConfigurationProperties and then autowiring, but directly create the Bean using `@Bean`. This prevents the conflict. Resolves #244 Resolves #315 --- .../config/KafkaBinderConfiguration.java | 29 +++++++++++-------- 1 file changed, 17 insertions(+), 12 deletions(-) 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 c11dfe585..2073aa12d 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 @@ -62,14 +62,11 @@ import org.springframework.util.ObjectUtils; @Configuration @ConditionalOnMissingBean(Binder.class) @Import({ PropertyPlaceholderAutoConfiguration.class}) -@EnableConfigurationProperties({ KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class }) +@EnableConfigurationProperties({ KafkaExtendedBindingProperties.class }) public class KafkaBinderConfiguration { protected static final Log logger = LogFactory.getLog(KafkaBinderConfiguration.class); - @Autowired - private KafkaBinderConfigurationProperties configurationProperties; - @Autowired private KafkaExtendedBindingProperties kafkaExtendedBindingProperties; @@ -83,14 +80,20 @@ public class KafkaBinderConfiguration { private KafkaProperties kafkaProperties; @Bean - KafkaTopicProvisioner provisioningProvider() { - return new KafkaTopicProvisioner(this.configurationProperties, this.kafkaProperties); + KafkaBinderConfigurationProperties configurationProperties() { + return new KafkaBinderConfigurationProperties(); } @Bean - KafkaMessageChannelBinder kafkaMessageChannelBinder() { + KafkaTopicProvisioner provisioningProvider(KafkaBinderConfigurationProperties configurationProperties) { + return new KafkaTopicProvisioner(configurationProperties, this.kafkaProperties); + } + + @Bean + KafkaMessageChannelBinder kafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties, + KafkaTopicProvisioner provisioningProvider) { KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder( - this.configurationProperties, provisioningProvider()); + configurationProperties, provisioningProvider); kafkaMessageChannelBinder.setProducerListener(producerListener); kafkaMessageChannelBinder.setExtendedBindingProperties(this.kafkaExtendedBindingProperties); return kafkaMessageChannelBinder; @@ -103,7 +106,8 @@ public class KafkaBinderConfiguration { } @Bean - KafkaBinderHealthIndicator healthIndicator(KafkaMessageChannelBinder kafkaMessageChannelBinder) { + KafkaBinderHealthIndicator healthIndicator(KafkaMessageChannelBinder kafkaMessageChannelBinder, + KafkaBinderConfigurationProperties configurationProperties) { Map props = new HashMap<>(); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); @@ -111,17 +115,18 @@ public class KafkaBinderConfiguration { props.putAll(configurationProperties.getConsumerConfiguration()); } if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) { - props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString()); } ConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>(props); KafkaBinderHealthIndicator indicator = new KafkaBinderHealthIndicator(kafkaMessageChannelBinder, consumerFactory); - indicator.setTimeout(this.configurationProperties.getHealthTimeout()); + indicator.setTimeout(configurationProperties.getHealthTimeout()); return indicator; } @Bean - public MeterBinder kafkaBinderMetrics(KafkaMessageChannelBinder kafkaMessageChannelBinder) { + public MeterBinder kafkaBinderMetrics(KafkaMessageChannelBinder kafkaMessageChannelBinder, + KafkaBinderConfigurationProperties configurationProperties) { return new KafkaBinderMetrics(kafkaMessageChannelBinder, configurationProperties); }