From e4fed0a52d1202baa76d32d9f0f37ca78833e6b2 Mon Sep 17 00:00:00 2001 From: Massimiliano Poggi Date: Tue, 22 Oct 2019 17:58:00 +0200 Subject: [PATCH] Added qualifiers to CompositeMessageConverterBean injections When more than one CompositeMessageConverter bean was defined in the same ApplicationContext, not having the qualifiers on the injection points was causing the application to fail during instantiation due to bean conflicts being raised. The injection points for CompositeMessageConverter have been marked with the appropriate qualifier to inject the Spring Cloud CompositeMessageConverter. Resolves #775 --- ...fkaStreamsBinderSupportAutoConfiguration.java | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 27c1cc0a0..03d2fbb44 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -58,6 +58,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.core.env.ConfigurableEnvironment; import org.springframework.core.env.Environment; import org.springframework.core.env.MapPropertySource; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.kafka.config.KafkaStreamsConfiguration; import org.springframework.kafka.core.CleanupConfig; import org.springframework.kafka.streams.RecoveringDeserializationExceptionHandler; @@ -277,23 +278,26 @@ public class KafkaStreamsBinderSupportAutoConfiguration { @Bean public KafkaStreamsMessageConversionDelegate messageConversionDelegate( - CompositeMessageConverter compositeMessageConverter, - SendToDlqAndContinue sendToDlqAndContinue, - KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue, - KafkaStreamsBinderConfigurationProperties binderConfigurationProperties) { + @Qualifier(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME) + CompositeMessageConverter compositeMessageConverter, + SendToDlqAndContinue sendToDlqAndContinue, + KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue, + KafkaStreamsBinderConfigurationProperties binderConfigurationProperties) { return new KafkaStreamsMessageConversionDelegate(compositeMessageConverter, sendToDlqAndContinue, KafkaStreamsBindingInformationCatalogue, binderConfigurationProperties); } @Bean public MessageConverterDelegateSerde messageConverterDelegateSerde( - CompositeMessageConverter compositeMessageConverterFactory) { + @Qualifier(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME) + CompositeMessageConverter compositeMessageConverterFactory) { return new MessageConverterDelegateSerde(compositeMessageConverterFactory); } @Bean public CompositeNonNativeSerde compositeNonNativeSerde( - CompositeMessageConverter compositeMessageConverterFactory) { + @Qualifier(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME) + CompositeMessageConverter compositeMessageConverterFactory) { return new CompositeNonNativeSerde(compositeMessageConverterFactory); }