diff --git a/pom.xml b/pom.xml
index 86e7a10f2..749862dc5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -25,7 +25,7 @@
 		<java.version>1.8</java.version>
 		<reactor.version>Dysprosium-SR6</reactor.version>
 		<objenesis.version>2.1</objenesis.version>
-		<spring-cloud-function.version>3.0.10.RELEASE</spring-cloud-function.version>
+		<spring-cloud-function.version>3.0.11.BUILD-SNAPSHOT</spring-cloud-function.version>
 		<maven-checkstyle-plugin.failsOnError>true</maven-checkstyle-plugin.failsOnError>
 		<maven-checkstyle-plugin.failsOnViolation>true</maven-checkstyle-plugin.failsOnViolation>
 		<maven-checkstyle-plugin.includeTestSourceDirectory>true</maven-checkstyle-plugin.includeTestSourceDirectory>
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java
index 76b61d4fd..2f7d4e86b 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java
@@ -18,12 +18,13 @@ package org.springframework.cloud.stream.converter;
 
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
+import java.util.Collection;
+import java.util.stream.Collectors;
 
 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;
 
 /**
@@ -71,7 +72,6 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter {
 
 	@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)) {
@@ -82,6 +82,22 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter {
 							StandardCharsets.UTF_8);
 				}
 			}
+			else if (message.getPayload() instanceof Collection) {
+				Collection<?> collection = ((Collection<?>) message.getPayload()).stream()
+						.map(value -> {
+							if (byte[].class.isAssignableFrom(targetClass)) {
+								return value;
+							}
+							else if (value instanceof byte[]) {
+								return new String((byte[]) value, StandardCharsets.UTF_8);
+							}
+							else {
+								return value; // String
+							}
+						}).collect(Collectors.toList());
+
+				return collection;
+			}
 			else {
 				if (byte[].class.isAssignableFrom(targetClass)) {
 					return message.getPayload().toString()
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
index b89d47853..df913d4ad 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
@@ -423,6 +423,7 @@ public class FunctionConfiguration {
 					BindingProperties bindingProperties = this.serviceProperties.getBindings().get(inputBindingName);
 					ConsumerProperties consumerProperties = bindingProperties == null ? null : bindingProperties.getConsumer();
 					if (consumerProperties != null) {
+						function.setSkipInputConversion(consumerProperties.isUseNativeDecoding());
 						Assert.isTrue(consumerProperties.getConcurrency() <= 1, "Concurrency > 1 is not supported by reactive "
 								+ "consumer, given that project reactor maintains its own concurrency mechanism. Was '..."
 								+ inputBindingName + ".consumer.concurrency=" + consumerProperties.getConcurrency() + "'");
@@ -435,6 +436,9 @@ public class FunctionConfiguration {
 				if (!CollectionUtils.isEmpty(outputBindingNames)) {
 					BindingProperties bindingProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next());
 					ProducerProperties producerProperties = bindingProperties == null ? null : bindingProperties.getProducer();
+					if (producerProperties != null) {
+						function.setSkipOutputConversion(producerProperties.isUseNativeEncoding());
+					}
 					functionToInvoke = new PartitionAwareFunctionWrapper(function, this.applicationContext, producerProperties);
 				}
 
@@ -457,9 +461,15 @@ public class FunctionConfiguration {
 								if (logger.isInfoEnabled()) {
 									logger.info("Output message is sent to '" + destinationName + "' destination");
 								}
+								if (!(message instanceof Message)) {
+									message = MessageBuilder.withPayload(message).build();
+								}
 								dynamicChannel.send((Message) message);
 							}
 							else {
+								if (!(message instanceof Message)) {
+									message = MessageBuilder.withPayload(message).build();
+								}
 								outputChannel.send((Message) message);
 							}
 						});
@@ -471,9 +481,6 @@ public class FunctionConfiguration {
 			}
 			else {
 				String outputDestinationName = this.determineOutputDestinationName(0, bindableProxyFactory, functionType);
-				if (StringUtils.hasText(outputDestinationName)) {
-					this.adjustFunctionForNativeEncodingIfNecessary(outputDestinationName, function, 0);
-				}
 				String inputDestinationName = inputBindingNames.iterator().next();
 				Object inputDestination = this.applicationContext.getBean(inputDestinationName);
 				if (inputDestination != null && inputDestination instanceof SubscribableChannel) {
@@ -486,25 +493,6 @@ public class FunctionConfiguration {
 			}
 		}
 
-		private void adjustFunctionForNativeEncodingIfNecessary(String outputDestinationName, FunctionInvocationWrapper function, int index) {
-			if (function.isConsumer()) {
-				return;
-			}
-			BindingProperties properties = this.serviceProperties.getBindingProperties(outputDestinationName);
-			if (properties.getProducer() != null && properties.getProducer().isUseNativeEncoding()) {
-				Field acceptedOutputMimeTypesField = ReflectionUtils
-						.findField(FunctionInvocationWrapper.class, "acceptedOutputMimeTypes", String[].class);
-				acceptedOutputMimeTypesField.setAccessible(true);
-				try {
-					String[] acceptedOutputMimeTypes = (String[]) acceptedOutputMimeTypesField.get(function);
-					acceptedOutputMimeTypes[index] = "";
-				}
-				catch (Exception e) {
-					// ignore
-				}
-			}
-		}
-
 		private ServiceActivatingHandler createFunctionHandler(FunctionInvocationWrapper function,
 				String inputChannelName, String outputChannelName) {
 			ConsumerProperties consumerProperties = StringUtils.hasText(inputChannelName)
@@ -621,7 +609,7 @@ public class FunctionConfiguration {
 	 * not attempt any conversion and sends a raw Message.
 	 */
 	@SuppressWarnings("rawtypes")
