Talend tJava消费Kafka主题JSON数据读取null值报错排查

问题背景 #

• 作业场景：在Talend Studio中搭建基于Kafka的数据接入作业工作流，读取Kafka主题为work队列中的JSON格式数据，最终存储到Hive表中。
• 待解析JSON样例：

{"Header": {"Vers":"1.0","Message": "318","Owner": {"ID": 102,"FID": 101},"Mode":"8"},"Request": {"Type":"4","ObjType": "S","OrderParam":[{"Code": "OpType","Value": "30"},{"Code": "Time","Value": "202"},{"Code": "AddProperty","ObjParam": [{"Param": [{"Code": "Sync","Value": "Y"}]}]}]}}
{"Header": {"Vers":"2.0","Message": "318","Owner": {"ID": 103,"FID": 102},"Mode":"8"},"Request": {"Type":"5","ObjType": "S","OrderParam":[{"Code": "OpType","Value": "90"},{"Code": "Time","Value": "203"},{"Code": "AddProperty","ObjParam": [{"Param": [{"Code": "Sync","Value": "Y"}]}]}]}}

• Talend工作流参考：
  
• 问题范围：不涉及Talend组件配置，仅解决tJava组件中JSON字段读取报错的问题。

故障现象 #

原有tJava组件代码如下，代码可成功从tLogRow获取数据赋值给output变量，但运行时触发JSONObject读取null值报错，无法获取目标字段值：

String output=((String)globalMap.get("tLogRow_1_OUTPUT"));

JSONObject jsonObject = new JSONObject(output);
System.out.println(jsonObject);

String sourceDBName=(jsonObject.getString("Vers"));

故障原因 #

1. 缺少入参非空校验：如果output获取到空字符串、null值，直接构造JSONObject会触发空指针或格式错误。
2. JSON层级读取错误：观察JSON结构，Vers字段是Header对象的子属性，并非根节点的直接属性，直接从根节点调用getString("Vers")必然读取到null值。

修正后代码 #

按照JSON嵌套结构逐层读取字段，同时增加非空校验逻辑，代码如下：

// 从全局上下文获取tLogRow输出内容
String output = ((String)globalMap.get("tLogRow_1_OUTPUT"));

// 前置非空校验，拦截无效输入
if (output != null && !output.trim().isEmpty()) {
    // 构造根层级JSON对象
    JSONObject rootJson = new JSONObject(output);
    
    // 先获取嵌套的Header对象，再从Header中读取Vers字段
    JSONObject headerJson = rootJson.getJSONObject("Header");
    String sourceDBName = headerJson.getString("Vers");
    System.out.println("读取到的Vers字段值：" + sourceDBName);

    // 其余嵌套字段读取参考如下逻辑
    // 读取Header下Owner对象的ID字段
    JSONObject ownerJson = headerJson.getJSONObject("Owner");
    Integer ownerId = ownerJson.getInt("ID");
    // 读取Request下OrderParam数组
    JSONObject requestJson = rootJson.getJSONObject("Request");
    JSONArray orderParamArr = requestJson.getJSONArray("OrderParam");
} else {
    System.out.println("获取到的Kafka消息为空，跳过解析");
}

注：代码中用到的JSONObject、JSONArray类均为Talend内置org.json包下的实现，无需额外引入依赖。

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