org.apache.avro.AvroTypeException: 未知联合分支 EventId

Question

我正在尝试使用 'kafka-avro-console-producer' 将 json 转换为 avro 并将其发布到 kafka 主题。

我可以做到 json/schema 但对于下面给定的模式和 json 我得到 “org.apache.avro.AvroTypeException：未知联合分支 EventId”错误。

如有任何帮助，我们将不胜感激。

架构：

{
    "type": "record",
    "name": "Envelope",
    "namespace": "CoreOLTPEvents.dbo.Event",
    "fields": [{
        "name": "before",
        "type": ["null", {
            "type": "record",
            "name": "Value",
            "fields": [{
                "name": "EventId",
                "type": "long"
            }, {
                "name": "CameraId",
                "type": ["null", "long"],
                "default": null
            }, {
                "name": "SiteId",
                "type": ["null", "long"],
                "default": null
            }],
            "connect.name": "CoreOLTPEvents.dbo.Event.Value"
        }],
        "default": null
    }, {
        "name": "after",
        "type": ["null", "Value"],
        "default": null
    }, {
        "name": "op",
        "type": "string"
    }, {
        "name": "ts_ms",
        "type": ["null", "long"],
        "default": null
    }],
    "connect.name": "CoreOLTPEvents.dbo.Event.Envelope"
}

而 Json 输入如下：

{
    "before": null,
    "after": {
        "EventId": 12,
        "CameraId": 10,
        "SiteId": 11974
    },
    "op": "C",
    "ts_ms": null
}

在我的例子中，我不能改变模式，我只能改变json这样它可以工作的方式

Answer 1

删除 "connect.name": "CoreOLTPEvents.dbo.Event.Value" 和 "connect.name": "CoreOLTPEvents.dbo.Event.Envelope" 作为 The RecordType can only contains {'namespace', 'aliases', 'fields', 'name', 'type', 'doc'} keys。您能否尝试使用以下架构，看看您是否能够生成消息？

{
  "type": "record",
  "name": "Envelope",
  "namespace": "CoreOLTPEvents.dbo.Event",
  "fields": [
    {
      "name": "before",
      "type": [
        "null",
        {
          "type": "record",
          "name": "Value",
          "fields": [
            {
              "name": "EventId",
              "type": "long"
            },
            {
              "name": "CameraId",
              "type": [
                "null",
                "long"
              ],
              "default": "null"
            },
            {
              "name": "SiteId",
              "type": [
                "null",
                "long"
              ],
              "default": "null"
            }
          ]
        }
      ],
      "default": null
    },
    {
      "name": "after",
      "type": [
        "null",
        "Value"
      ],
      "default": null
    },
    {
      "name": "op",
      "type": "string"
    },
    {
      "name": "ts_ms",
      "type": [
        "null",
        "long"
      ],
      "default": null
    }
  ]
}

Answer 2

如果您使用的是 Avro JSON 格式，则您的输入略有偏差。对于联合，需要指定非空值，以便列出类型信息：https://avro.apache.org/docs/current/spec.html#json_encoding

请参阅下面的示例，我认为应该可行。

{
    "before": null,
    "after": {
        "CoreOLTPEvents.dbo.Event.Value": {
            "EventId": 12,
            "CameraId": {
                "long": 10
            },
            "SiteId": {
                "long": 11974
            }
        }
    },
    "op": "C",
    "ts_ms": null
}

org.apache.avro.AvroTypeException: 未知联合分支 EventId

org.apache.avro.AvroTypeException: Unknown union branch EventId

avro

apache-kafka

confluent-schema-registry

confluent-cloud

confluent-platform