INT-4397: Fix headers filtering for @transformer

JIRA: https://jira.spring.io/browse/INT-4397

The `AbstractMessageProcessingTransformer` doesn't honor a configured
`notPropagatedHeaders` and copies all the request headers to the
message to return

* Add `setNotPropagatedHeaders()` into the `AbstractMessageProcessingTransformer`
and implement there a logic to filter headers, similar to what we have
in the `AbstractMessageProducingHandler`
* Overrider `updateNotPropagatedHeaders()` in the `MessageTransformingHandler`
to propagate `notPropagatedHeaders` to the `AbstractMessageProcessingTransformer`
delegate

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java

@@ -139,7 +139,7 @@ public void setNotPropagatedHeaders(String... headers) {
 	 * @param merge true to merge with existing patterns; false to replace.
 	 * @since 5.0.2
 	 */
-	protected final void updateNotPropagatedHeaders(String[] headers, boolean merge) {
+	protected void updateNotPropagatedHeaders(String[] headers, boolean merge) {
 		Set<String> headerPatterns = new HashSet<>();
 
 		if (merge && this.notPropagatedHeaders != null) {
@@ -151,7 +151,7 @@ protected final void updateNotPropagatedHeaders(String[] headers, boolean merge)
 
 			headerPatterns.addAll(Arrays.asList(headers));
 
-			this.notPropagatedHeaders = headerPatterns.toArray(new String[headerPatterns.size()]);
+			this.notPropagatedHeaders = headerPatterns.toArray(new String[0]);
 		}
 
 		boolean hasAsterisk = headerPatterns.contains("*");

spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractMessageProcessingTransformer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,15 +16,23 @@
 
 package org.springframework.integration.transformer;
 
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.handler.MessageProcessor;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.DefaultMessageBuilderFactory;
 import org.springframework.integration.support.MessageBuilderFactory;
 import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.PatternMatchUtils;
 
 /**
  * Base class for Message Transformers that delegate to a {@link MessageProcessor}.
@@ -37,11 +45,15 @@ public abstract class AbstractMessageProcessingTransformer
 
 	private final MessageProcessor<?> messageProcessor;
 
-	private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
+	private BeanFactory beanFactory;
 
-	private volatile boolean messageBuilderFactorySet;
+	private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
 
-	private BeanFactory beanFactory;
+	private boolean messageBuilderFactorySet;
+
+	private String[] notPropagatedHeaders;
+
+	private boolean selectiveHeaderPropagation;
 
 	protected AbstractMessageProcessingTransformer(MessageProcessor<?> messageProcessor) {
 		Assert.notNull(messageProcessor, "messageProcessor must not be null");
@@ -85,6 +97,22 @@ public boolean isRunning() {
 		return !(this.messageProcessor instanceof Lifecycle) || ((Lifecycle) this.messageProcessor).isRunning();
 	}
 
+	/**
+	 * Set headers that will NOT be copied from the inbound message if
+	 * the handler is configured to copy headers.
+	 * @param headers the headers to not propagate from the inbound message.
+	 * @since 5.1
+	 */
+	public void setNotPropagatedHeaders(String... headers) {
+		if (!ObjectUtils.isEmpty(headers)) {
+			Assert.noNullElements(headers, "null elements are not allowed in 'headers'");
+
+			this.notPropagatedHeaders = Arrays.copyOf(headers, headers.length);
+		}
+
+		this.selectiveHeaderPropagation = !ObjectUtils.isEmpty(this.notPropagatedHeaders);
+	}
+
 	@Override
 	public final Message<?> transform(Message<?> message) {
 		Object result = this.messageProcessor.processMessage(message);
@@ -94,7 +122,26 @@ public final Message<?> transform(Message<?> message) {
 		if (result instanceof Message<?>) {
 			return (Message<?>) result;
 		}
-		return getMessageBuilderFactory().withPayload(result).copyHeaders(message.getHeaders()).build();
+
+		MessageHeaders requestHeaders = message.getHeaders();
+
+		AbstractIntegrationMessageBuilder<?> messageBuilder =
+				getMessageBuilderFactory()
+						.withPayload(result);
+
+		if (this.selectiveHeaderPropagation) {
+			Map<String, Object> headersToCopy = new HashMap<>(requestHeaders);
+
+			headersToCopy.entrySet()
+					.removeIf(entry -> PatternMatchUtils.simpleMatch(this.notPropagatedHeaders, entry.getKey()));
+
+			messageBuilder.copyHeadersIfAbsent(headersToCopy);
+		}
+		else {
+			messageBuilder.copyHeadersIfAbsent(requestHeaders);
+		}
+
+		return messageBuilder.build();
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.transformer;
 
+import java.util.Collection;
+
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
@@ -64,6 +66,18 @@ protected void doInit() {
 		}
 	}
 
+	@Override
+	protected void updateNotPropagatedHeaders(String[] headers, boolean merge) {
+		super.updateNotPropagatedHeaders(headers, merge);
+
+		Collection<String> notPropagatedHeaders = getNotPropagatedHeaders();
+
+		if (this.transformer instanceof AbstractMessageProcessingTransformer && !notPropagatedHeaders.isEmpty()) {
+			((AbstractMessageProcessingTransformer) this.transformer)
+					.setNotPropagatedHeaders(notPropagatedHeaders.toArray(new String[0]));
+		}
+	}
+
 	@Override
 	public void start() {
 		if (this.transformer instanceof Lifecycle) {

spring-integration-core/src/test/java/org/springframework/integration/handler/HeaderAnnotationTransformerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 package org.springframework.integration.handler;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.mockito.Mockito.mock;
 
@@ -34,6 +35,8 @@
 /**
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class HeaderAnnotationTransformerTests {
@@ -103,6 +106,27 @@ public void headerAnnotationWithUnprefixedHeaderAndRelativeExpression() {
 	}
 
 
+	@Test
+	public void testNotPropagatedHeaders() {
+		Object target = new TestTransformer();
+		MethodInvokingTransformer transformer = new MethodInvokingTransformer(target, "evalFoo");
+		MessageTransformingHandler handler = new MessageTransformingHandler(transformer);
+		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.setNotPropagatedHeaders(IntegrationMessageHeaderAccessor.CORRELATION_ID);
+		handler.afterPropertiesSet();
+		QueueChannel outputChannel = new QueueChannel();
+		handler.setOutputChannel(outputChannel);
+		handler.handleMessage(
+				MessageBuilder.withPayload("test")
+						.setCorrelationId("abc")
+						.setHeader("foo", "bar")
+						.build());
+		Message<?> result = outputChannel.receive(0);
+		assertNotNull(result);
+		assertEquals("BAR", result.getPayload());
+		assertFalse(result.getHeaders().containsKey(IntegrationMessageHeaderAccessor.CORRELATION_ID));
+	}
+
 	public static class TestTransformer {
 
 		public String appendCorrelationId(Object payload,