我试图使用Long作为消息密钥的类型,但我明白了
Exception in thread "kafka_stream_app-f236aaca-3f90-469d-9d32-20ff694806ff-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Failed to deserialize key for record. topic=test, partition=0, offset=0
at org.apache.kafka.streams.processor.internals.SourceNodeRecordDeserializer.deserialize(SourceNodeRecordDeserializer.java:38)
at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:84)
at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)
at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:474)
at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:642)
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:548)
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:519)
Caused by: org.apache.kafka.common.errors.SerializationException: Size of data received by LongDeserializer is not 8
我检查了 data.length
是 7
.
在streamsConfiguration我已经设置
streamsConfiguration.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.Long().getClass().getName());
我用
KStream<Long, GenericRecord> stream = builder.stream(topic);
我尝试通过一个简单的应用程序和 kafka-avro-console-producer
发送消息:
/opt/confluent-3.3.0/bin/kafka-avro-console-producer \
--broker-list localhost:9092 \
--topic test \
--property key.separator=, \
--property parse.key=true \
--property key.schema='{"type":"long"}' \
--property value.schema='{"type":"string"}' \
--property schema.registry.url=http://localhost:8081
与消息
123,"293"
使用 kafka-avro-console-consumer
我可以使用该消息并查看(使用 --property print.key=true
,发送的密钥正确 123
)
知道在解码消息时会出现什么问题吗?
1 回答
因为您正在使用
kafka-avro-console-producer
,所以密钥不是作为普通Long
序列化,而是作为Avro类型 . 因此,您需要使用与写入路径上使用的相同模式的相应Avro Serde(即'{"type":"long"}"
) .此外,您的返回类型不是
Long
,而是Avro类型 .