Spring Kafka Streams：缺少类型信息

Question

我已经使用 Kafka Streams 设置了一个简单的 Spring 引导应用程序。 Spring boot 正在使用 AutoConfiguration for Kafka 创建工厂。该消息是 Json 消息，没有消息密钥。使用以下配置会导致 headers 消息中没有类型信息。有任何输入吗？

spring.kafka.streams.properties.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.kafka.streams.properties.default.value.serde=org.springframework.kafka.support.serializer.JsonSerde
spring.kafka.consumer.properties.spring.json.key.default.type=java.lang.String
spring.kafka.consumer.properties.spring.json.value.default.type=com.abc.xyz.Person

错误

java.lang.IllegalStateException: No type information in headers and no default type provided
at org.springframework.util.Assert.state(Assert.java:73) ~[spring-core-5.1.5.RELEASE.jar:5.1.5.RELEASE]
at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:353) ~[spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:63) ~[kafka-streams-2.0.1.jar:?]
at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66) [kafka-streams-2.0.1.jar:?]
at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:97) [kafka-streams-2.0.1.jar:?]
at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117) [kafka-streams-2.0.1.jar:?]
at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:677) [kafka-streams-2.0.1.jar:?]
at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:943) [kafka-streams-2.0.1.jar:?]
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:831) [kafka-streams-2.0.1.jar:?]
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767) [kafka-streams-2.0.1.jar:?]
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736) [kafka-streams-2.0.1.jar:?]

Answer 1

Streams 不使用常规 consumer 属性。

某些文档是否让您认为在使用流时需要设置属性？如果是这样，请打开 GitHub 问题以便我们修复它。

是

spring.kafka.streams.properties.spring.json.value.default.type=com.abc.xyz.Person

使用流时。

您不需要 key.default.type 因为您没有 JSON 密钥 serde。

Spring Kafka Streams：缺少类型信息

Spring Kafka Streams: Type Information Missing

spring-kafka