Fix issue with multiple ObjectMapper beans

Author: olegz

core/spring-cloud-stream-integration-tests/pom.xml

@@ -25,6 +25,11 @@
 			<artifactId>spring-cloud-stream-test-binder</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>tools.jackson.dataformat</groupId>
+			<artifactId>jackson-dataformat-cbor</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
@@ -43,16 +48,19 @@
 		<dependency>
 			<groupId>org.jetbrains.kotlin</groupId>
 			<artifactId>kotlin-test</artifactId>
+			<version>2.2.20</version>
 			<scope>test</scope>
 		</dependency>
 		<dependency>
 			<groupId>org.jetbrains.kotlin</groupId>
 			<artifactId>kotlin-stdlib-jdk8</artifactId>
+			<version>2.2.20</version>
 			<optional>true</optional>
 		</dependency>
 		<dependency>
 			<groupId>org.jetbrains.kotlin</groupId>
 			<artifactId>kotlin-reflect</artifactId>
+			<version>2.2.20</version>
 			<optional>true</optional>
 		</dependency>
 		<dependency>

core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/ScenarioTests.java

@@ -33,6 +33,8 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
+import tools.jackson.databind.ObjectMapper;
+import tools.jackson.dataformat.cbor.CBORMapper;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -131,6 +133,16 @@ void test2113() {
 
 
 
+	@Test
+	void testMultipleObjectMappers() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+			TestChannelBinderConfiguration.getCompleteConfiguration(MultipleJsonConverterConfiguration.class))
+			.web(WebApplicationType.NONE)
+			.run("--spring.jmx.enabled=false")) {
+		}
+	}
+
+
 	@EnableAutoConfiguration
 	@Configuration
 	public static class TestConfiguration {
@@ -143,6 +155,19 @@ public <I, O> Function<I, O> genericTypeFunction() {
 		}
 	}
 
+	@EnableAutoConfiguration
+	@Configuration
+	public static class MultipleJsonConverterConfiguration {
+		@Bean
+		public ObjectMapper cborObjectMapper() {
+			return new CBORMapper();
+		}
+		@Bean
+		public ObjectMapper objectMapper() {
+			return new ObjectMapper();
+		}
+	}
+
 	@EnableAutoConfiguration
 	@Configuration
 	public static class SupplierConfiguration {

core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java

@@ -381,7 +381,7 @@ private MessageConverterConfigurer createConverterConfigurer(String channelName,
 		bindingServiceProperties.afterPropertiesSet();
 		MessageConverterConfigurer messageConverterConfigurer = new MessageConverterConfigurer(
 				bindingServiceProperties,
-				new CompositeMessageConverterFactory(null, null, null).getMessageConverterForAllRegistered());
+				new CompositeMessageConverterFactory(null, null).getMessageConverterForAllRegistered());
 		messageConverterConfigurer.setBeanFactory(applicationContext.getBeanFactory());
 		return messageConverterConfigurer;
 	}

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java

@@ -23,6 +23,7 @@
 import java.util.function.Consumer;
 
 import org.apache.commons.logging.Log;
+import org.springframework.cloud.function.json.JacksonMapper;
 import tools.jackson.core.JacksonException;
 import tools.jackson.core.JsonGenerator;
 import tools.jackson.databind.ObjectMapper;
