Avro Schema适配问题：向Kafka写入数据时解码失败求助

问题详情 #

使用指定Avro Schema向Kafka Topic生产数据时触发解码错误，相关信息如下：

Avro Schema #

{"doc": "Sample schema to help you get started.","fields": [{"name": "specversion","type": "string"},{"name": "event","type": {"fields": [{"name": "communicationMessage","type": {"fields": [{"name": "id","type": "string"},{"default": null,"name": "additionalInfo","type": ["null",{"fields": [{"default": null,"name": "nationalId","type": ["null","string"]},{"default": null,"name": "productNameArabic","type": ["null","string"]}],"name": "additionalInfo","type": "record"}]}],"name": "communicationMessage","type": "record"}]}],"name": "event","type": "record"}}],"name": "example","namespace": "com.example","type": "record"}

待生产的输入数据 #

{"specversion": "1.0","event": {"communicationMessage": {"id": "message123","additionalInfo": {"additionalInfo": {"nationalId": "1234567890","productNameArabic": "example"}}}}}

报错信息 #

Error: cannot decode textual record "com.example.example": cannot decode textual record "com.example.event": cannot decode textual record "com.example.communicationMessage": cannot decode textual union: cannot decode textual map: cannot determine codec: "additionalInfo" for key: "additionalInfo" for key: "communicationMessage" for key: "event"

错误原因 #

Schema与输入数据结构不匹配：

• Schema定义中，communicationMessage下的additionalInfo是["null", record]联合类型，其中的record直接包含nationalId和productNameArabic字段，无需额外嵌套。
• 但输入数据里，additionalInfo多套了一层{"additionalInfo": {...}}结构，导致解码器无法匹配Schema规则，触发错误。

修复方案 #

方案1：修改输入数据（保留原Schema） #

移除多余的嵌套层，直接传递符合Schema定义的record结构：

{"specversion": "1.0","event": {"communicationMessage": {"id": "message123","additionalInfo": {"nationalId": "1234567890","productNameArabic": "example"}}}}

方案2：修改Avro Schema（保留原输入结构） #

如果需要维持输入的嵌套格式，调整Schema中additionalInfo的定义，新增一层包裹字段：

{
  "doc": "Sample schema to help you get started.",
  "name": "example",
  "namespace": "com.example",
  "type": "record",
  "fields": [
    {"name": "specversion","type": "string"},
    {
      "name": "event",
      "type": {
        "type": "record",
        "name": "event",
        "fields": [
          {
            "name": "communicationMessage",
            "type": {
              "type": "record",
              "name": "communicationMessage",
              "fields": [
                {"name": "id","type": "string"},
                {
                  "default": null,
                  "name": "additionalInfo",
                  "type": [
                    "null",
                    {
                      "type": "record",
                      "name": "additionalInfoWrapper",
                      "fields": [
                        {
                          "type": [
                            "null",
                            {
                              "type": "record",
                              "name": "additionalInfo",
                              "fields": [
                                {"default": null,"name": "nationalId","type": ["null","string"]},
                                {"default": null,"name": "productNameArabic","type": ["null","string"]}
                              ]
                            }
                          ],
                          "name": "additionalInfo",
                          "default": null
                        }
                      ]
                    }
                  ]
                }
              ]
            }
          }
        ]
      }
    }
  ]
}

内容的提问来源于stack exchange，提问作者Disha Giriyan