使用MuleSoft DataWeave 2.0递归匹配JSON中Avro异常路径

在MuleSoft中使用Kafka消息监听器连接器搭配Avro Schema连接器时，后者会把JSON中验证错误的部分替换为包含AvroReadingException的对象。需要用DataWeave 2.0递归遍历JSON，找出所有值为AvroReadingException的对象的完整路径及对应错误信息。

示例Payload #

{
  "id": "123",
  "user": {
    "name": "Alice",
    "age": {
      "errorType": "AvroReadingException",
      "message": "Invalid integer value: 'abc'"
    },
    "contact": {
      "email": {
        "errorType": "AvroReadingException",
        "message": "Invalid email format: 'alice@.com'"
      },
      "phone": "1234567890"
    }
  },
  "orders": [
    {
      "orderId": "456",
      "amount": {
        "errorType": "AvroReadingException",
        "message": "Expected positive number: '-100'"
      }
    },
    {
      "orderId": "789",
      "amount": 200
    }
  ]
}

预期结果 #

[
  {
    "path": "user.age",
    "errorMessage": "Invalid integer value: 'abc'"
  },
  {
    "path": "user.contact.email",
    "errorMessage": "Invalid email format: 'alice@.com'"
  },
  {
    "path": "orders[0].amount",
    "errorMessage": "Expected positive number: '-100'"
  }
]

解决方法 #

以下是实现需求的DataWeave 2.0代码，通过递归遍历JSON结构追踪节点路径，精准定位AvroReadingException异常信息：

%dw 2.0
output application/json

fun findAvroErrors(obj: Any, currentPath: String = ""): Array<Object> =
  obj match {
    // 处理对象类型，遍历键值对
    case o is Object -> 
      o flatMap ((value, key) -> 
        if (value is Object and value.errorType == "AvroReadingException") 
          [{ path: if (currentPath == "") key else currentPath ++ "." ++ key, errorMessage: value.message }]
        else 
          findAvroErrors(value, if (currentPath == "") key else currentPath ++ "." ++ key)
      )
    // 处理数组类型，记录元素索引
    case a is Array -> 
      a flatMap ((item, index) -> 
        findAvroErrors(item, if (currentPath == "") "[$index]" else currentPath ++ "[$index]")
      )
    // 非复杂类型直接返回空数组
    case _ -> []
  }

---
findAvroErrors(payload)

代码说明 #

• 递归遍历：通过findAvroErrors函数递归处理JSON的对象和数组，跳过字符串、数字等非复杂类型。
• 路径拼接：对象节点用.连接键名，数组节点用[索引]标记位置，保证路径完整可追溯。
• 异常匹配：一旦检测到包含errorType: "AvroReadingException"的对象，立即生成包含路径和错误信息的条目。

将示例Payload传入上述脚本，即可得到预期的结果数组。

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