@@ -141,12 +142,19 @@ public AbstractMessageChannelBinder(String[] headersToEmbed,
 
 	@Override
 	protected void onInit() throws Exception {
-		if (!CollectionUtils.isEmpty(this.getApplicationContext().getBeansOfType(ObjectMapper.class))) {
-			this.objectMapper = this.getApplicationContext().getBean(ObjectMapper.class);
+		if (!CollectionUtils.isEmpty(this.getApplicationContext().getBeansOfType(JacksonMapper.class))) {
+			JacksonMapper jacksonMapper = this.getApplicationContext().getBean(JacksonMapper.class);
+			this.objectMapper = jacksonMapper.getObjectMapper();
 		}
 		else {
 			this.objectMapper = new ObjectMapper();
 		}
+//		if (!CollectionUtils.isEmpty(this.getApplicationContext().getBeansOfType(ObjectMapper.class))) {
+//			this.objectMapper = this.getApplicationContext().getBean(ObjectMapper.class);
+//		}
+//		else {
+//
+//		}
 		SimpleModule module = new SimpleModule();
 		module.addSerializer(Expression.class, new ExpressionSerializer(Expression.class));
 		this.objectMapper = this.objectMapper.rebuild().build();

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ContentTypeConfiguration.java

@@ -51,7 +51,7 @@ public CompositeMessageConverter configurableCompositeMessageConverter(
 				.filter(c -> isConverterEligible(c)).collect(Collectors.toList());
 
 		CompositeMessageConverterFactory factory =
-				new CompositeMessageConverterFactory(customMessageConverters, objectMapperObjectProvider.getIfAvailable(ObjectMapper::new), jsonMapper);
+				new CompositeMessageConverterFactory(customMessageConverters, jsonMapper);
 
 		return factory.getMessageConverterForAllRegistered();
 	}

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java

@@ -54,25 +54,22 @@ public class CompositeMessageConverterFactory {
 
 	private final Log log = LogFactory.getLog(CompositeMessageConverterFactory.class);
 
-	private final ObjectMapper objectMapper;
+	//private final ObjectMapper objectMapper;
 
 	private final List<MessageConverter> converters;
 
 	private final JsonMapper jsonMapper;
 
 	public CompositeMessageConverterFactory() {
-		this(Collections.<MessageConverter>emptyList(), new ObjectMapper(), null);
+		this(Collections.<MessageConverter>emptyList(), null);
 	}
 
 	/**
 	 * @param customConverters a list of {@link AbstractMessageConverter}
-	 * @param objectMapper object mapper for for serialization / deserialization
 	 */
 	public CompositeMessageConverterFactory(
-			List<? extends MessageConverter> customConverters,
-			ObjectMapper objectMapper, JsonMapper jsonMapper) {
-		this.objectMapper = objectMapper == null ? new ObjectMapper() : objectMapper;
-		this.jsonMapper = jsonMapper == null ? new JacksonMapper(this.objectMapper) : jsonMapper;
+			List<? extends MessageConverter> customConverters, JsonMapper jsonMapper) {
+		this.jsonMapper = jsonMapper == null ? new JacksonMapper(new ObjectMapper()) : jsonMapper;
 		if (!CollectionUtils.isEmpty(customConverters)) {
 			this.converters = new ArrayList<>(customConverters);
 		}

core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java

@@ -50,7 +50,7 @@ void configureOutputChannelWithBadContentType() {
 		bindingProps.setContentType("application/json");
 		props.setBindings(Collections.singletonMap("foo", bindingProps));
 		CompositeMessageConverterFactory converterFactory = new CompositeMessageConverterFactory(
-				Collections.<MessageConverter>emptyList(), null, null);
+				Collections.<MessageConverter>emptyList(), null);
 		MessageConverterConfigurer configurer = new MessageConverterConfigurer(props,
 				converterFactory.getMessageConverterForAllRegistered());
 		QueueChannel out = new QueueChannel();
@@ -82,7 +82,7 @@ protected Object convertToInternal(Object payload, MessageHeaders headers,
 
 		};
 		CompositeMessageConverterFactory converterFactory = new CompositeMessageConverterFactory(
-				Collections.<MessageConverter>singletonList(converter), null, null);
+				Collections.<MessageConverter>singletonList(converter), null);
 		MessageConverterConfigurer configurer = new MessageConverterConfigurer(props,
 				converterFactory.getMessageConverterForAllRegistered());
 		QueueChannel out = new QueueChannel();