INT-4249: AMQP: Configurable Header Mapping Order

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

Add `headersMappedLast` to outbound endpoints and channels.

* Fix typos in JavaDocs and `amqp.adoc`

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -19,6 +19,7 @@ package org.springframework.integration.amqp.channel;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
 import org.springframework.integration.amqp.support.MappingUtils;
@@ -46,6 +47,8 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
 
 	private MessageDeliveryMode defaultDeliveryMode;
 
+	private boolean headersMappedLast;
+
 	/**
 	 * Construct an instance with the supplied template and default header mappers
 	 * used if the template is a {@link RabbitTemplate} and the message is mapped.
@@ -123,6 +126,26 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
 		return this.extractPayload;
 	}
 
+	/**
+	 * When mapping headers for the outbound message, determine whether the headers are
+	 * mapped before the message is converted, or afterwards. This only affects headers
+	 * that might be added by the message converter. When false, the converter's headers
+	 * win; when true, any headers added by the converter will be overridden (if the
+	 * source message has a header that maps to those headers). You might wish to set this
+	 * to true, for example, when using a
+	 * {@link org.springframework.amqp.support.converter.SimpleMessageConverter} with a
+	 * String payload that contains json; the converter will set the content type to
+	 * {@code text/plain} which can be overridden to {@code application/json} by setting
+	 * the {@link AmqpHeaders#CONTENT_TYPE} message header.
+	 * Only applies when {@link #setExtractPayload(boolean) extractPayload} is true.
+	 * Default: false.
+	 * @param headersMappedLast true if headers are mapped after conversion.
+	 * @since 5.0
+	 */
+	public void setHeadersMappedLast(boolean headersMappedLast) {
+		this.headersMappedLast = headersMappedLast;
+	}
+
 	/**
 	 * Subclasses may override this method to return an Exchange name.
 	 * By default, Messages will be sent to the no-name Direct Exchange.
@@ -159,7 +182,8 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
 	protected boolean doSend(Message<?> message, long timeout) {
 		if (this.extractPayload) {
 			this.amqpTemplate.send(getExchangeName(), getRoutingKey(), MappingUtils.mapMessage(message,
-					this.rabbitTemplate.getMessageConverter(), this.outboundHeaderMapper, this.defaultDeliveryMode));
+					this.rabbitTemplate.getMessageConverter(), this.outboundHeaderMapper, this.defaultDeliveryMode,
+					this.headersMappedLast));
 		}
 		else {
 			this.amqpTemplate.convertAndSend(getExchangeName(), getRoutingKey(), message);

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -139,6 +139,8 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	private volatile AmqpHeaderMapper inboundHeaderMapper = DefaultAmqpHeaderMapper.inboundMapper();
 
+	private boolean headersLast;
+
 	public AmqpChannelFactoryBean() {
 		this(true);
 	}
@@ -341,6 +343,10 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		this.inboundHeaderMapper = inboundMapper;
 	}
 
+	public void setHeadersLast(boolean headersLast) {
+		this.headersLast = headersLast;
+	}
+
 	@Override
 	public Class<?> getObjectType() {
 		return (this.channel != null) ? this.channel.getClass() : AbstractAmqpChannel.class;
@@ -401,6 +407,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		if (this.extractPayload != null) {
 			this.channel.setExtractPayload(this.extractPayload);
 		}
+		this.channel.setHeadersMappedLast(this.headersLast);
 		this.channel.afterPropertiesSet();
 		return this.channel;
 	}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -94,6 +94,7 @@ public class AmqpChannelParser extends AbstractChannelParser {
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "outbound-header-mapper");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "inbound-header-mapper");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "headers-last");
 		return builder;
 	}
 

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -83,6 +83,7 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delay-expression",
 				"delayExpressionString");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "headers-last", "headersMappedLast");
 
 		return builder.getBeanDefinition();
 	}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -102,6 +102,7 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser {
 				null);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delay-expression",
 				"delayExpressionString");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "headers-last", "headersMappedLast");
 
 		return builder;
 	}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseOutboundEndpointSpec.java

@@ -128,4 +128,14 @@ public abstract class
 		return _this();
 	}
 
+	/**
+	 * @param headersLast true to map headers last.
+	 * @return the spec.
+	 * @see AbstractAmqpOutboundEndpoint#setHeadersMappedLast(boolean)
+	 */
+	public S headersMappedLast(boolean headersLast) {
+		this.target.setHeadersMappedLast(headersLast);
+		return _this();
+	}
+
 }

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpPollableMessageChannelSpec.java

