You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何用kafka-console-consumer打印Avro消息Schema ID?配置报错求助

问题

我希望从kafka-console-consumer中打印接收到的Avro消息的schema id。在基于Docker的Confluent Platform(cp-all-in-one环境)中尝试了以下命令:

kafka-console-consumer --bootstrap-server localhost:9092 --topic topic_name \
--property print.schema.ids=true --property \
 schema.registry.url="http://schema-registry:8081" \
--value-deserializer io.confluent.kafka.serializers.KafkaAvroDeserializer

但出现异常:

Exception in thread "main" org.apache.kafka.common.config.ConfigException: Missing required configuration "schema.registry.url" which has no default value.
        at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:515)
        at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:505)
        at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:112)
        at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:132)
        at io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig.<init>(AbstractKafkaSchemaSerDeConfig.java:295)
        at io.confluent.kafka.serializers.KafkaAvroDeserializerConfig.<init>(KafkaAvroDeserializerConfig.java:58)
        at io.confluent.kafka.serializers.KafkaAvroDeserializer.configure(KafkaAvroDeserializer.java:48)
        at kafka.tools.DefaultMessageFormatter.getDeserializerProperty(ConsoleConsumer.scala:609)
        at kafka.tools.DefaultMessageFormatter.$anonfun$configure$22(ConsoleConsumer.scala:515)
        at kafka.tools.DefaultMessageFormatter.configure(ConsoleConsumer.scala:615)
        at kafka.tools.ConsoleConsumer$ConsumerConfig.<init>(ConsoleConsumer.scala:325)
        at kafka.tools.ConsoleConsumer$.main(ConsoleConsumer.scala:51)
        at kafka.tools.ConsoleConsumer.main(ConsoleConsumer.scala)

请问遗漏了什么配置?是否有其他可用的参数可以替代?

解决方案

问题出在反序列化器的配置参数传递方式:用--property传递的schema.registry.url是给消息格式化器的,而非KafkaAvroDeserializer本身。反序列化器需要的配置必须通过--consumer-property传递。

修正后的命令如下:

kafka-console-consumer --bootstrap-server localhost:9092 --topic topic_name \
--property print.schema.ids=true \
--consumer-property schema.registry.url="http://schema-registry:8081" \
--value-deserializer io.confluent.kafka.serializers.KafkaAvroDeserializer

另外,也可以直接指定Confluent的Avro专用格式化器,无需显式声明反序列化器,同样能正确打印Schema ID:

kafka-console-consumer --bootstrap-server localhost:9092 --topic topic_name \
--formatter io.confluent.kafka.formatter.AvroMessageFormatter \
--property print.schema.ids=true \
--property schema.registry.url="http://schema-registry:8081"

内容的提问来源于stack exchange,提问作者Marinos An

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.06 15:22:21