Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-652: JsonSerializer add null check for Headers

Browse Source 
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**
This commit is contained in:
Gary Russell
2018-04-18 10:12:18 -04:00
committed by Artem Bilan
parent fd694470ad
commit ad28b9183b
3 changed files with 4 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -130,7 +130,7 @@ public class JsonSerializer<T> implements ExtendedSerializer<T> {
@Override @Override
public byte[] serialize(String topic, Headers headers, T data) { 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); this.typeMapper.fromJavaType(this.objectMapper.constructType(data.getClass()), headers);
} }
return serialize(topic, data); return serialize(topic, data);

2
src/reference/asciidoc/kafka.adoc
View File

@@ -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. 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. 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.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.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. - `JsonDeserializer.TRUSTED_PACKAGES` (default `java.util`, `java.lang`); comma-delimited list of package patterns allowed for deserialization; `*` means deserialize all.

2
src/reference/asciidoc/streams.adoc
View File

@@ -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"); 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 ==== Configuration
To configure the Kafka Streams environment, the `StreamsBuilderFactoryBean` requires a `Map` of particular properties or a `StreamsConfig` instance. To configure the Kafka Streams environment, the `StreamsBuilderFactoryBean` requires a `Map` of particular properties or a `StreamsConfig` instance.