@@ -193,6 +193,16 @@ public class AmqpPollableMessageChannelSpec<S extends AmqpPollableMessageChannel
 		return _this();
 	}
 
+	/**
+	 * @param headersLast true to map headers last.
+	 * @return the spec.
+	 * @see AbstractAmqpChannel#setHeadersMappedLast(boolean)
+	 */
+	public S headersMappedLast(boolean headersLast) {
+		this.target.setHeadersMappedLast(headersLast);
+		return _this();
+	}
+
 	@Override
 	protected AbstractAmqpChannel doGet() {
 		Assert.notNull(getId(), "The 'id' or 'queueName' must be specified");

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java

@@ -84,6 +84,8 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
 
 	private volatile ExpressionEvaluatingMessageProcessor<Integer> delayGenerator;
 
+	private boolean headersMappedLast;
+
 	private volatile boolean running;
 
 	public void setHeaderMapper(AmqpHeaderMapper headerMapper) {
@@ -91,6 +93,24 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
 		this.headerMapper = headerMapper;
 	}
 
+	/**
+	 * When mapping headers for the outbound message, determine whether the headers are
+	 * mapped before the message is converted, or afterwards. This only affects headers
+	 * that might be added by the message converter. When false, the converter's headers
+	 * win; when true, any headers added by the converter will be overridden (if the
+	 * source message has a header that maps to those headers). You might wish to set this
+	 * to true, for example, when using a
+	 * {@link org.springframework.amqp.support.converter.SimpleMessageConverter} with a
+	 * String payload that contains json; the converter will set the content type to
+	 * {@code text/plain} which can be overridden to {@code application/json} by setting
+	 * the {@link AmqpHeaders#CONTENT_TYPE} message header. Default: false.
+	 * @param headersMappedLast true if headers are mapped after conversion.
+	 * @since 5.0
+	 */
+	public void setHeadersMappedLast(boolean headersMappedLast) {
+		this.headersMappedLast = headersMappedLast;
+	}
+
 	public void setExchangeName(String exchangeName) {
 		Assert.notNull(exchangeName, "exchangeName must not be null");
 		this.exchangeName = exchangeName;
@@ -293,6 +313,10 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
 		return this.lazyConnect;
 	}
 
+	protected boolean isHeadersMappedLast() {
+		return this.headersMappedLast;
+	}
+
 	@Override
 	protected final void doInit() {
 		Assert.state(this.exchangeNameExpression == null || this.exchangeName == null,

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java

@@ -102,7 +102,7 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
 		if (this.amqpTemplate instanceof RabbitTemplate) {
 			MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
 			org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter,
-					getHeaderMapper(), getDefaultDeliveryMode());
+					getHeaderMapper(), getDefaultDeliveryMode(), isHeadersMappedLast());
 			addDelayProperty(requestMessage, amqpMessage);
 			((RabbitTemplate) this.amqpTemplate).send(exchangeName, routingKey, amqpMessage, correlationData);
 		}
@@ -122,7 +122,7 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
 				"RabbitTemplate implementation is required for publisher confirms");
 		MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
 		org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter,