-	private static class FunctionWrapper implements Function<Message, Object> {
+	private static class FunctionWrapper implements Function<Message<byte[]>, Object> {
 		private final Function function;
 
 		private final ConsumerProperties consumerProperties;
@@ -642,14 +630,20 @@ public class FunctionConfiguration {
 			this.applicationContext = applicationContext;
 			this.function = new PartitionAwareFunctionWrapper((FunctionInvocationWrapper) function, this.applicationContext, producerProperties);
 			this.consumerProperties = consumerProperties;
+			if (this.consumerProperties != null) {
+				((FunctionInvocationWrapper) function).setSkipInputConversion(this.consumerProperties.isUseNativeDecoding());
+			}
 			this.producerProperties = producerProperties;
+			if (this.producerProperties != null) {
+				((FunctionInvocationWrapper) function).setSkipOutputConversion(this.producerProperties.isUseNativeEncoding());
+			}
 			this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");
 			this.headersField.setAccessible(true);
 		}
 
 		@SuppressWarnings("unchecked")
 		@Override
-		public Object apply(Message message) {
+		public Object apply(Message<byte[]> message) {
 			if (message != null && consumerProperties != null) {
 				Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
 						.getField(this.headersField, message.getHeaders());
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
index 91046aad7..858003f9d 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
@@ -869,17 +869,77 @@ public class ImplicitFunctionBindingTests {
 		}
 	}
 
+	@SuppressWarnings("rawtypes")
 	@Test
-	public void foo() {
-		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
-				.getCompleteConfiguration(PojoFunctionConfiguration.class))
+	public void testWithNativeEncodingImperative() {
+		System.clearProperty("spring.cloud.function.definition");
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(SingleFunctionConfiguration2.class))
 						.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false",
-								"--spring.cloud.function.definition=f1|f2")) {
+								"--spring.cloud.function.definition=imperative")) {
 
 			InputDestination inputDestination = context.getBean(InputDestination.class);
+			inputDestination.send(new GenericMessage<byte[]>("hello".getBytes()));
+
 			OutputDestination outputDestination = context.getBean(OutputDestination.class);
-			inputDestination.send(MessageBuilder.withPayload("hello".getBytes()).build());
-			assertThat(outputDestination.receive(1000)).isNotNull();
+
+			Message result = outputDestination.receive(2000);
+			assertThat(result.getPayload()).isInstanceOf(byte[].class); // check output type
+			assertThat(new String((byte[]) result.getPayload())).isEqualTo("String"); // check input type
+		}
+
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(SingleFunctionConfiguration2.class))
+						.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false",
+								"--spring.cloud.function.definition=imperative",
+								"--spring.cloud.stream.bindings.imperative-in-0.consumer.useNativeDecoding=true",
+								"--spring.cloud.stream.bindings.imperative-out-0.producer.useNativeEncoding=true"
+								)) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			inputDestination.send(new GenericMessage<byte[]>("hello".getBytes()));
+
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+			Message result = outputDestination.receive(2000);
+			assertThat(result.getPayload()).isInstanceOf(String.class); // check output type
+			assertThat(result.getPayload()).isEqualTo("byte[]"); // check input type
+		}
+	}
+
+	@SuppressWarnings("rawtypes")
+	@Test
+	public void testWithNativeEncodingReactive() {
+		System.clearProperty("spring.cloud.function.definition");
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(SingleFunctionConfiguration2.class))
+						.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false",
+								"--spring.cloud.function.definition=reactive")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			inputDestination.send(new GenericMessage<byte[]>("hello".getBytes()));
+
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+			Message result = outputDestination.receive(2000);
+			assertThat(result.getPayload()).isInstanceOf(byte[].class); // check output type
+			assertThat(new String((byte[]) result.getPayload())).isEqualTo("String"); // check input type
+		}
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(SingleFunctionConfiguration2.class))
+						.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false",
+								"--spring.cloud.function.definition=reactive",
+								"--spring.cloud.stream.bindings.reactive-in-0.consumer.useNativeDecoding=true",
+								"--spring.cloud.stream.bindings.reactive-out-0.producer.useNativeEncoding=true")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			inputDestination.send(new GenericMessage<byte[]>("hello".getBytes()));
+
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+			Message result = outputDestination.receive(2000);
+			assertThat(result.getPayload()).isInstanceOf(String.class); // no output conversion to byte[] has happened.
+			assertThat(result.getPayload()).isEqualTo("byte[]");
 		}
 	}
 
@@ -954,6 +1014,24 @@ public class ImplicitFunctionBindingTests {
 		}
 	}
 
+	@EnableAutoConfiguration
+	public static class SingleFunctionConfiguration2 {
+
+		@Bean
+		public Function<Object, String> imperative() {
+			return x -> {
+				return x.getClass().getSimpleName();
+			};
+		}
+
+		@Bean
+		public Function<Flux<Object>, Flux<String>> reactive() {
+			return flux -> flux.map(x -> {
+				return x.getClass().getSimpleName();
+			});
+		}
+	}
+
 	@EnableAutoConfiguration
 	public static class SingleFunctionConfiguration {