GH-1139: Json Serialization: Add Fluent API

Resolves #1139

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java

@@ -463,4 +463,51 @@ public void close() {
 		// No-op
 	}
 
+	// Fluent API
+
+	/**
+	 * Designate this deserializer for deserializing keys (default is values); only
+	 * applies if the default type mapper is used.
+	 * @return the deserializer.
+	 * @since 2.3
+	 */
+	public JsonDeserializer<T> forKeys() {
+		setUseTypeMapperForKey(true);
+		return this;
+	}
+
+	/**
+	 * Don't remove type information headers.
+	 * @return the deserializer.
+	 * @since 2.3
+	 * @see #setRemoveTypeHeaders(boolean)
+	 */
+	public JsonDeserializer<T> dontRemoveTypeHeaders() {
+		setRemoveTypeHeaders(false);
+		return this;
+	}
+
+	/**
+	 * Ignore type information headers and use the configured target class.
+	 * @return the deserializer.
+	 * @since 2.3
+	 * @see #setUseTypeHeaders(boolean)
+	 */
+	public JsonDeserializer<T> ignoreTypeHeaders() {
+		setUseTypeHeaders(false);
+		return this;
+	}
+
+	/**
+	 * Use the supplied {@link Jackson2JavaTypeMapper}.
+	 * @param mapper the mapper.
+	 * @return the deserializer.
+	 * @since 2.3
+	 * @see #setTypeMapper(Jackson2JavaTypeMapper)
+	 */
+	public JsonDeserializer<T> typeMapper(Jackson2JavaTypeMapper mapper) {
+		setTypeMapper(mapper);
+		return this;
+	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerde.java

@@ -24,6 +24,7 @@
 
 import org.springframework.core.ResolvableType;
 import org.springframework.kafka.support.JacksonUtils;
+import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -111,10 +112,65 @@ public Deserializer<T> deserializer() {
 	 * @param isKey Use key type headers if true
 	 * @return the JsonSerde
 	 * @since 2.1.3
+	 * @deprecated in favor of {@link #forKeys()}.
 	 */
+	@Deprecated
 	public JsonSerde<T> setUseTypeMapperForKey(boolean isKey) {
-		this.jsonSerializer.setUseTypeMapperForKey(isKey);
-		this.jsonDeserializer.setUseTypeMapperForKey(isKey);
+		return forKeys();
+	}
+
+	// Fluent API
+
+	/**
+	 * Designate this Serde for serializing/deserializing keys (default is values).
+	 * @return the serde.
+	 * @since 2.3
+	 */
+	public JsonSerde<T> forKeys() {
+		this.jsonSerializer.forKeys();
+		this.jsonDeserializer.forKeys();
+		return this;
+	}
+
+	/**
+	 * Configure the serializer to not add type information.
+	 * @return the serde.
+	 * @since 2.3
+	 */
+	public JsonSerde<T> noTypeInfo() {
+		this.jsonSerializer.noTypeInfo();
+		return this;
+	}
+
+	/**
+	 * Don't remove type information headers after deserialization.
+	 * @return the serde.
+	 * @since 2.3
+	 */
+	public JsonSerde<T> dontRemoveTypeHeaders() {
+		this.jsonDeserializer.dontRemoveTypeHeaders();
+		return this;
+	}
+
+	/**
+	 * Ignore type information headers and use the configured target class.
+	 * @return the serde.
+	 * @since 2.3
+	 */
+	public JsonSerde<T> ignoreTypeHeaders() {
+		this.jsonDeserializer.ignoreTypeHeaders();
+		return this;
+	}
+
+	/**
+	 * Use the supplied {@link Jackson2JavaTypeMapper}.
+	 * @param mapper the mapper.
+	 * @return the serde.
+	 * @since 2.3
+	 */
+	public JsonSerde<T> typeMapper(Jackson2JavaTypeMapper mapper) {
+		this.jsonSerializer.setTypeMapper(mapper);
+		this.jsonDeserializer.setTypeMapper(mapper);
 		return this;
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java

@@ -186,4 +186,41 @@ public void close() {
 		// No-op
 	}
 
+	// Fluent API
+
+	/**
+	 * Designate this serializer for serializing keys (default is values); only applies if
+	 * the default type mapper is used.
+	 * @return the serializer.
+	 * @since 2.3
+	 * @see #setUseTypeMapperForKey(boolean)
+	 */
+	public JsonSerializer<T> forKeys() {
+		setUseTypeMapperForKey(true);
+		return this;
+	}
+
+	/**
+	 * Do not include type info headers.
+	 * @return the serializer.
+	 * @since 2.3
+	 * @see #setAddTypeInfo(boolean)
+	 */
+	public JsonSerializer<T> noTypeInfo() {
+		setAddTypeInfo(false);
+		return this;
+	}
+
+	/**
+	 * Use the supplied {@link Jackson2JavaTypeMapper}.
+	 * @param mapper the mapper.
+	 * @return the serializer.
+	 * @since 2.3
+	 * @see #setTypeMapper(Jackson2JavaTypeMapper)
+	 */
+	public JsonSerializer<T> typeMapper(Jackson2JavaTypeMapper mapper) {
+		setTypeMapper(mapper);
+		return this;
+	}
+
 }

spring-kafka/src/test/java/org/springframework/kafka/streams/KafkaStreamsJsonSerializationTests.java

@@ -79,7 +79,7 @@ public class KafkaStreamsJsonSerializationTests {
 	public static final String OBJECT_OUTPUT_TOPIC = "object-output-topic";
 
 	public static final JsonSerde<JsonObjectKey> jsonObjectKeySerde =
-			new JsonSerde<>(JsonObjectKey.class).setUseTypeMapperForKey(true);
+			new JsonSerde<>(JsonObjectKey.class).forKeys();
 
 	public static final JsonSerde<JsonObjectValue> jsonObjectValueSerde = new JsonSerde<>(JsonObjectValue.class);
 

spring-kafka/src/test/java/org/springframework/kafka/support/serializer/JsonSerdeTests.java

@@ -0,0 +1,51 @@
+/*
+ * 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.serializer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.kafka.support.converter.AbstractJavaTypeMapper;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+
+/**
+ * @author Gary Russell
+ * @since 2.3
+ *
+ */
+public class JsonSerdeTests {
+
+	@Test
+	void noTypeInfo() {
+		JsonSerde<String> serde = new JsonSerde<>(String.class)
+				.forKeys()
+				.noTypeInfo()
+				.ignoreTypeHeaders()
+				.dontRemoveTypeHeaders();
+		assertThat(KafkaTestUtils.getPropertyValue(serde, "jsonSerializer.typeMapper.classIdFieldName"))
+				.isEqualTo(AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME);
+		assertThat(KafkaTestUtils.getPropertyValue(serde, "jsonDeserializer.typeMapper.classIdFieldName"))
+				.isEqualTo(AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME);
+		assertThat(KafkaTestUtils.getPropertyValue(serde, "jsonSerializer.addTypeInfo", Boolean.class)).isFalse();
+		assertThat(KafkaTestUtils.getPropertyValue(serde, "jsonDeserializer.useTypeHeaders", Boolean.class)).isFalse();
+		assertThat(KafkaTestUtils.getPropertyValue(serde, "jsonDeserializer.removeTypeHeaders", Boolean.class))
+				.isFalse();
+		serde.close();
+	}
+
+}

src/reference/asciidoc/kafka.adoc

@@ -2471,6 +2471,37 @@ In addition, you can configure the serializer and deserializer by using the foll
 Starting with version 2.2, the type information headers (if added by the serializer) are removed by the deserializer.
 You can revert to the previous behavior by setting the `removeTypeHeaders` property to `false`, either directly on the deserializer or with the configuration property described earlier.
 
+When constructing the serializer/deserializer programmatically for use in the producer/consumer factory, since version 2.3, you can use the fluent API, which simplifies configuration.
+
+The following example assumes you are using Spring Boot:
+
+====
+[source, java]
+----
+@Bean
+public DefaultKafkaProducerFactory pf(KafkaProperties properties) {
+    Map<String, Object> props = properties.buildProducerProperties();
+    DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(props,
+        new JsonSerializer<>(MyKeyType.class)
+            .forKeys()
+            .noTypeInfo(),
+        new JsonSerializer<>(MyValueType.class)
+            .noTypeInfo());
+}
+
+@Bean
+public DefaultKafkaConsumerFactory pf(KafkaProperties properties) {
+    Map<String, Object> props = properties.buildConsumerProperties();
+    DefaultKafkaConsumerFactory pf = new DefaultKafkaConsumerFactory(props,
+        new JsonDeserializer<>(MyKeyType.class)
+            .forKeys()
+            .ignoreTypeHeaders(),
+        new JsonSerializer<>(MyValueType.class)
+            .ignoreTypeHeaders());
+}
+----
+====
+
 [[serdes-mapping-types]]
 ===== Mapping Types
 

src/reference/asciidoc/streams.adoc

@@ -121,7 +121,8 @@ private StreamsBuilderFactoryBean myKStreamBuilderFactoryBean;
 ----
 ====
 
-==== JSON Serialization and Deserialization
+[[serde]]
+==== Streams JSON Serialization and Deserialization
 
 For serializing and deserializing data when reading or writing to topics or state stores in JSON format, Spring Kafka provides a `JsonSerde` implementation that uses JSON, delegating to the `JsonSerializer` and `JsonDeserializer` described in <<serdes>>.
 The `JsonSerde` implementation provides the same configuration options through its constructor (target type or `ObjectMapper`).
@@ -134,7 +135,19 @@ stream.through(Serdes.Integer(), new JsonSerde<>(Cat.class), "cats");
 ----
 ====
 
-IMPORTANT: Since Kafka Streams do not support headers, the `addTypeInfo` property on the `JsonSerializer` is ignored.
+When constructing the serializer/deserializer programmatically for use in the producer/consumer factory, since version 2.3, you can use the fluent API, which simplifies configuration.
+
+====
+[source, java]
+----
+stream.through(new JsonSerde<>(MyKeyType.class)
+        .forKeys()
+        .noTypeInfo(),
+    new JsonSerde<>(MyValueType.class)
+        .noTypeInfo(),
+    "myTypes");
+----
+====
 
 ==== Using `KafkaStreamsBrancher`
 

src/reference/asciidoc/whats-new.adoc

@@ -78,6 +78,9 @@ A `ByteArrayJsonMessageConverter` has been provided as well as a new super class
 Also, a `StringOrBytesSerializer` is now available; it can serialize `byte[]`, `Bytes` and `String` values in `ProducerRecord` s.
 See <<messaging-message-conversion>> for more information.
 
+The `JsonSerializer`, `JsonDeserializer` and `JsonSerde` now have fluent APIs to make programmatic configuration simpler.
+See the javadocs, <<serdes>>, and <<serde>> for more informaion.
+
 ==== ReplyingKafkaTemplate
 
 When a reply times out, the future is completed exceptionally with a `KafkaReplyTimeoutException` instead of a `KafkaException`.