Json (de)serialization configuration improvements

[nivertius]

This changes improve ability to configure JsonSerde as well as JsonSerializer and JsonDeserializer.

First, allow TypeReference and JavaType when constructing JsonSerde. This can be helpful when using Serde for deserialization, but there are container types involved.

Second, allow forcing type format when serializing. This is actually a problem that cannot be currently solved using JsonSerializer: Lets assume theres a serialized type A, which is polymorphic and there is a list of objects of type A, possibly subtypes. Currently, the mechanism finds typed jackson serializer only for raw list, and serializes contents without configuration, so as simple beans, skipping properties that should be added. This can be seen in test JsonSerializationTests#testDeserializerTypeForcedType which fails when not providing forced type to JsonSerializer.

Lastly, this pull request adds ability to create copies of JsonSerde, JsonSerializer and JsonDeserializer with changed forced and default type. This change allows to create a single bean with JsonSerde<?> as actual type, configure its typeMapper, and other behaviors, and use this bean as a base for all the serdes used. Unfortunately, this couldn't be done with the fluent API, because generic parameter of type changes.

There is a backward incompatibility in this changes - previously, when creating JsonSerde with specified class or type reference, this was only passed to deserialized as default type, and now is also passed to serializer as a forced type. This means that if user created JsonSerde instance with specified target type, but used its serializer to serialize unrelated type, now this will fail, instead of serializing the data without type information. I don't think this is problematic: Either user didn't specify a type and serialized types based on header data or object mapper configuration, or they specified type and serialized objects of this type.

In any case, I think specifing type for JsonSerde and using its serializer to serialize unrelated types could very well be a invalid configuration and probably should produce an error. If you don't agree, I can add backward compatiblity flag.

[garyrussell]

Thanks for the contribution, but this adds a lot of complexity as well as, perhaps, providing too many choices for how to configure the deserializer.

Given the new mechanism to determine the type by calling a method I am not sure this really adds much, if any, value.

Furthermore, when using the method mechanism, the polymorphism problem is solved with a single deserializer rather than making clones with a different JavaType (and it doesn't have to be a bean, Kafka can instantiate it via properties).

@artembilan WDYT?

[artembilan]

Yep! I think method reference is much cleaner solution with less code to support.

[nivertius]

Thank you for looking at the code. Even with the method to determine target type when deserialization occurs, I still think there are things to consider here:

• There's still problem with serializer getting data with no target type, and it incorrectly writes collections of polymorphic objects as beans. I don't think there's a way to fix that with type mapper, as it is only used in serializer to add headers. This is most important part of this pull request - everything else is just some overloads and convenience methods.
• Creating single JsonSerde and to use to (de)serialize multiple entities and using it when configuring streams would kinda of defeat the type system - created object should be typed as JsonSerde<?> and needs to be unchecked casted into speciailized types when used in Materialized, Consumed an so on. As for copies with specified types, they are not expensive (they share ObjectMapper, which can be heavy).

[garyrussell]

I don't think there's a way to fix that with type mapper,

I don't follow - the type mapper is ignored when using a method call to determine the type.

Starting with version 2.5, you can now configure the deserializer, via properties, to invoke a method to determine the target type. If present, this will override any of the other techniques discussed above.

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

Lines 468 to 473 in b2f8be6

	if (this.typeResolver != null) {
	javaType = this.typeResolver.resolveType(topic, data, headers);
	}
	if (javaType == null && this.typeMapper.getTypePrecedence().equals(TypePrecedence.TYPE_ID)) {
	javaType = this.typeMapper.toJavaType(headers);
	}

You have a good point about generics when using the Serde with streams. I think a simpler solution for that is to add

public <A> JsonSerde<A> forClass(Class<A> clazz) {
	return (JsonSerde<A>) this;
}

to the Serde.

[nivertius]

I don't follow - the type mapper is ignored when using a method call to determine the type.

I was talking about JsonSerializer. My observation about typeMapper was unrelated to our conversation about selecting type for deserialization, but it got mixed somehow. 😊

You have a good point about generics when using the Serde with streams. I think a simpler solution for that is to add (...) to the Serde.

I accept your point about this being simpler, and I won't argue more - if you don't want that kind of copying, I can just replace last commit with one that does the cast. I do understand that maintaining this new code might fall on any of you, and nobody wants to maintain unnecessary code.
But my thoughts about that was that in some cases JsonSerde can be tied to the type as much as possible: beside generic parameter, there's also configuration for ObjectWriter and ObjectReader in serializer and deserializer with the type, and there might be different trusted packages and so on.

[garyrussell]

OK; I get it now.

However, I don't see how it will work when setting the "forced" type via properties since it's a simple class, not a TypeReference.

If I am missing something, can you add a test for that use case?

[nivertius]

You are right, from config this might be useless. I thought that com.fasterxml.jackson.databind.type.TypeFactory#constructType(java.lang.reflect.Type) works with parameterized types, as the name suggest, but it actually only loads a raw type.

In this case I can either try to extend the type resolution for config to parse parameterized types (which might be tricky), or I can remove setting forced type by config property, what do you think would be better?

[garyrussell]

I think just removing it is fine - that would be consistent with the deserializer only getting a TypeReference via a constructor.

For these "advanced" configurations, it's better to construct it as a bean and inject it into the producer factory.

[nivertius]

Very well, I removed the property configuration, will push with next change:

In the light of above problem, what's your final decision on copyWithType. Should I remove them and replace with casts that you suggested? Because that will not affect serializers and deserializers as user might expect i.e.
new JsonSerde<>(new TypeReference<List<Bean>> {})
will be different in behavior than
new JsonSerde().forClass((Class<List<Bean>>) List.class)

I might also suggest that the argument in forClass method is actually not necessary and might be an obstacle. We could do just

public <T> JsonSerde<T> adjustType() { 
   return (JsonSerde<T>) this; 
}

In most cases, the generic parameter would be just inferred, so Materialized.with(JsonSerde.adjustType(), JsonSerde.adjustType()) would just work. Worst case scenario if type somewhat cannot be inferred, or if user really wants to see the types, they can use type witness: Materialized.with(JsonSerde.<KeyBean>adjustType(), JsonSerde.<ValueBean>adjustType(). This would also suggest to user that this is not something that reconfigures the serde, but something that just changes type of expression.

[garyrussell]

I am ok with keeping your copy... methods instead of adding adjustType.

[garyrussell]

Thanks for the contribution (and discussion),

[nivertius]

Thank you!