GH-1129: Add JacksonMimeTypeModule

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

@@ -42,6 +42,7 @@
 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.node.TextNode;
 import com.fasterxml.jackson.databind.type.TypeFactory;
 
 /**
@@ -66,12 +67,6 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper {
 					"org.springframework.util"
 			);
 
-	private static final List<String> DEFAULT_TO_STRING_CLASSES =
-			Arrays.asList(
-					"org.springframework.util.MimeType",
-					"org.springframework.http.MediaType"
-			);
-
 	/**
 	 * Header name for java types of other headers.
 	 */
@@ -81,7 +76,7 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper {
 
 	private final Set<String> trustedPackages = new LinkedHashSet<>(DEFAULT_TRUSTED_PACKAGES);
 
-	private final Set<String> toStringClasses = new LinkedHashSet<>(DEFAULT_TO_STRING_CLASSES);
+	private final Set<String> toStringClasses = new LinkedHashSet<>();
 
 	/**
 	 * Construct an instance with the default object mapper and default header patterns
@@ -269,8 +264,8 @@ public void toHeaders(Headers source, final Map<String, Object> headers) {
 						}
 						catch (IOException e) {
 							logger.error(e, () ->
-								"Could not decode json type: " + new String(header.value()) + " for key: "
-								+ header.key());
+									"Could not decode json type: " + new String(header.value()) + " for key: "
+											+ header.key());
 							headers.put(header.key(), header.value());
 						}
 					}
@@ -361,14 +356,19 @@ private class MimeTypeJsonDeserializer extends StdNodeBasedDeserializer<MimeType
 
 		@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);
+			if (root instanceof TextNode) {
+				return MimeType.valueOf(root.asText());
+			}
+			else {
+				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);
+			}
 		}
 
 	}

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.kafka.support;
+
+import java.io.IOException;
+
+import org.springframework.util.MimeType;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
+/**
+ * A {@link SimpleModule} extension for {@link MimeType} serialization.
+ *
+ * @author Artem Bilan
+ *
+ * @since 2.3
+ */
+public final class JacksonMimeTypeModule extends SimpleModule {
+
+	private static final long serialVersionUID = 1L;
+
+	public JacksonMimeTypeModule() {
+		addSerializer(MimeType.class, new MimeTypeSerializer());
+	}
+
+	/**
+	 * Simple {@link JsonSerializer} extension to represent a {@link MimeType} object in the
+	 * target JSON as a plain string.
+	 */
+	private static final class MimeTypeSerializer extends JsonSerializer<MimeType> {
+
+		MimeTypeSerializer() {
+			super();
+		}
+
+		@Override
+		public void serialize(MimeType value, JsonGenerator generator, SerializerProvider serializers)
+				throws IOException {
+
+			generator.writeString(value.toString());
+		}
+
+	}
+
+}

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

@@ -63,6 +63,7 @@ public static ObjectMapper enhancedObjectMapper(ClassLoader classLoader) {
 
 	@SuppressWarnings("unchecked")
 	private static void registerWellKnownModulesIfAvailable(ObjectMapper objectMapper, ClassLoader classLoader) {
+		objectMapper.registerModule(new JacksonMimeTypeModule());
 		try {
 			Class<? extends Module> jdk8Module = (Class<? extends Module>)
 					ClassUtils.forName("com.fasterxml.jackson.datatype.jdk8.Jdk8Module", classLoader);

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

@@ -19,19 +19,19 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.entry;
 
-import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeader;
 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;
@@ -42,6 +42,8 @@
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 1.3
  *
  */
@@ -51,7 +53,7 @@ public class DefaultKafkaHeaderMapperTests {
 	public void testTrustedAndNot() {
 		DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
 		mapper.addToStringClasses(Bar.class.getName());
-		MimeType utf8Text = new MimeType(MimeTypeUtils.TEXT_PLAIN, Charset.forName("UTF-8"));
+		MimeType utf8Text = new MimeType(MimeTypeUtils.TEXT_PLAIN, StandardCharsets.UTF_8);
 		Message<String> message = MessageBuilder.withPayload("foo")
 				.setHeader("foo", "bar".getBytes())
 				.setHeader("baz", "qux")
@@ -60,7 +62,7 @@ public void testTrustedAndNot() {
 				.setHeader(MessageHeaders.REPLY_CHANNEL, new ExecutorSubscribableChannel())
 				.setHeader(MessageHeaders.ERROR_CHANNEL, "errors")
 				.setHeader(MessageHeaders.CONTENT_TYPE, utf8Text)
-				.setHeader("simpleContentType", MimeTypeUtils.TEXT_PLAIN)
+				.setHeader("simpleContentType", MimeTypeUtils.TEXT_PLAIN_VALUE)
 				.setHeader("customToString", new Bar("fiz"))
 				.build();
 		RecordHeaders recordHeaders = new RecordHeaders();
@@ -73,8 +75,8 @@ public void testTrustedAndNot() {
 		assertThat(headers.get("baz")).isEqualTo("qux");
 		assertThat(headers.get("fix")).isInstanceOf(NonTrustedHeaderType.class);
 		assertThat(headers.get("linkedMVMap")).isInstanceOf(LinkedMultiValueMap.class);
-		assertThat(headers.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(utf8Text.toString());
-		assertThat(headers.get("simpleContentType")).isEqualTo(MimeTypeUtils.TEXT_PLAIN.toString());
+		assertThat(headers.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(utf8Text);
+		assertThat(headers.get("simpleContentType")).isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE);
 		assertThat(headers.get(MessageHeaders.REPLY_CHANNEL)).isNull();
 		assertThat(headers.get(MessageHeaders.ERROR_CHANNEL)).isEqualTo("errors");
 		assertThat(headers.get("customToString")).isEqualTo("Bar [field=fiz]");
@@ -94,7 +96,7 @@ public void testTrustedAndNot() {
 	}
 
 	@Test
-	public void testReserializedNonTrusted() {
+	public void testDeserializedNonTrusted() {
 		DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
 		Message<String> message = MessageBuilder.withPayload("foo")
 				.setHeader("fix", new Foo())
@@ -126,27 +128,19 @@ public void testReserializedNonTrusted() {
 	}
 
 	@Test
-	public void testMimeBackwardsCompat() {
+	public void testMimeTypeInHeaders() {
 		DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
 		MessageHeaders headers = new MessageHeaders(
-				Collections.singletonMap("foo", MimeType.valueOf("application/json")));
+				Collections.singletonMap("foo",
+						Arrays.asList(MimeType.valueOf("application/json"), MimeType.valueOf("text/plain"))));
 
 		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"));
+		assertThat(fooHeader).isInstanceOf(List.class);
+		assertThat(fooHeader).asList().containsExactly("application/json", "text/plain");
 	}
 
 	@Test

src/reference/asciidoc/kafka.adoc

@@ -2450,6 +2450,8 @@ You can also extend them to implement some particular configuration logic in the
 Starting with version 2.3, all the JSON-aware components are configured by default with a `JacksonUtils.enhancedObjectMapper()` instance, which comes with the `MapperFeature.DEFAULT_VIEW_INCLUSION` and `DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES` features disabled.
 Also such an instance is supplied with well-known modules for custom data types, such a Java time and Kotlin support.
 See `JacksonUtils.enhancedObjectMapper()` JavaDocs for more information.
+This method also registers a `org.springframework.kafka.support.JacksonMimeTypeModule` for `org.springframework.util.MimeType` objects serialization into the plain string for inter-platform compatibility over the network.
+A `JacksonMimeTypeModule` can be registered as a bean in the application context and it will be auto-configured into https://docs.spring.io/spring-boot/docs/current/reference/html/howto-spring-mvc.html#howto-customize-the-jackson-objectmapper[Spring Boot `ObjectMapper` instance].
 
 Also starting with version 2.3, the `JsonDeserializer` provides `TypeReference`-based constructors for better handling of target generic container types.
 

src/reference/asciidoc/whats-new.adoc

@@ -71,6 +71,7 @@ See <<streams-messaging>> and <<streams-integration>> for more information.
 
 Now all the JSON-aware components are configured by default with a Jackson `ObjectMapper` produced by the `JacksonUtils.enhancedObjectMapper()`.
 The `JsonDeserializer` now provides `TypeReference`-based constructors for better handling of target generic container types.
+Also a `JacksonMimeTypeModule` has been introduced for serialization of `org.springframework.util.MimeType` to plain string.
 See its JavaDocs and <<serdes>> for more information.
 
 A `ByteArrayJsonMessageConverter` has been provided as well as a new super class for all Json converters, `JsonMessageConverter`.