INT-4387: Fix incorrect contentType mapping

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

Don't put a copy in the message properties headers map.

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

@@ -359,7 +359,8 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 			MessageProperties amqpMessageProperties) {
 		// do not overwrite an existing header with the same key
 		// TODO: do we need to expose a boolean 'overwrite' flag?
-		if (!amqpMessageProperties.getHeaders().containsKey(headerName)) {
+		if (!amqpMessageProperties.getHeaders().containsKey(headerName)
+				&& !AmqpHeaders.CONTENT_TYPE.equals(headerName)) {
 			amqpMessageProperties.setHeader(headerName, headerValue);
 		}
 	}

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

@@ -97,8 +97,8 @@ public class DefaultAmqpHeaderMapperTests {
 		headerMapper.fromHeadersToRequest(integrationHeaders, amqpProperties);
 		Set<String> headerKeys = amqpProperties.getHeaders().keySet();
 		for (String headerKey : headerKeys) {
-			if (headerKey.startsWith(AmqpHeaders.PREFIX)) {
-				fail();
+			if (headerKey.startsWith(AmqpHeaders.PREFIX) || headerKey.equals("contentType")) {
+				fail("Unexpected header in properties.headers: " + headerKey);
 			}
 		}
 		assertEquals("test.appId", amqpProperties.getAppId());

src/reference/asciidoc/amqp.adoc

@@ -1388,6 +1388,8 @@ public IntegrationFlow pubSubInFlow(ConnectionFactory connectionFactory) {
 [[amqp-message-headers]]
 === AMQP Message Headers
 
+==== Overview
+
 The Spring Integration AMQP Adapters will map all AMQP properties and headers automatically.
 (This is a change in 4.3 - previously, only standard headers were mapped).
 These properties will be copied by default to and from Spring Integration `MessageHeaders` using the
@@ -1423,7 +1425,7 @@ Class `org.springframework.amqp.support.AmqpHeaders` identifies the default head
 
 * amqp_contentLength
 
-* content-type
+* content-type (see <<amqp-content-type>>)
 
 * amqp_correlationId
 
@@ -1508,6 +1510,19 @@ NOTE: Starting with _version 5.1_, the `DefaultAmqpHeaderMapper` will fall back
 Inbound properties will be mapped to the `amqp_*` headers as before.
 It is useful to populate the `messageId` property when message consumers are using stateful retry.
 
+[[amqp-content-type]]
+==== contentType Header
+
+Unlike other headers, the `AmqpHeaders.CONTENT_TYPE` is not prefixed with `amqp_`; this allows transparent passing of the contentType header across different technologies.
+E.g. an inbound HTTP message sent to a RabbitMQ queue.
+
+The `contentType` header is mapped to Spring AMQP's `MessageProperties.contentType` property and that is subsequently mapped to RabbitMQ's `content_type` property.
+
+Prior to _version 5.1_, this header was also mapped as an entry in the `MessageProperties.headers` map; this was incorrect and, furthermore, the value could be wrong since the underlying Spring AMQP message converter might have changed the content type.
+Such a change would be reflected in the first-class `content_type` property, but not in the RabbitMQ headers map.
+Inbound mapping ignored the headers map value.
+`contentType` is no longer mapped to an entry in the headers map.
+
 [[amqp-strict-ordering]]
 === Strict Message Ordering
 

src/reference/asciidoc/whats-new.adoc

@@ -55,3 +55,6 @@ See <<java-dsl-flows>> for more information.
 
 `ID` and `Timestamp` header mapping changes in the `DefaultAmqpHeaderMapper`.
 See the note near the bottom of <<amqp-message-headers>> for more information.
+
+The `contentType` header is no longer incorrectly mapped as an entry in the general headers map.
+See <<amqp-content-type>> for more information.