Cannot use JsonSerde for both key and value serialization in a stream

[elliotkennedy]

Am I correct in thinking I should be able to use a JsonSerde for both the key and value in a stream?

Deserialization fails when using JsonSerde to serialize both the key and value of a topic.

The key and value are both given the same header value here:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/support/converter/DefaultJackson2JavaTypeMapper.java

Line 177 in 31431fa

addHeader(headers, getClassIdFieldName(), javaType.getRawClass());

So when deserializing, the key and value header chosen is simply the first header in the list. This results in attempting to deserialize the value as the key type.

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/support/converter/AbstractJavaTypeMapper.java

Line 116 in 31431fa

classId = new String(headerValue.value(), StandardCharsets.UTF_8);

I've created a test to demonstrate the issue here:

elliotkennedy@ff15715

I might be able to attempt a fix using different headers soon if people think that would be appropriate?

[artembilan]

That's definitely good catch, indeed!

I think as a workaround we have to use an extension of the DefaultJackson2JavaTypeMapper and store key or value under different header name overriding the:

public String getClassIdFieldName() {
	return DEFAULT_CLASSID_FIELD_NAME;
}

To fix the issue I guess we need to take into account isKey argument in the:

    /**
     * Configure this class.
     * @param configs configs in key/value pairs
     * @param isKey whether is for key or value
     */
    void configure(Map<String, ?> configs, boolean isKey);

And configure different classIdFieldName.

In addition that looks like we have to require isKey as a ctor argument for the cases like:

Produced.with(Serdes.Integer(), new JsonSerde<Foo>() {})

Feel free to go ahead with the Contribution.

Thank you for the report!

[elliotkennedy]

Thanks @artembilan, I've attempted to configure the Serdes - please let me know what you think.

The problem is that when you run the tests now, void configure(Map<String, ?> configs, boolean isKey); is never called on the JsonDeserializer. This feels like a bug in Kafka itself - I'd expect the topology builder to configure the serde at some point. What do you think?

Could you clarify on why we need isKey as a constructor arg on JsonSerde too please? I don't quite follow.

Thanks again.

[artembilan]

I’ll check tomorrow how and why it isn’t called. We need to check for isKey to populate a different header for the key type.

[artembilan]

@elliotkennedy ,

We have KafkaMessageListenerContainerTests.testJsonSerDeConfiguredType() with the code like:

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);

and when I place a breakpoint in the JsonDeserializer.configure() I see how it is called.

So, I'm not sure what's going on in your case...

Also see my comment above:

In addition that looks like we have to require isKey as a ctor argument for the cases like:

  Produced.with(Serdes.Integer(), new JsonSerde<Foo>() {})

When we don't use properties, but instantiate SerDe classes ourselves we have to have some other way to configure the situation. In this case it is indeed that configure() is called. If you provide an instance, you have to ensure its configure yourself. Kafka doesn't assumes a mutation of the external objects.

[elliotkennedy]

Cool, thanks for looking @artembilan - I didn't realise about configuring SerDe yourself when you instantiate it - that's what was happening in my case. I'll get a PR together now.