如何用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
相关产品推荐
相关产品推荐

