修改bug

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

@@ -58,12 +58,12 @@ public class AsyncMysqlDataTransferFunctionNew extends RichAsyncFunction<JSONObj
                 String type = value.getString("type");
                 JSONArray dataList = value.getJSONArray("data");
                 JSONObject mysqlType = value.getJSONObject("mysqlType");
-                JSONArray oldDataList = value.getJSONArray("old");
+                //JSONArray oldDataList = value.getJSONArray("old");
                 String table = value.getString("table");
                 Boolean isDdl = value.getBoolean("isDdl");
                 JSONArray pkNames = value.getJSONArray("pkNames");
                 Set<String> pkNameSet = new HashSet<>();
-                long ts = value.getLong("ts");
+                long ts = value.getLong("ts") == null ? System.currentTimeMillis() : value.getLong("ts");
                 if(CollUtil.isNotEmpty(pkNames)){
                     pkNames.forEach(name -> pkNameSet.add(String.valueOf(name)));
                 }