GH-892: Reinstate MimeTypeJsonDeserializer

Fixes #892

For compatibility with older Spring Kafka producer, it is better to keep
a `MimeTypeJsonDeserializer` when they can send `MimeType` headers
as a serialized version, not string representation

Author: artembilan

spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java

@@ -34,9 +34,15 @@
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
+import org.springframework.util.MimeType;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.deser.std.StdNodeBasedDeserializer;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.databind.type.TypeFactory;
 
 /**
  * Default header mapper for Apache Kafka.
@@ -46,6 +52,8 @@
  * Header types are added to a special header {@link #JSON_TYPES}.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 1.3
  *
  */
@@ -54,7 +62,8 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper {
 	private static final List<String> DEFAULT_TRUSTED_PACKAGES =
 			Arrays.asList(
 					"java.util",
-					"java.lang"
+					"java.lang",
+					"org.springframework.util"
 			);
 
 	private static final List<String> DEFAULT_TO_STRING_CLASSES =
@@ -136,6 +145,8 @@ public DefaultKafkaHeaderMapper(ObjectMapper objectMapper, String... patterns) {
 		Assert.notNull(objectMapper, "'objectMapper' must not be null");
 		Assert.noNullElements(patterns, "'patterns' must not have null elements");
 		this.objectMapper = objectMapper;
+		this.objectMapper
+				.registerModule(new SimpleModule().addDeserializer(MimeType.class, new MimeTypeJsonDeserializer()));
 	}
 
 	/**
@@ -233,7 +244,6 @@ public void fromHeaders(MessageHeaders headers, Headers target) {
 		}
 	}
 
-	@SuppressWarnings("unchecked")
 	@Override
 	public void toHeaders(Headers source, final Map<String, Object> headers) {
 		final Map<String, String> jsonTypes = decodeJsonTypes(source);
@@ -257,7 +267,8 @@ public void toHeaders(Headers source, final Map<String, Object> headers) {
 							headers.put(h.key(), getObjectMapper().readValue(h.value(), type));
 						}
 						catch (IOException e) {
-							logger.error("Could not decode json type: " + new String(h.value()) + " for key: " + h.key(),
+							logger.error("Could not decode json type: " + new String(h.value()) + " for key: " + h
+											.key(),
 									e);
 							headers.put(h.key(), h.value());
 						}
@@ -310,6 +321,34 @@ protected boolean trusted(String requestedType) {
 		return true;
 	}
 
+
+	/**
+	 * The {@link StdNodeBasedDeserializer} extension for {@link MimeType} deserialization.
+	 * It is presented here for backward compatibility when older producers send {@link MimeType}
+	 * headers as serialization version.
+	 */
+	private class MimeTypeJsonDeserializer extends StdNodeBasedDeserializer<MimeType> {
+
+		private static final long serialVersionUID = 1L;
+
+		MimeTypeJsonDeserializer() {
+			super(MimeType.class);
+		}
+
+		@Override
+		public MimeType convert(JsonNode root, DeserializationContext ctxt) throws IOException {
+			JsonNode type = root.get("type");
+			JsonNode subType = root.get("subtype");
+			JsonNode parameters = root.get("parameters");
+			Map<String, String> params =
+					DefaultKafkaHeaderMapper.this.objectMapper.readValue(parameters.traverse(),
+							TypeFactory.defaultInstance()
+									.constructMapType(HashMap.class, String.class, String.class));
+			return new MimeType(type.asText(), subType.asText(), params);
+		}
+
+	}
+
 	/**
 	 * Represents a header that could not be decoded due to an untrusted type.
 	 */

spring-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java

@@ -19,13 +19,16 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.nio.charset.Charset;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.junit.Test;
 
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper.NonTrustedHeaderType;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.ExecutorSubscribableChannel;
@@ -83,6 +86,30 @@ public void test() {
 		assertThat(headers).hasSize(7);
 	}
 
+	@Test
+	public void testMimeBackwardsCompat() {
+		DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
+		MessageHeaders headers = new MessageHeaders(
+				Collections.singletonMap("foo", MimeType.valueOf("application/json")));
+
+		RecordHeaders recordHeaders = new RecordHeaders();
+		mapper.fromHeaders(headers, recordHeaders);
+		Map<String, Object> receivedHeaders = new HashMap<>();
+		mapper.toHeaders(recordHeaders, receivedHeaders);
+		Object fooHeader = receivedHeaders.get("foo");
+		assertThat(fooHeader).isInstanceOf(String.class);
+		assertThat(fooHeader).isEqualTo("application/json");
+
+		KafkaTestUtils.getPropertyValue(mapper, "toStringClasses", Set.class).clear();
+		recordHeaders = new RecordHeaders();
+		mapper.fromHeaders(headers, recordHeaders);
+		receivedHeaders = new HashMap<>();
+		mapper.toHeaders(recordHeaders, receivedHeaders);
+		fooHeader = receivedHeaders.get("foo");
+		assertThat(fooHeader).isInstanceOf(MimeType.class);
+		assertThat(fooHeader).isEqualTo(MimeType.valueOf("application/json"));
+	}
+
 	public static final class Foo {
 
 		private String bar = "bar";