kingbase

src/main/java/com/dsk/flink/dsc/common/function/MysqlDataTransferFunction.java

@@ -81,8 +81,8 @@ public class MysqlDataTransferFunction extends ProcessFunction<JSONObject, Tuple
         }
         String groupKey = groupKeyBuilder.toString();
 
-        //添加分表参数
-        String shardingRule = dataObj.getString("shardingRule");
+        //添加分表参数（shardingRule 位于消息顶层，由 dsk-flink-upgrade 注入）
+        String shardingRule = value.getString("shardingRule");
         if (StrUtil.isNotBlank(shardingRule)) {
             Map<String,Object> map = JSON.parseObject(shardingRule, Map.class);
             String strategy = MapUtil.getStr(map, "strategy");