GH-2179 provide mechanism to remove short-lived headers

Headers such as 'spring.cloud.stream.sendto.destination' are no longer valid once the message has been sent, os it will be removed

Resolves #2179

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

@@ -352,6 +352,11 @@ public class FunctionConfiguration {
 						: MessageBuilder.withPayload(value).build();
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	private static Message sanitize(Message inputMessage) {
+		return MessageBuilder.fromMessage(inputMessage).removeHeader("spring.cloud.stream.sendto.destination").build();
+	}
+
 	private static class FunctionToDestinationBinder implements InitializingBean, ApplicationContextAware {
 
 		protected final Log logger = LogFactory.getLog(getClass());
@@ -466,7 +471,12 @@ public class FunctionConfiguration {
 								+ inputBindingName + ".consumer.concurrency=" + consumerProperties.getConcurrency() + "'");
 					}
 					SubscribableChannel inputChannel = this.applicationContext.getBean(inputBindingName, SubscribableChannel.class);
-					return IntegrationReactiveUtils.messageChannelToFlux(inputChannel);
+					return IntegrationReactiveUtils.messageChannelToFlux(inputChannel).map(m -> {
+						if (m instanceof Message) {
+							m = sanitize(m);
+						}
+						return m;
+					});
 				})
 				.map(publisher -> {
 					if (targetProtocolEnhancer.get() != null) {
@@ -713,6 +723,7 @@ public class FunctionConfiguration {
 		@SuppressWarnings("unchecked")
 		@Override
 		public Object apply(Message<byte[]> message) {
+			message = sanitize(message);
 			Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
 					.getField(this.headersField, message.getHeaders());
 			if (StringUtils.hasText(targetProtocol)) {