我正在尝试使用springcloudkafkastreams绑定器来使用来自该主题的avro消息,但无法修复这个classcast异常。
这是我的密码:
@Bean
public Consumer<KStream<EventKey, Event>> process(){
return input -> {
input.peek(((key, value) -> logger.info("key value: "+ key.toString()+" value: "+value.toString())));
logger.info("Received:" + input);
};
}
@Bean
public Serde<EventKey> avroInSerde(){
final SpecificAvroSerde<EventKey> avroInSerde = new SpecificAvroSerde<>();
Map<String, Object> serdeProperties = new HashMap<>();
return avroInSerde;
}
@Bean
public Serde<Event> avroOutSerde(){
final SpecificAvroSerde<Event> avroOutSerde = new SpecificAvroSerde<>();
return avroOutSerde;
}
活页夹:
spring:
application:
name: ${applicaton-name}
cloud:
stream:
function:
definition: process
bindings:
process-in-0:
destination: ${input-topic-name}
contentType: application/Avro
process-out-0:
destination: ${enriched-topic-name}
contentType: application/Avro
binding-retry-interval: 30
kafka:
streams:
binder:
brokers: ${kafka-broker}
application-id: ${consumer-group-name}
auto-create-topics: false
auto-add-partitions: false
configuration:
processing.guarantee: at_least_once
auto.offset.reset: earliest
schema.registry.url: ${kafka-schema-registry}
auto-register-schema: false
security.protocol: SSL
useNativeEncoding: true
specific.avro.reader: true
错误:
Note that although incorrect Serdes are a common cause of error, the cast exception might have another cause (in user code, for example). For example, if a processor wires in a store, but casts the generics incorrectly, a class cast exception could be raised during processing, but the cause would not be wrong Serdes.
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:146)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:236)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:216)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:168)
at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:96)
at org.apache.kafka.streams.processor.internals.StreamTask.lambda$process$1(StreamTask.java:679)
at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:836)
at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:679)
at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:1033)
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:690)
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:551)
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:510)
Caused by: java.lang.ClassCastException: EventKey cannot be cast to EventKey
at org.apache.kafka.streams.kstream.internals.KStreamPeek$KStreamPeekProcessor.process(KStreamPeek.java:42)
at org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$process$2(ProcessorNode.java:142)
at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:836)
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
我尝试了这两种方法在这个链接中提到https://spring.io/blog/2019/12/04/stream-processing-with-spring-cloud-stream-and-apache-kafka-streams-part-3-data-deserialization-and-serialization 但运气不好
我错过什么了吗?。
1条答案
按热度按时间shstlldc1#
原因:java.lang.classcastexception:eventkey不能强制转换为eventkey
这可能是一个类装入器问题;反序列化程序和使用者bean由不同的类装入器装入,您使用的是spring devtools吗?
使用springkafka,可以通过显式创建使用者工厂并将反序列化器注入其中来避免这种情况。
使用springcloudstream(从版本3.0.6开始),您可以提供
ClientFactoryCustomizer
bean并注入反序列化程序示例(定义为@Bean
s、 因此它们使用相同的类装入器)。或者,停止使用devtools。