Merge branch '1.8_release_3.10.x_jingdongfang' into 1.8_release_3.10.x

Author: gituser

flinkx-rdb/flinkx-rdb-reader/src/main/java/com.dtstack.flinkx.rdb.inputformat/JdbcInputFormat.java

@@ -53,6 +53,7 @@
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.sql.Timestamp;
+import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -129,6 +130,8 @@ public class JdbcInputFormat extends BaseRichInputFormat {
 
     protected Row lastRow = null;
 
+    protected transient final static ThreadLocal<SimpleDateFormat> sdf = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyyMMddHHmmss"));
+
     protected String querySql;
 
     /**
@@ -688,7 +691,12 @@ protected String getLocationSql(String incrementColType, String incrementCol, St
         } else if (ColumnType.isNumberType(incrementColType)) {
             endLocationSql = incrementCol + operator + location;
         } else {
-            endTimeStr = String.format("'%s'", location);
+            // FIXME 京东方特殊逻辑
+            String part1 = location.substring(0, 13);
+            String part2 = location.substring(13);
+            String timeStr = sdf.get().format(new Date(Long.parseLong(part1))) + part2;
+
+            endTimeStr = String.format("'%s'",timeStr);
             endLocationSql = incrementCol + operator + endTimeStr;
         }