[fix-32570][core]关闭enableObjectReuse，并修复关闭之后带来的序列化问题

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/environment/StreamEnvConfigManager.java

@@ -73,8 +73,6 @@ public static void streamExecutionEnvironmentConfig(StreamExecutionEnvironment s
 
         confProperties = PropertiesUtils.propertiesTrim(confProperties);
         streamEnv.getConfig().disableClosureCleaner();
-        // Disables reusing object
-        streamEnv.getConfig().enableObjectReuse();
 
         Configuration globalJobParameters = new Configuration();
         //Configuration unsupported set properties key-value

core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java

@@ -45,6 +45,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 import org.apache.flink.api.common.typeinfo.LocalTimeTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeHint;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.streaming.api.datastream.DataStream;
@@ -65,6 +66,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.sql.Timestamp;
 import java.time.LocalDateTime;
 import java.util.Arrays;
 import java.util.LinkedList;
@@ -442,6 +444,10 @@ private void joinFun(Object pollObj,
             if (fieldDataTypes[i].getClass().equals(LegacyLocalDateTimeTypeInfo.class)) {
                 fieldDataTypes[i] = LocalTimeTypeInfo.LOCAL_DATE_TIME;
             }
+
+            if (fieldDataTypes[i].getClass().equals(TimeIndicatorTypeInfo.class)) {
+                fieldDataTypes[i] = TypeInformation.of(new TypeHint<Timestamp>() {});
+            }
         }
 
         RowTypeInfo typeInfo = new RowTypeInfo(fieldDataTypes, targetTable.getSchema().getFieldNames());