GH-2115 Ensure outbound interceptor is not used for functional approach

Resolves #2115

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java

@@ -155,13 +155,10 @@ public class MessageConverterConfigurer
 		ConsumerProperties consumerProperties = bindingProperties.getConsumer();
 		if (this.isNativeEncodingNotSet(producerProperties, consumerProperties, inbound)) {
 			if (inbound) {
-				messageChannel.addInterceptor(
-						new InboundContentTypeEnhancingInterceptor(contentType));
+				messageChannel.addInterceptor(new InboundContentTypeEnhancingInterceptor(contentType));
 			}
-			else {
-				messageChannel.addInterceptor(
-						new OutboundContentTypeConvertingInterceptor(contentType,
-								this.compositeMessageConverter));
+			else if (!functional) {
+				messageChannel.addInterceptor(new OutboundContentTypeConvertingInterceptor(contentType, this.compositeMessageConverter));
 			}
 		}
 	}

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

@@ -64,7 +64,8 @@ public class DynamicDestinationFunctionTests {
 							serviceProperties.getProducerProperties("fooDestination").getPartitionKeyExtractorName());
 
 					OutputDestination output = context.getBean(OutputDestination.class);
-					assertThat(output.receive(1000).getPayload()).isEqualTo("fooDestination".getBytes());
+					Object result = output.receive(1000).getPayload();
+					assertThat(result).isEqualTo("fooDestination");
 				});
 	}
 
@@ -88,7 +89,6 @@ public class DynamicDestinationFunctionTests {
 		@Bean
 		public Consumer<String> cons() {
 			return value -> {
-				System.out.println("Sending to " + value);
 				resolver.resolveDestination(value).send(new GenericMessage<String>(value));
 			};
 		}