GH-1967 Fixed support for converting collection types

The main fix is actually in spring-cloud-function 417e54e0af
The changes here are mainly polishing and adjusting around function changes

Resolves #1967

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

@@ -23,6 +23,7 @@ import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.AbstractMessageConverter;
+import org.springframework.util.Assert;
 import org.springframework.util.MimeType;
 
 /**
@@ -43,6 +44,7 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter {
 		setStrictContentTypeMatch(true);
 	}
 
+	@Override
 	protected boolean supports(Class<?> clazz) {
 		return true;
 	}
@@ -67,8 +69,9 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter {
 		return super.supportsMimeType(headers);
 	}
 
-	protected Object convertFromInternal(Message<?> message, Class<?> targetClass,
-			Object conversionHint) {
+	@Override
+	protected Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint) {
+		Assert.isTrue(String.class.isAssignableFrom(targetClass) || targetClass == Object.class, "This converter can only convert byte[] to String");
 		if (message.getPayload() != null) {
 			if (message.getPayload() instanceof byte[]) {
 				if (byte[].class.isAssignableFrom(targetClass)) {
@@ -92,6 +95,7 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter {
 		return null;
 	}
 
+	@Override
 	protected Object convertToInternal(Object payload, MessageHeaders headers,
 			Object conversionHint) {
 		if (payload != null) {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java

@@ -523,7 +523,7 @@ public class FunctionConfiguration {
 	 * not attempt any conversion and sends a raw Message.
 	 */
 	@SuppressWarnings("rawtypes")
-	private static class FunctionWrapper implements Function<Message<byte[]>, Object> {
+	private static class FunctionWrapper implements Function<Message, Object> {
 		private final Function function;
 
 		private final ConsumerProperties consumerProperties;
@@ -551,7 +551,7 @@ public class FunctionConfiguration {
 
 		@SuppressWarnings("unchecked")
 		@Override
-		public Object apply(Message<byte[]> message) {
+		public Object apply(Message message) {
 			if (message != null && consumerProperties != null) {
 				Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
 						.getField(this.headersField, message.getHeaders());

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

@@ -20,6 +20,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Function;
 
+import org.assertj.core.util.Arrays;
 import org.junit.Test;
 
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -94,6 +95,48 @@ public class FunctionBatchingTests {
 				});
 	}
 
+	@Test
+	public void testListStringPayloadConfigurationTextPlain() {
+		TestChannelBinderConfiguration.applicationContextRunner(ListStringPayloadConfiguration.class)
+				.withPropertyValues("spring.jmx.enabled=false",
+						"spring.cloud.stream.function.definition=func",
+						"spring.cloud.stream.bindings.func-in-0.content-type=text/plain")
+				.run(context -> {
+					InputDestination inputDestination = context.getBean(InputDestination.class);
+					OutputDestination outputDestination = context
+							.getBean(OutputDestination.class);
+
+					List bytes = Arrays.asList(new Object[] {"abc".getBytes(), "xyz".getBytes()});
+					Message inputMessage = MessageBuilder.withPayload(bytes).build();
+					inputDestination.send(inputMessage);
+
+					Message<byte[]> outputMessage = outputDestination.receive();
+					assertThat(new String(outputMessage.getPayload())).isEqualTo("[abc, xyz]");
+					context.stop();
+				});
+	}
+
+	@Test
+	public void testListObjectPayloadObjectConfigurationTextPlain() {
+		TestChannelBinderConfiguration.applicationContextRunner(ListObjectPayloadConfiguration.class)
+				.withPropertyValues("spring.jmx.enabled=false",
+						"spring.cloud.stream.function.definition=func",
+						"spring.cloud.stream.bindings.func-in-0.content-type=text/plain")
+				.run(context -> {
+					InputDestination inputDestination = context.getBean(InputDestination.class);
+					OutputDestination outputDestination = context
+							.getBean(OutputDestination.class);
+
+					List bytes = Arrays.asList(new Object[] {"abc".getBytes(), "xyz".getBytes()});
+					Message inputMessage = MessageBuilder.withPayload(bytes).build();
+					inputDestination.send(inputMessage);
+
+					Message<byte[]> outputMessage = outputDestination.receive();
+					assertThat(new String(outputMessage.getPayload())).isEqualTo("[abc, xyz]");
+					context.stop();
+				});
+	}
+
 	@Test
 	public void testSimpleBatchConfiguration() {
 		TestChannelBinderConfiguration.applicationContextRunner(SimpleBatchConfiguration.class)
@@ -172,6 +215,21 @@ public class FunctionBatchingTests {
 
 	}
 
+	@EnableAutoConfiguration
+	public static class ListStringPayloadConfiguration {
+		@Bean
+		public Function<List<String>, String> func() {
+			return x -> x.toString();
+		}
+	}
+
+	@EnableAutoConfiguration
+	public static class ListObjectPayloadConfiguration {
+		@Bean
+		public Function<List<Object>, String> func() {
+			return x -> x.toString();
+		}
+	}
 
 	@EnableAutoConfiguration
 	public static class ListPayloadNotBatchConfiguration {
@@ -192,9 +250,7 @@ public class FunctionBatchingTests {
 			public void setName(String name) {
 				this.name = name;
 			}
-
 		}
-
 	}
 
 	@EnableAutoConfiguration