INT-4288: Add blacklist headers at runtime

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

- Add `AbstractMessageProducingHandler.addNotPropagatedHeaders()`
so that new headers can be added without overriding existing ones.
- Add `AbstractMessageProducingHandler.getNotPropagatedHeaders()`
so that existing excluded headers can be consulted.

* Polishing JavaDocs and imports order

**Cherry-pick to 4.3.x**

Conflicts:
	spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java
	spring-integration-core/src/test/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandlerTests.java

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

@@ -17,6 +17,7 @@
 package org.springframework.integration.handler;
 
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -49,6 +50,8 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
  * @author David Liu
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Marius Bogoevici
+ *
  * since 4.1
  */
 public abstract class AbstractMessageProducingHandler extends AbstractMessageHandler
@@ -111,14 +114,44 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 	 * @since 4.3.10
 	 */
 	public void setNotPropagatedHeaders(String... headers) {
+		updateNotPropagatedHeaders(headers, false);
+	}
+
+	private void updateNotPropagatedHeaders(String[] headers, boolean merge) {
 		if (!ObjectUtils.isEmpty(headers)) {
 			Assert.noNullElements(headers, "null elements are not allowed in 'headers'");
-			this.notPropagatedHeaders.clear();
+			if (!merge) {
+				this.notPropagatedHeaders.clear();
+			}
 			this.notPropagatedHeaders.addAll(Arrays.asList(headers));
 		}
 		this.selectiveHeaderPropagation = this.notPropagatedHeaders.size() > 0;
 	}
 
+	/**
+	 * Get the header names this handler doesn't propagate.
+	 * @return an immutable {@link java.util.Collection} of headers that will
+	 *         not be copied from the inbound message if
+	 *         {@link #shouldCopyRequestHeaders()} is true.
+	 * @since 4.3.10
+	 * @see #setNotPropagatedHeaders(String...)
+	 */
+	public Collection<String> getNotPropagatedHeaders() {
+		return Collections.unmodifiableSet(this.notPropagatedHeaders);
+	}
+
+	/**
+	 * Add headers that will NOT be copied from the inbound message if
+	 * {@link #shouldCopyRequestHeaders()} is true, instead of overwriting
+	 * the existing set.
+	 * @param headers the headers to not propagate from the inbound message.
+	 * @since 4.3.10
+	 * @see #setNotPropagatedHeaders(String...)
+	 */
+	public void addNotPropagatedHeaders(String... headers) {
+		updateNotPropagatedHeaders(headers, true);
+	}
+
 	@Override
 	protected void onInit() throws Exception {
 		super.onInit();

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

@@ -20,6 +20,8 @@ import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.emptyCollectionOf;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.fail;
 import static org.mockito.BDDMockito.given;
@@ -42,6 +44,7 @@ import org.springframework.messaging.support.GenericMessage;
  * @author Iwein Fuld
  * @author Gunnar Hillert
  * @author Gary Russell
+ * @author Marius Bogoevici
  */
 @RunWith(org.mockito.runners.MockitoJUnitRunner.class)
 public class AbstractReplyProducingMessageHandlerTests {
@@ -84,8 +87,10 @@ public class AbstractReplyProducingMessageHandlerTests {
 			}
 
 		};
+		assertThat(handler.getNotPropagatedHeaders(), emptyCollectionOf(String.class));
 		handler.setNotPropagatedHeaders("foo", "bar");
 		handler.setOutputChannel(this.channel);
+		assertThat(handler.getNotPropagatedHeaders(), containsInAnyOrder("foo", "bar"));
 		ArgumentCaptor<Message<?>> captor = (ArgumentCaptor<Message<?>>) (ArgumentCaptor<?>) ArgumentCaptor.forClass(Message.class);
 		willReturn(true).given(this.channel).send(captor.capture());
 		handler.handleMessage(MessageBuilder.withPayload("hello")
@@ -100,4 +105,35 @@ public class AbstractReplyProducingMessageHandlerTests {
 		assertThat(out.getHeaders().get("baz"), equalTo("BAZ"));
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testNotPropagateAdd() {
+		AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
+
+			@Override
+			protected Object handleRequestMessage(Message<?> requestMessage) {
+				return new GenericMessage<String>("world", Collections.singletonMap("bar", "RAB"));
+			}
+
+		};
+		assertThat(handler.getNotPropagatedHeaders(), emptyCollectionOf(String.class));
+		handler.setNotPropagatedHeaders("foo");
+		handler.addNotPropagatedHeaders("bar");
+		handler.setOutputChannel(this.channel);
+		assertThat(handler.getNotPropagatedHeaders(), containsInAnyOrder("foo", "bar"));
+		ArgumentCaptor<Message<?>> captor =
+				(ArgumentCaptor<Message<?>>) (ArgumentCaptor<?>) ArgumentCaptor.forClass(Message.class);
+		willReturn(true).given(this.channel).send(captor.capture());
+		handler.handleMessage(MessageBuilder.withPayload("hello")
+				.setHeader("foo", "FOO")
+				.setHeader("bar", "BAR")
+				.setHeader("baz", "BAZ")
+				.build());
+		Message<?> out = captor.getValue();
+		assertThat(out, notNullValue());
+		assertThat(out.getHeaders().get("foo"), nullValue());
+		assertThat(out.getHeaders().get("bar"), equalTo("RAB"));
+		assertThat(out.getHeaders().get("baz"), equalTo("BAZ"));
+	}
+
 }