kafka streams globalktable在tombstone-null值-records上引发反序列化异常

inb24sb2  于 2021-06-30  发布在  Java
关注(0)|答案(1)|浏览(441)

我有一个基于springcloudstream的kafka streams应用程序,我将一个全局ktable绑定到一个紧凑的主题。当我将tombstone记录推送到主题(具有null值的非空键)时,我的kafka streams应用程序失败,出现反序列化异常。失败是因为我的反序列化程序不处理空记录。
从文档中,我认为globalktable甚至不会“看到”空值记录。不是这样吗?我需要在反序列化程序中处理空记录吗?

org.apache.kafka.common.errors.SerializationException: Unable to deserialize
Caused by: java.lang.IllegalArgumentException: argument "src" is null
    at com.fasterxml.jackson.databind.ObjectMapper._assertNotNull(ObjectMapper.java:4693)
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3511)
    at common.domain.serdes.MySerde$MyDeserializer.deserialize(MySerde.java:47)
    at common.domain.serdes.MySerde$MyDeserializer.deserialize(MySerde.java:39)
    at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:63)
    at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66)
    at org.apache.kafka.streams.processor.internals.GlobalStateUpdateTask.update(GlobalStateUpdateTask.java:91)
    at org.apache.kafka.streams.processor.internals.GlobalStreamThread$StateConsumer.pollAndUpdate(GlobalStreamThread.java:240)
    at org.apache.kafka.streams.processor.internals.GlobalStreamThread.run(GlobalStreamThread.java:289)
3ks5zfa0

3ks5zfa01#

对;你得检查一下 null 然后回来 null . 查看任何标准反序列化程序。
不像 KafkaConsumerFetcher (检查 null 在打电话之前) kafka-streams 无条件地说。看见

at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:63)

相关问题