diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
index 1bfff8c76..e84acaea9 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
@@ -35,6 +35,7 @@ import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.converter.MessageConverterUtils;
+import org.springframework.cloud.stream.function.StreamFunctionProperties;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.support.MessageBuilderFactory;
@@ -79,10 +80,12 @@ public class MessageConverterConfigurer
 
 	private final Field headersField;
 
+	private final StreamFunctionProperties streamFunctionProperties;
+
 	private ConfigurableListableBeanFactory beanFactory;
 
 	public MessageConverterConfigurer(BindingServiceProperties bindingServiceProperties,
-			CompositeMessageConverter compositeMessageConverter) {
+			CompositeMessageConverter compositeMessageConverter, StreamFunctionProperties streamFunctionProperties) {
 		Assert.notNull(compositeMessageConverter,
 				"The message converter factory cannot be null");
 		this.bindingServiceProperties = bindingServiceProperties;
@@ -90,6 +93,12 @@ public class MessageConverterConfigurer
 
 		this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");
 		this.headersField.setAccessible(true);
+		this.streamFunctionProperties = streamFunctionProperties;
+	}
+
+	public MessageConverterConfigurer(BindingServiceProperties bindingServiceProperties,
+			CompositeMessageConverter compositeMessageConverter) {
+		this(bindingServiceProperties, compositeMessageConverter, null);
 	}
 
 	@Override
@@ -135,24 +144,26 @@ public class MessageConverterConfigurer
 				.getBindingProperties(channelName);
 		String contentType = bindingProperties.getContentType();
 		ProducerProperties producerProperties = bindingProperties.getProducer();
-		if (!inbound && producerProperties != null
-				&& producerProperties.isPartitioned()) {
+		boolean partitioned = !inbound && producerProperties != null && producerProperties.isPartitioned();
+		boolean functional = streamFunctionProperties != null && StringUtils.hasText(streamFunctionProperties.getDefinition());
+		if (partitioned) {
 			messageChannel.addInterceptor(new PartitioningInterceptor(bindingProperties,
 					getPartitionKeyExtractorStrategy(producerProperties),
 					getPartitionSelectorStrategy(producerProperties)));
 		}
 
 		ConsumerProperties consumerProperties = bindingProperties.getConsumer();
-		if (this.isNativeEncodingNotSet(producerProperties, consumerProperties,
-				inbound)) {
-			if (inbound) {
-				messageChannel.addInterceptor(
-						new InboundContentTypeEnhancingInterceptor(contentType));
-			}
-			else {
-				messageChannel.addInterceptor(
-						new OutboundContentTypeConvertingInterceptor(contentType,
-								this.compositeMessageConverter));
+		if (this.isNativeEncodingNotSet(producerProperties, consumerProperties, inbound)) {
+			if (partitioned || !functional) {
+				if (inbound) {
+					messageChannel.addInterceptor(
+							new InboundContentTypeEnhancingInterceptor(contentType));
+				}
+				else {
+					messageChannel.addInterceptor(
+							new OutboundContentTypeConvertingInterceptor(contentType,
+									this.compositeMessageConverter));
+				}
 			}
 		}
 	}
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java
index 844048ccb..a95e45621 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java
@@ -44,12 +44,12 @@ import org.springframework.cloud.stream.binding.MessageChannelConfigurer;
 import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
 import org.springframework.cloud.stream.binding.MessageSourceBindingTargetFactory;
 import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory;
+import org.springframework.cloud.stream.function.StreamFunctionProperties;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.context.annotation.Role;
-import org.springframework.core.env.Environment;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.UrlResource;
 import org.springframework.core.io.support.PropertiesLoaderUtils;
@@ -211,13 +211,9 @@ public class BinderFactoryAutoConfiguration {
 	public MessageConverterConfigurer messageConverterConfigurer(
 			BindingServiceProperties bindingServiceProperties,
 			@Qualifier(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME) CompositeMessageConverter compositeMessageConverter,
-			Environment environment, BinderTypeRegistry binderTypeRegistry) {
+			@Nullable StreamFunctionProperties streamFunctionProperties) {
 
-		if (binderTypeRegistry.getAll().keySet().contains("kstream") || !environment.containsProperty("spring.cloud.stream.function.definition")) {
-			return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverter);
-		}
-
-		return null;
+		return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverter, streamFunctionProperties);
 	}
 
 
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java
index 640a50641..9ac871bec 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java
@@ -70,7 +70,7 @@ public class GreenfieldFunctionEnableBindingTests {
 				TestChannelBinderConfiguration
 						.getCompleteConfiguration(SourceFromSupplier.class))
 								.web(WebApplicationType.NONE)
-								.run("--spring.cloud.stream.function.definition=date",
+								.run("--spring.cloud.function.definition=date",
 										"--spring.jmx.enabled=false")) {
 
 			OutputDestination target = context.getBean(OutputDestination.class);
@@ -93,7 +93,7 @@ public class GreenfieldFunctionEnableBindingTests {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
 				TestChannelBinderConfiguration.getCompleteConfiguration(
 						ProcessorFromFunction.class)).web(WebApplicationType.NONE).run(
-								"--spring.cloud.stream.function.definition=toUpperCase",
+								"--spring.cloud.function.definition=toUpperCase",
 								"--spring.jmx.enabled=false")) {
 
 			InputDestination source = context.getBean(InputDestination.class);
@@ -110,7 +110,7 @@ public class GreenfieldFunctionEnableBindingTests {
 				TestChannelBinderConfiguration
 						.getCompleteConfiguration(SinkFromConsumer.class))
 								.web(WebApplicationType.NONE)
-								.run("--spring.cloud.stream.function.definition=sink",
+								.run("--spring.cloud.function.definition=sink",
 										"--spring.jmx.enabled=false")) {
 
 			InputDestination source = context.getBean(InputDestination.class);
@@ -126,7 +126,7 @@ public class GreenfieldFunctionEnableBindingTests {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
 				TestChannelBinderConfiguration.getCompleteConfiguration(
 						HttpInboundEndpoint.class)).web(WebApplicationType.SERVLET).run(
-								"--spring.cloud.stream.function.definition=upperCase",
+								"--spring.cloud.function.definition=upperCase",
 								"--spring.jmx.enabled=false", "--server.port=0")) {
 			TestRestTemplate restTemplate = new TestRestTemplate();
 			restTemplate.postForLocation(
@@ -146,7 +146,7 @@ public class GreenfieldFunctionEnableBindingTests {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
 				TestChannelBinderConfiguration.getCompleteConfiguration(
 						FooTransform.class)).web(WebApplicationType.NONE).run(
-								"--spring.cloud.stream.function.definition=fooFunction",
+								"--spring.cloud.function.definition=fooFunction",
 								"--spring.jmx" + ".enabled=false",
 								"--logging.level.org.springframework.integration=TRACE")) {
 			MessageChannel input = context.getBean("input", MessageChannel.class);