我正在尝试根据此kafka documentation实现此连接。
我不知道为什么这个连接不起作用...
首先，我通过了所有值(value)观。
在这里，它试图强制泛型为对象类型，这是非常错误的。
如果不使用序列化选项加入，我会收到以下运行时异常:
Exception in thread "StreamAPP-stream-event-b3dc5fff-abee-4fa0-92f9-e1690f8fd152-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: ClassCastException while producing data to topic StreamAPP-stream-event-KSTREAM-KEY-SELECT-0000000025-repartition. A serializer (key: org.apache.kafka.common.serialization.ByteArraySerializer / value: org.apache.kafka.common.serialization.ByteArraySerializer) is not compatible to the actual key or value type (key type: br.com.calebebrim.kafka.entities.stream.sharing.registry.StreamRegistryKey / value type: br.com.calebebrim.kafka.entities.stream.sharing.stream.Event). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters (for example if using the DSL,
#to(String topic, Produced<K, V> produced)withProduced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class)))
谁能帮我？
谢谢!
最佳答案
解决了，
我发现联接操作无法转换数据。
所以，我只是在像这样之前应用了mapValues:
关于java - Kafka DSL Kstream-> Ktable Join-连接的序列化编译错误，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/65618953/