无法通过kafka avro控制台读取avro消息消费者（最终目标通过spark streaming读取）

Question

（最终目标）在尝试是否最终可以从Confluent平台读取avro数据usng spark stream之前，如这里所述：将spark结构化流与Confluent Schema Registry集成

我要验证是否可以使用以下命令来读取它们：

$ kafka-avro-console-consumer \
> --topic my-topic-produced-using-file-pulse-xml \
> --from-beginning \
> --bootstrap-server localhost:9092 \
> --property schema.registry.url=http://localhost:8081

我收到这个错误消息，未知的魔法字节

Processed a total of 1 messages
[2020-09-10 12:59:54,795] ERROR Unknown error when running consumer:  (kafka.tools.ConsoleConsumer$:76)
org.apache.kafka.common.errors.SerializationException: Unknown magic byte!
[2020-09-10 12:59:54,795] ERROR Unknown error when running consumer:  (kafka.tools.ConsoleConsumer$:76)
org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

注意，可以这样读取消息（使用console consumer而不是avro console consumer）：

kafka-console-consumer \
--bootstrap-server localhost:9092 --group my-group-console \
--from-beginning \
--topic my-topic-produced-using-file-pulse-xml

该消息是使用conFluent connect file-Pulch（1.5.2）读取xml文件（Stremoth/kafka-Connect-file-Pulp）生成的

请在此处提供帮助：我是否错误地使用了kafka-avro-console-消费者？我尝试了此处描述的“反序列化器”属性选项：https://stackoverflow.com/a/57703102/4582240，没有帮助

我不想勇敢地开始火花流读取数据。

我使用的文件pulse 1.5.2属性如下所示，于2020年9月11日添加完成。

name=connect-file-pulse-xml
connector.class=io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector
topic= my-topic-produced-using-file-pulse-xml
tasks.max=1

# File types
fs.scan.filters=io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter
file.filter.regex.pattern=.*\\.xml$
task.reader.class=io.streamthoughts.kafka.connect.filepulse.reader.XMLFileInputReader
force.array.on.fields=sometagNameInXml

# File scanning
fs.cleanup.policy.class=io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy
fs.scanner.class=io.streamthoughts.kafka.connect.filepulse.scanner.local.LocalFSDirectoryWalker

fs.scan.directory.path=/tmp/kafka-connect/xml/
fs.scan.interval.ms=10000

# Internal Reporting
internal.kafka.reporter.bootstrap.servers=localhost:9092
internal.kafka.reporter.id=connect-file-pulse-xml
internal.kafka.reporter.topic=connect-file-pulse-status

# Track file by name
offset.strategy=name

Answer 1

如果您向消费者获取未知魔法字节，则生产者没有使用ConFluent AvroSerializer，并且可能推送了不使用Schema注册表的Avro数据。

如果看不到生产者代码或使用和检查二进制格式的数据，很难知道是哪种情况。

消息是使用汇合连接文件脉冲生成的

您是否使用了值。AvroConverter类的转换器？