-				getHeaderMapper(), getDefaultDeliveryMode());
+				getHeaderMapper(), getDefaultDeliveryMode(), isHeadersMappedLast());
 		addDelayProperty(requestMessage, amqpMessage);
 		org.springframework.amqp.core.Message amqpReplyMessage =
 				((RabbitTemplate) this.amqpTemplate).sendAndReceive(exchangeName, routingKey, amqpMessage,

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java

@@ -76,7 +76,7 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint {
 	@Override
 	protected Object handleRequestMessage(Message<?> requestMessage) {
 		org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage,
-				this.messageConverter, getHeaderMapper(), getDefaultDeliveryMode());
+				this.messageConverter, getHeaderMapper(), getDefaultDeliveryMode(), isHeadersMappedLast());
 		addDelayProperty(requestMessage, amqpMessage);
 		RabbitMessageFuture future = this.template.sendAndReceive(generateExchangeName(requestMessage),
 				generateRoutingKey(requestMessage), amqpMessage);

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/MappingUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -22,6 +22,8 @@ import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.MimeType;
 
 /**
  * Utility methods used during message mapping.
@@ -37,29 +39,54 @@ public final class MappingUtils {
 	}
 
 	/**
-	 * Map an o.s.Message to an o.s.a.core.Message.
+	 * Map an o.s.Message to an o.s.a.core.Message. When using a
+	 * {@link ContentTypeDelegatingMessageConverter}, {@link AmqpHeaders#CONTENT_TYPE} and
+	 * {@link MessageHeaders#CONTENT_TYPE} will be used for the selection, with the AMQP
+	 * header taking precedence.
 	 * @param requestMessage the request message.
 	 * @param converter the message converter to use.
 	 * @param headerMapper the header mapper to use.
 	 * @param defaultDeliveryMode the default delivery mode.
+	 * @param headersMappedLast true if headers are mapped after conversion.
 	 * @return the mapped Message.
 	 */
 	public static org.springframework.amqp.core.Message mapMessage(Message<?> requestMessage,
-			MessageConverter converter, AmqpHeaderMapper headerMapper, MessageDeliveryMode defaultDeliveryMode) {
+			MessageConverter converter, AmqpHeaderMapper headerMapper, MessageDeliveryMode defaultDeliveryMode,
+			boolean headersMappedLast) {
 		MessageProperties amqpMessageProperties = new MessageProperties();
 		org.springframework.amqp.core.Message amqpMessage;
-		if (converter instanceof ContentTypeDelegatingMessageConverter) {
+		if (!headersMappedLast) {
 			headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
-			amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
 		}
-		else { // See INT-3002 - map headers last if we're not using a CTDMC
-			amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
+		if (converter instanceof ContentTypeDelegatingMessageConverter && headersMappedLast) {
+			String contentType = contentTypeAsString(requestMessage.getHeaders());
+			if (contentType != null) {
+				amqpMessageProperties.setContentType(contentType);
+			}
+		}
+		amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
+		if (headersMappedLast) {
 			headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
 		}
 		checkDeliveryMode(requestMessage, amqpMessageProperties, defaultDeliveryMode);
 		return amqpMessage;
 	}
 
+	private static String contentTypeAsString(MessageHeaders headers) {
+		Object contentType = headers.get(AmqpHeaders.CONTENT_TYPE);
+		if (contentType instanceof MimeType) {
+			contentType = contentType.toString();
+		}
+		if (contentType instanceof String) {
+			return (String) contentType;
+		}
+		else if (contentType != null) {
+			throw new IllegalArgumentException(AmqpHeaders.CONTENT_TYPE
+					+ " header must be a MimeType or String, found: " + contentType.getClass().getName());
+		}
+		return null;
+	}
+
 	/**
 	 * Check the delivery mode and update with the default if not already present.
 	 * @param requestMessage the request message.

spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-5.0.xsd

@@ -364,6 +364,7 @@
 	Set to 'true' to extract the message payload and map the o.s.messaging.Message to an o.s.amqp.core.Message in
 	a similar manner to a pair of channel adapters. When 'false' the entire message is converted requiring either
 	Java serializable contents or a custom message converter. Also see inbound and outbound mapped headers.
+	Also see 'headers-last', which only applies if this is 'true'.
 				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
@@ -384,7 +385,8 @@
 			</xsd:annotation>
 		</xsd:attribute>
 		<xsd:attributeGroup ref="containerAndTemplateAttributes"/>
-		<xsd:attributeGroup ref="integration:subscribersAttributeGroup" />
+		<xsd:attributeGroup ref="integration:subscribersAttributeGroup"/>
+		<xsd:attributeGroup ref="headersLast"/>
 	</xsd:complexType>
 
 	<xsd:complexType name="outboundType">
@@ -557,6 +559,7 @@ property set to TRUE.
 			</xsd:annotation>
 		</xsd:attribute>
 		<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
+		<xsd:attributeGroup ref="headersLast"/>
 	</xsd:complexType>
 
 	<xsd:complexType name="outboundGatewayType">
@@ -1014,6 +1017,21 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
 		<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
 	</xsd:attributeGroup>
 
+	<xsd:attributeGroup name="headersLast">
+		<xsd:attribute name="headers-last">
+			<xsd:annotation>
+				<xsd:documentation>
+					Whether headers are mapped before or after conversion from a messaging Message to
+					a spring amqp Message. Set to true, for example, if you wish to override the
+					contentType header set by the converter.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:union memberTypes="xsd:boolean xsd:string" />
+			</xsd:simpleType>
+		</xsd:attribute>
+	</xsd:attributeGroup>
+
 	<xsd:simpleType name="ackModeEnumeration">
 		<xsd:restriction base="xsd:string">
 			<xsd:enumeration value="AUTO"/>

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml

@@ -26,6 +26,7 @@
 		inbound-header-mapper="inMapper" outbound-header-mapper="outMapper" />
 
 	<amqp:channel id="pollableWithEP" extract-payload="true" message-driven="false"
+		headers-last="true"
 		inbound-header-mapper="inMapper" outbound-header-mapper="outMapper" />
 
 	<amqp:publish-subscribe-channel id="pubSubWithEP" extract-payload="true"

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -106,7 +106,9 @@ public class AmqpChannelParserTests {
 		checkExtract(this.pubSubWithEP);
 		assertEquals(MessageDeliveryMode.NON_PERSISTENT,
 				TestUtils.getPropertyValue(this.withEP, "defaultDeliveryMode"));
+		assertFalse(TestUtils.getPropertyValue(this.withEP, "headersMappedLast", Boolean.class));
 		assertNull(TestUtils.getPropertyValue(this.pollableWithEP, "defaultDeliveryMode"));
+		assertTrue(TestUtils.getPropertyValue(this.pollableWithEP, "headersMappedLast", Boolean.class));
 	}
 
 	private void checkExtract(AbstractAmqpChannel channel) {

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml

@@ -34,6 +34,7 @@
 
 	<amqp:outbound-channel-adapter id="withCustomHeaderMapper"
 								   exchange-name="test.exchange"
+								   headers-last="true"
 								   header-mapper="customHeaderMapper"/>
 
 	<int:channel id="requestChannel"/>

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java

@@ -128,6 +128,7 @@ public class AmqpOutboundChannelAdapterParserTests {
 		assertEquals("42",
 				TestUtils.getPropertyValue(endpoint, "delayExpression", org.springframework.expression.Expression.class)
 						.getExpressionString());
+		assertFalse(TestUtils.getPropertyValue(endpoint, "headersMappedLast", Boolean.class));
 
 		Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
 		amqpTemplateField.setAccessible(true);
@@ -297,6 +298,8 @@ public class AmqpOutboundChannelAdapterParserTests {
 		AmqpHeaderMapper headerMapper = TestUtils.getPropertyValue(this.amqpMessageHandlerWithCustomHeaderMapper,
 				"headerMapper", AmqpHeaderMapper.class);
 		assertSame(this.context.getBean("customHeaderMapper"), headerMapper);
+		assertTrue(TestUtils.getPropertyValue(this.amqpMessageHandlerWithCustomHeaderMapper,
+				"headersMappedLast", Boolean.class));
 	}
 
 	@Test

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml

@@ -49,6 +49,7 @@
 		order="5"
 		default-delivery-mode="NON_PERSISTENT"
 		requires-reply="false"
+		headers-last="true"
 		mapped-request-headers="foo*"
 		mapped-reply-headers="bar*"/>
 

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java

@@ -101,14 +101,15 @@ public class AmqpOutboundGatewayParserTests {
 	public void withHeaderMapperCustomRequestResponse() {
 		ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
 				"AmqpOutboundGatewayParserTests-context.xml", this.getClass());
-		Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomRequestResponse");
+		Object eventDrivenConsumer = context.getBean("withHeaderMapperCustomRequestResponse");
 
-		AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler",
+		AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler",
 				AmqpOutboundEndpoint.class);
 		assertNotNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));
 		assertFalse(TestUtils.getPropertyValue(endpoint, "lazyConnect", Boolean.class));
 
 		assertFalse(TestUtils.getPropertyValue(endpoint, "requiresReply", Boolean.class));
+		assertTrue(TestUtils.getPropertyValue(endpoint, "headersMappedLast", Boolean.class));
 
 		Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
 		amqpTemplateField.setAccessible(true);
@@ -170,11 +171,12 @@ public class AmqpOutboundGatewayParserTests {
 	public void withHeaderMapperCustomAndStandardResponse() {
 		ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
 				"AmqpOutboundGatewayParserTests-context.xml", this.getClass());
-		Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomAndStandardResponse");
+		Object eventDrivenConsumer = context.getBean("withHeaderMapperCustomAndStandardResponse");
 
-		AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler",
+		AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler",
 				AmqpOutboundEndpoint.class);
 		assertNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));
+		assertFalse(TestUtils.getPropertyValue(endpoint, "headersMappedLast", Boolean.class));
 
 		Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
 		amqpTemplateField.setAccessible(true);

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java

@@ -119,6 +119,7 @@ public class OutboundEndpointTests {
 		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
 		RabbitTemplate amqpTemplate = spy(new RabbitTemplate(connectionFactory));
 		AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate);
+		endpoint.setHeadersMappedLast(true);
 		final AtomicReference<Message> amqpMessage =
 				new AtomicReference<Message>();
 		willAnswer(invocation -> {
@@ -138,6 +139,7 @@ public class OutboundEndpointTests {
 		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
 		TestRabbitTemplate amqpTemplate = spy(new TestRabbitTemplate(connectionFactory));
 		AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate);
+		endpoint.setHeadersMappedLast(true);
 		endpoint.setExpectReply(true);
 		DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper();
 		mapper.setRequestHeaderNames("*");

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/MappingUtilsTests.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2017 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.amqp.support;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+import org.junit.Test;
+
+import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.amqp.support.converter.SimpleMessageConverter;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+
+/**
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+public class MappingUtilsTests {
+
+	@Test
+	public void testMapping() {
+		Message<?> requestMessage = MessageBuilder.withPayload("foo")
+				.setHeader(AmqpHeaders.CONTENT_TYPE, "my/ct")
+				.build();
+		MessageConverter converter = new SimpleMessageConverter();
+		AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper();
+		MessageDeliveryMode defaultDeliveryMode = MessageDeliveryMode.NON_PERSISTENT;
+		boolean headersMappedLast = false;
+		org.springframework.amqp.core.Message mapped = MappingUtils.mapMessage(requestMessage, converter, headerMapper,
+				defaultDeliveryMode, headersMappedLast);
+		assertThat(mapped.getMessageProperties().getContentType(), equalTo("text/plain"));
+
+		headersMappedLast = true;
+		mapped = MappingUtils.mapMessage(requestMessage, converter, headerMapper,
+				defaultDeliveryMode, headersMappedLast);
+		assertThat(mapped.getMessageProperties().getContentType(), equalTo("my/ct"));
+
+		ContentTypeDelegatingMessageConverter ctdConverter = new ContentTypeDelegatingMessageConverter();
+		ctdConverter.addDelegate("my/ct", converter);
+		mapped = MappingUtils.mapMessage(requestMessage, ctdConverter, headerMapper,
+				defaultDeliveryMode, headersMappedLast);
+		assertThat(mapped.getMessageProperties().getContentType(), equalTo("my/ct"));
+
+		headersMappedLast = false;
+		mapped = MappingUtils.mapMessage(requestMessage, ctdConverter, headerMapper,
+				defaultDeliveryMode, headersMappedLast);
+		assertThat(mapped.getMessageProperties().getContentType(), equalTo("text/plain"));
+
+		headersMappedLast = true;
+		requestMessage = MessageBuilder.withPayload("foo")
+				.setHeader(AmqpHeaders.CONTENT_TYPE, 42)
+				.build();
+		try {
+			mapped = MappingUtils.mapMessage(requestMessage, ctdConverter, headerMapper,
+					defaultDeliveryMode, headersMappedLast);
+			fail("Expected IllegalArgumentException");
+		}
+		catch (IllegalArgumentException e) {
+			assertThat(e.getMessage(),
+					equalTo("contentType header must be a MimeType or String, found: java.lang.Integer"));
+		}
+	}
+
+}

src/reference/asciidoc/amqp.adoc

@@ -1140,6 +1140,22 @@ JSON converter to be selected.
 
 This applies to both the outbound channel adapter and gateway.
 
+NOTE
+====
+Starting with _version 5.0_, headers that are added to the `MessageProperties` of the outbound message are never overwritten by mapped headers (by default).
+Previously, this was only the case if the message converter was a `ContentTypeDelegatingMessageConverter` (in that case, the header was mapped first, so that the proper converter could be selected).
+For other converters, such as the `SimpleMessageConverter`, mapped headers overwrote any headers added by the converter.
+This caused problems when an outbound message had some left over `contentType` header (perhaps from an inbound channel adapter) and the correct outbound `contentType` was incorrectly overwritten.
+The work-around was to use a header filter to remove the header before sending the message to the outbound endpoint.
+
+There are, however, cases where the previous behavior is desired.
+For example, with a `String` payload containing JSON, the `SimpleMessageConverter` is not aware of the content and sets the `contentType` message property to `text/plain`, but your application would like to override that to `application/json` by setting the the `contentType` header of the message sent to the outbound endpoint.
+The `ObjectToJsonTransformer` does exactly that (by default).
+
+There is now a property on the outbound channel adapter and gateway (as well as AMQP-backed channels) `headersMappedLast`.
+Setting this to `true` will restore the behavior of overwriting the property added by the converter.
+====
+
 [[amqp-user-id]]
 === Outbound User Id
 

src/reference/asciidoc/whats-new.adoc

@@ -153,6 +153,10 @@ See <<amqp-inbound-channel-adapter>> for more information.
 Pollable AMQP-backed channels now block the poller thread for the poller's configured `receiveTimeout` (default 1 second).
 See <<amqp-channels>> for more information.
 
+Headers, such as `contentType` that are added to message properties by the message converter are now used in the final message; previously, it depended on the converter type as to which headers/message properties appeared in the final message.
+To override headers set by the converter, set the `headersMappedLast` property to `true`.
+See <<content-type-conversion-outbound>> for more information.
+
 ==== HTTP Changes
 
 The `DefaultHttpHeaderMapper.userDefinedHeaderPrefix` property is now an empty string by default instead of `X-`.