Comments (5)
@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 thefunction
method. 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_NAME
header.
- The other option is to force the
function
method to use native encoding on the outbound so that it uses Kafka's serialization using theJsonSerializer
from Spring Kafka. This way, the propertypeid
header 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.
from spring-cloud-stream-binder-kafka.
Since the regular binder uses message conversion by default from Spring, it doesn't go through the native Kafka serializer, and that code for specifically adding that header is part of the JsonSerializer
class. Therefore, this is the expected behavior.
from spring-cloud-stream-binder-kafka.
@sobychacko thanks for the quick response!
Both options are working for me. I personally find option 2 more preferable.
The fact that the regular Kafka function
doesn't properly update __typeid__
header and just proxies existing one - is that the expected behavior?
from spring-cloud-stream-binder-kafka.
Closing the issue now, feel free to re-open if you find something else related to this.
from spring-cloud-stream-binder-kafka.
Got it, thanks!
from spring-cloud-stream-binder-kafka.
Related Issues (20)
- Live producer threads and dynamic-destination-cache-size property HOT 2
- How to wait for future inside Kafka Stream map()? HOT 2
- DefaultErrorHandler not exists HOT 4
- NOT_ENOUGH_REPLICAS Errors After Upgrading The Spring Boot And Spring Cloud Version HOT 5
- KafkaBinderMetrics prevent JVM shutdown due to not shutting down scheduler thread pool! HOT 8
- Kafka bindings from previous test class remains active in the next test class HOT 2
- Kafka Streams Serdes having nested generic doesn't work HOT 3
- Support for DLQ in batch mode HOT 1
- spring.cloud.stream.bindings.process-in-0.consumer.concurrency not working in case of UpperCase letter in process name HOT 1
- ListenerContainerWithDlqAndRetryCustomizer template points to localhost:9092 HOT 5
- Looking for a sample with AWS MSK IAM
- [Feature Request] Allow to accept a list of customizers instead of a unique one for KafkaMessageChannelBinder HOT 1
- spring-kafka-test is compile (not test) dependency of spring-cloud-stream-binder-kafka-streams:3.2.3 and 3.2.4
- Batch mode is not working HOT 1
- Unknown Autoconfigured KafkaAvroSerializer and KafkaAvroDeSerializers HOT 2
- Spring Cloud stream binder kafka DLQ Handling for Batching doesn't work HOT 2
- Enable DLQ with destinations as pattern
- Fetch message timestamp
- Health Check for Binders Always Shows 'UNKNOWN' Status Due to Empty binderHealthContributors Map (Missing Bean Configuration for ReactorKafkaBinderHealthIndicator?) HOT 2
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from spring-cloud-stream-binder-kafka.