GH-1898 Fix map/collection tyoe conversion

Delegated conversion of individual collection elements that are  non-String and non-byte[] to ObjectMapper
In the future we can potentially explore delegation to Spring's ConversionService

Resolves #1898

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -100,7 +100,6 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
 		else if (conversionHint instanceof ParameterizedType) {
 			result = convertParameterizedType(message, (Type) conversionHint);
 		}
-
 		if (result == null) {
 			if (message.getPayload() instanceof byte[]
 					&& targetClass.isAssignableFrom(String.class)) {
@@ -146,6 +145,11 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
 									else if (value instanceof String) {
 										return objectMapper.readValue((String) value, typeToUse.getContentType());
 									}
+									else {
+										// fall back to simple type-conversion
+										// see https://github.com/spring-cloud/spring-cloud-stream/issues/1898
+										return objectMapper.convertValue(value, typeToUse.getContentType());
+									}
 								}
 								catch (Exception e) {
 									logger.error("Failed to convert payload " + value, e);

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java

@@ -18,6 +18,8 @@ package org.springframework.cloud.stream.function;
 
 import java.io.Serializable;
 import java.time.Duration;
+import java.util.List;
+import java.util.Map;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -372,6 +374,26 @@ public class ImplicitFunctionBindingTests {
 		}
 	}
 
+	@Test
+	public void testCollectionAndMapConversionDuringComposition() {
+		System.clearProperty("spring.cloud.function.definition");
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+				.getCompleteConfiguration(CompositionWithCollectionConfiguration.class))
+						.web(WebApplicationType.NONE).run("--spring.cloud.function.definition=funcA|funcB",
+								"--spring.jmx.enabled=false")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder.withPayload("[{\"key1\":1, \"key2\":2},{\"key3\":3}]".getBytes()).build();
+
+			inputDestination.send(inputMessage);
+			String result = new String(outputDestination.receive().getPayload());
+
+			assertThat(result).isEqualTo("[{\"key1\":\"1\",\"key2\":\"2\"},{\"key3\":\"3\"}]");
+		}
+	}
+
 	@EnableAutoConfiguration
 	public static class NoEnableBindingConfiguration {
 
@@ -514,4 +536,22 @@ public class ImplicitFunctionBindingTests {
 		}
 	}
 
+	@EnableAutoConfiguration
+	public static class CompositionWithCollectionConfiguration {
+
+		@Bean
+		public Function<Message<List<Map<String, Integer>>>, Message<List<Map<String, Integer>>>> funcA() {
+			return v -> {
+				return v;
+			};
+		}
+
+		@Bean
+		public Function<Message<List<Map<String, String>>>, Message<List<Map<String, String>>>> funcB() {
+			return v -> {
+				return v;
+			};
+		}
+	}
+
 }