通过 kafka-avro-console-producer 生成 Avro 序列化对象的问题

Question

我正在生成一条消息，方法是使用 kafka-avro-console-producer 二进制通过做：

kafka-avro-console-producer --broker-list broker:9092 --topic example-topic --property schema.registry.url='http://schema-registry:8081 --property value.schema='{"type": "record","name": "test","fields": [{"name": "before", "type": ["null", {"type": "record", "name": "columns", "fields":[{"name": "name", "type": "string"}]}],"default": "null"},{"name": "after", "type": ["null", "columns"],"default": "null"}]}'
{"before": null,"after": {"name": "John"}}'

发送以下消息：

{"before": null,"after": {"name": "John"}}

并应用以下 Avro 架构：

{
    "type": "record",
    "name": "test",
    "fields": [{
        "name": "before",
        "type": ["null", {
            "type": "record",
            "name": "columns",
            "fields": [{
                "name": "name",
                "type": "string"
            }]
        }],
        "default": "null"
    }, {
        "name": "after",
        "type": ["null", "columns"],
        "default": "null"
    }]
}

我得到的错误 return 如下：

Caused by: org.apache.avro.AvroTypeException: Unknown union branch name
    at org.apache.avro.io.JsonDecoder.readIndex(JsonDecoder.java:445)
    at org.apache.avro.io.ResolvingDecoder.doAction(ResolvingDecoder.java:290)
    at org.apache.avro.io.parsing.Parser.advance(Parser.java:88)
    at org.apache.avro.io.ResolvingDecoder.readIndex(ResolvingDecoder.java:267)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:178)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:152)
    at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:240)
    at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:230)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:174)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:152)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:144)
    at io.confluent.kafka.formatter.AvroMessageReader.jsonToAvro(AvroMessageReader.java:213)
    at io.confluent.kafka.formatter.AvroMessageReader.readMessage(AvroMessageReader.java:180)
    at kafka.tools.ConsoleProducer$.main(ConsoleProducer.scala:55)
    at kafka.tools.ConsoleProducer.main(ConsoleProducer.scala)

对于那些愿意深入兔子洞的人，我正在使用 Oracle Golden Gate Big Oracle Golden Gate 和 Apache Kafka 之间的集成数据连接器。我目前在使用此处描述的等效模型时遇到问题：

https://www.ateam-oracle.com/oracle-goldengate-big-data-adapter-apache-kafka-producer

当尝试将上述网页中描述的 模式应用到它对应的模型时（并且在完成 JSON 模型之后），我'我得到了与问题中的模型和模式相同的错误。

非常感谢大家

Answer 1

这就是问题所在

"type": ["null", "columns"]

您不能引用其他记录类型。您需要像对其他字段所做的那样扩展它

通过 kafka-avro-console-producer 生成 Avro 序列化对象的问题

Problem producing Avro serialized object through kafka-avro-console-producer

apache-kafka

avro

oracle-golden-gate