GH-3096: Skip RESOLVABLE_TYPE header in mapping (#3098)

* GH-3096: Skip RESOLVABLE_TYPE header in mapping

Fixes #3096

When we sent an AMQP message we should not map a
`JsonHeaders.RESOLVABLE_TYPE` header which is a `ResolvableType` and
isn not compatible after converting to string

Also improve `JsonToObjectTransformer` to ignore a
`JsonHeaders.RESOLVABLE_TYPE` when it is type of String

* * Fix `obtainResolvableTypeFromHeadersIfAny()` logic

Author: artembilan

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

@@ -263,13 +263,15 @@ private void mapJsonHeaders(Map<String, Object> headers, MessageProperties amqpM
 			Map<String, String> jsonHeaders = new HashMap<>();
 
 			for (String jsonHeader : JsonHeaders.HEADERS) {
-				Object value = getHeaderIfAvailable(headers, jsonHeader, Object.class);
-				if (value != null) {
-					headers.remove(jsonHeader);
-					if (value instanceof Class<?>) {
-						value = ((Class<?>) value).getName();
+				if (!JsonHeaders.RESOLVABLE_TYPE.equals(jsonHeader)) {
+					Object value = getHeaderIfAvailable(headers, jsonHeader, Object.class);
+					if (value != null) {
+						headers.remove(jsonHeader);
+						if (value instanceof Class<?>) {
+							value = ((Class<?>) value).getName();
+						}
+						jsonHeaders.put(jsonHeader.replaceFirst(JsonHeaders.PREFIX, ""), value.toString());
 					}
-					jsonHeaders.put(jsonHeader.replaceFirst(JsonHeaders.PREFIX, ""), value.toString());
 				}
 			}
 			amqpMessageProperties.getHeaders().putAll(jsonHeaders);

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

@@ -20,6 +20,7 @@
 import static org.assertj.core.api.Assertions.fail;
 import static org.mockito.Mockito.mock;
 
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
@@ -33,7 +34,9 @@
 import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
 import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.core.ResolvableType;
 import org.springframework.http.MediaType;
+import org.springframework.integration.mapping.support.JsonHeaders;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.GenericMessage;
@@ -310,4 +313,16 @@ public void inboundOutbound() {
 		assertThat(headers.get("x-death")).isEqualTo("foo");
 	}
 
+	@Test
+	public void jsonHeadersResolvableTypeSkipped() {
+		DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper();
+		MessageHeaders integrationHeaders =
+				new MessageHeaders(
+						Collections.singletonMap(JsonHeaders.RESOLVABLE_TYPE, ResolvableType.forClass(String.class)));
+		MessageProperties amqpProperties = new MessageProperties();
+		headerMapper.fromHeadersToReply(integrationHeaders, amqpProperties);
+
+		assertThat(amqpProperties.getHeaders()).doesNotContainKeys(JsonHeaders.RESOLVABLE_TYPE);
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/json/JsonToObjectTransformer.java

@@ -144,11 +144,11 @@ protected Object doTransform(Message<?> message) {
 		}
 	}
 
-
+	@Nullable
 	private ResolvableType obtainResolvableTypeFromHeadersIfAny(MessageHeaders headers) {
-		ResolvableType valueType = headers.get(JsonHeaders.RESOLVABLE_TYPE, ResolvableType.class);
+		Object valueType = headers.get(JsonHeaders.RESOLVABLE_TYPE);
 		Object typeIdHeader = headers.get(JsonHeaders.TYPE_ID);
-		if (valueType == null && typeIdHeader != null) {
+		if (typeIdHeader != null) {
 			Class<?> targetClass = getClassForValue(typeIdHeader);
 			Class<?> contentClass = null;
 			Class<?> keyClass = null;
@@ -163,8 +163,9 @@ private ResolvableType obtainResolvableTypeFromHeadersIfAny(MessageHeaders heade
 
 			valueType = JsonObjectMapper.buildResolvableType(targetClass, contentClass, keyClass);
 		}
-
-		return valueType;
+		return valueType instanceof ResolvableType
+				? (ResolvableType) valueType
+				: null;
 	}
 
 	private Class<?> getClassForValue(Object classValue) {