From 1f3d7cdb3f60f8a3b96cdedefe77268f278b15d6 Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Wed, 18 Apr 2018 10:12:18 -0400
Subject: [PATCH] GH-652: JsonSerializer add null check for Headers

Fixes: https://github.com/spring-projects/spring-kafka/issues/652

Kafka Streams `ChangedSerializer` calls the `Serde` with `null` in `headers`.

**cherry-pick to 2.1.x, 2.0.x, 1.3.x**

(cherry picked from commit ad28b91)
---
 .../kafka/support/serializer/JsonSerializer.java                | 2 +-
 src/reference/asciidoc/kafka.adoc                               | 2 +-
 src/reference/asciidoc/streams.adoc                             | 2 ++
 3 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java
index 2b829ff0..bc2cc465 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java
@@ -130,7 +130,7 @@ public class JsonSerializer<T> implements ExtendedSerializer<T> {
 
 	@Override
 	public byte[] serialize(String topic, Headers headers, T data) {
-		if (this.addTypeInfo) {
+		if (this.addTypeInfo && headers != null) {
 			this.typeMapper.fromJavaType(this.objectMapper.constructType(data.getClass()), headers);
 		}
 		return serialize(topic, data);
diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc
index 74729a06..1edb0e3f 100644
--- a/src/reference/asciidoc/kafka.adoc
+++ b/src/reference/asciidoc/kafka.adoc
@@ -1433,7 +1433,7 @@ You can also extend them to implement some particular configuration logic in the
 Starting with _version 2.1_, type information can be conveyed in record `Headers`, allowing the handling of multiple types.
 In addition, the serializer/deserializer can be configured using Kafka properties.
 
--  `JsonSerializer.ADD_TYPE_INFO_HEADERS` (default `true`); set to `false` to disable this feature.
+- `JsonSerializer.ADD_TYPE_INFO_HEADERS` (default `true`); set to `false` to disable this feature on the `JsonSerializer` (sets the `addTypeInfo` property).
 - `JsonDeserializer.KEY_DEFAULT_TYPE`; fallback type for deserialization of keys if no header information is present.
 - `JsonDeserializer.VALUE_DEFAULT_TYPE`; fallback type for deserialization of values if no header information is present.
 - `JsonDeserializer.TRUSTED_PACKAGES` (default `java.util`, `java.lang`); comma-delimited list of package patterns allowed for deserialization; `*` means deserialize all.
diff --git a/src/reference/asciidoc/streams.adoc b/src/reference/asciidoc/streams.adoc
index 60f63b6e..66b16d6f 100644
--- a/src/reference/asciidoc/streams.adoc
+++ b/src/reference/asciidoc/streams.adoc
@@ -111,6 +111,8 @@ In the following example we use the `JsonSerde` to serialize and deserialize the
 stream.through(Serdes.Integer(), new JsonSerde<>(Foo.class), "foos");
 ----
 
+IMPORTANT: Since Kafka Streams do not support headers, the `addTypeInfo` property on the `JsonSerializer` is ignored.
+
 ==== Configuration
 
 To configure the Kafka Streams environment, the `StreamsBuilderFactoryBean` requires a `Map` of particular properties or a `StreamsConfig` instance.