sobychacko
commented
2 years ago @aliaksandr-pleski Thanks for this report. After looking at this issue a bit further, I think we can resolve it in a couple of ways.
The reason you are getting this error is due to the fact that __typeid__ header is not set properly by your function(). This is supposed to be set by the JsonSerializer in Spring Kafka. Your streamFunction function is a Kafka Streams function and therefore by default it uses the JsonSerializer and adds the __typdid__ header as ClassB on the outbound. This is then passed down to function's input, but function is based on regular Kafka binder and it uses message conversion behind the scenes (on both input and output) and does not use JsonSerializer (or native Kafka serialization). When function publishes the data (as type ClassC), it doesn't properly update the type id header (since no JsonSerializer is involved). It simply passes down the same __typeid__ value it received on the input (which is ClassB) and when it reaches the next function (consumer1), you run into that ClassCastException you are getting.
You can get around to this problem in two ways. The first one is pretty trivial while the second one needs some configuration changes.
-
you can add the
__typeid__header programmatically when sending the messages from thefunctionmethod. Below is an example.@bean public Function<ClassB, Message<ClassC>> function() { return classB -> { final Map<String, Object> headers = new HashMap<>(); headers.put(KafkaHeaders.MESSAGE_KEY, classB.b); headers.put(DefaultJackson2JavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME, ClassC.class.getCanonicalName()); return MessageBuilder.createMessage( new ClassC(Integer.parseInt(classB.b)), new MessageHeaders(headers) ); }; }Note the explicit addition of
DEFAULT_CLASSID_FIELD_NAMEheader. -
The other option is to force the
functionmethod to use native encoding on the outbound so that it uses Kafka's serialization using theJsonSerializerfrom Spring Kafka. This way, the propertypeidheader is added when the record is published. Here are the relevant configuration changes needed.spring: cloud: function: definition: supplier;streamFunction;function;consumer1 stream: kafka: bindings: function-out-0: producer: configuration: value.serializer: org.springframework.kafka.support.serializer.JsonSerializer .... bindings: ... function-in-0: binder: kafka destination: scs-cc-streamFunction function-out-0: binder: kafka destination: scs-cc-consumer producer: useNativeEncoding: true .... ....I verified that both solutions work. Please let us know if they work on your end with these solutions.
aliaksandr-pleski
Hi everybody,
I have a
spring-cloud-streamapplication. Spring Boot is 2.6.2 and Spring Cloud is 2021.0.0. Working with Kafka and Kafka Streams.The app has the following data pipeline:
There are 3 consumers that expect
ClassCobjects:Consumers 2 and 3 work fine:
Where Consumer 1 gets following
ClassCastException:Seems like this behavior isn't correct. Probably it could be related to the fact that there is one regular Kafka function between 2 Kafka Streams functions.
You can find sample application here: https://github.com/aliaksandr-pleski/spring-cloud-stream-kafka-class-cast-exception