Pubsub嵌套Avro Schema测试报错：Message is invalid against schema

问题背景 #

尝试为Pub/Sub主题创建嵌套Avro Schema，Schema已通过验证，但测试消息时提示Message is invalid against schema。

所用Avro Schema #

{
  "type" : "record",
  "name" : "topLevelRecord",
  "fields" : [ {
    "name" : "dT_AssignmentExt",
    "type" : [ {
      "type" : "array",
      "items" : [ {
        "type" : "record",
        "name" : "dT_AssignmentExt",
        "namespace" : "topLevelRecord",
        "fields" : [ {
          "name" : "arrivalStation",
          "type" : [ "string", "null" ]
        }, {
          "name" : "arrivalTime",
          "type" : [ "string", "null" ]
        } ]
      }, "null" ]
    }, "null" ]
  }, {
    "name" : "dT_PlanHeaderExt",
    "type" : [ {
      "type" : "record",
      "name" : "dT_PlanHeaderExt",
      "namespace" : "topLevelRecord",
      "fields" : [ {
        "name" : "created",
        "type" : [ "string", "null" ]
      }, {
        "name" : "endDate",
        "type" : [ "string", "null" ]
      }, {
        "name" : "startDate",
        "type" : [ "string", "null" ]
      } ]
    }, "null" ]
  }, {
    "name" : "dT_RotationHeaderExt",
    "type" : [ {
      "type" : "record",
      "name" : "dT_RotationHeaderExt",
      "namespace" : "topLevelRecord",
      "fields" : [ {
        "name" : "locomotiveId",
        "type" : [ "string", "null" ]
      }, {
        "name" : "locomotiveType",
        "type" : [ "string", "null" ]
      } ]
    }, "null" ]
  } ]
}

测试消息（存在格式问题） #

// 错误：缺少开头的{，JSON结构不完整
 "dT_AssignmentExt": [
    {
       "arrivalStation": "DE.AM",
       "arrivalTime": "2023-03-01T10:33:00Z"
    },
    {
       "arrivalStation": "DE.AM",
       "arrivalTime": "2023-03-13T15:00:00Z"
    }, 
    {
       "arrivalStation": "PHM",
       "arrivalTime": "2023-03-14T00:51:00Z"
    }, 
    {
       "arrivalStation": "MGB",
       "arrivalTime": "2023-03-14T01:17:00Z"
    }
 ],
 "dT_PlanHeaderExt": 
 {
    "created": "2023-03-01T11:32:00Z",
    "endDate": "2023-03-15T11:32:00Z",
    "startDate": "2023-03-01T11:32:00Z"
 },
 "dT_RotationHeaderExt": 
 {
    "locomotiveId": "5405",
    "locomotiveType": "BR185"
 }
}

// 错误：两个JSON对象直接拼接，属于无效JSON
{
  "dT_PlanHeaderExt": 
     {
        "created": "2023-03-01T11:32:00Z",
        "endDate": "2023-03-15T11:32:00Z",
        "startDate": "2023-03-01T11:32:00Z"
     }
}

问题分析 #

Pub/Sub完全支持嵌套Avro Schema，报错原因出在测试消息格式和Schema的冗余定义上：

1. 测试消息格式错误 #

• 第一条消息开头缺少{，导致JSON结构不完整
• 两条JSON对象直接拼接，形成无效的JSON内容，Pub/Sub要求每条消息必须是单个有效的JSON对象

2. Schema定义存在冗余的null嵌套（可选优化） #

dT_AssignmentExt字段的类型定义为[array, null]，而数组的items又定义为[record, null]，这意味着：

• 整个dT_AssignmentExt字段可以为null
• 或者是一个数组，数组中的每个元素可以是dT_AssignmentExt记录或null
  如果业务场景中数组元素不会为null，可以简化items的定义为仅record类型，减少嵌套复杂度：

"name" : "dT_AssignmentExt",
"type" : [ {
  "type" : "array",
  "items" : {
    "type" : "record",
    "name" : "dT_AssignmentExt",
    "namespace" : "topLevelRecord",
    "fields" : [ {
      "name" : "arrivalStation",
      "type" : [ "string", "null" ]
    }, {
      "name" : "arrivalTime",
      "type" : [ "string", "null" ]
    } ]
  }
}, "null" ]

修复后的测试消息 #

每条消息单独作为有效JSON：

消息1

{
 "dT_AssignmentExt": [
    {
       "arrivalStation": "DE.AM",
       "arrivalTime": "2023-03-01T10:33:00Z"
    },
    {
       "arrivalStation": "DE.AM",
       "arrivalTime": "2023-03-13T15:00:00Z"
    }, 
    {
       "arrivalStation": "PHM",
       "arrivalTime": "2023-03-14T00:51:00Z"
    }, 
    {
       "arrivalStation": "MGB",
       "arrivalTime": "2023-03-14T01:17:00Z"
    }
 ],
 "dT_PlanHeaderExt": 
 {
    "created": "2023-03-01T11:32:00Z",
    "endDate": "2023-03-15T11:32:00Z",
    "startDate": "2023-03-01T11:32:00Z"
 },
 "dT_RotationHeaderExt": 
 {
    "locomotiveId": "5405",
    "locomotiveType": "BR185"
 }
}

消息2

{
  "dT_PlanHeaderExt": 
     {
        "created": "2023-03-01T11:32:00Z",
        "endDate": "2023-03-15T11:32:00Z",
        "startDate": "2023-03-01T11:32:00Z"
     }
}

验证说明 #

修复后的消息符合Schema定义：

• 所有可选字段（允许为null）如果未提供，默认视为null，符合Schema的联合类型定义
• 数组元素为有效record对象，即使Schema允许元素为null，非null元素也完全合规

内容的提问来源于stack exchange，提问作者Alex K