Merge remote-tracking branch 'origin/v1.10.0_dev' into v1.10.0_dev

Author: xuchao

.gitignore

@@ -8,6 +8,7 @@ target/
 *.eclipse.*
 *.iml
 plugins/
+sqlplugins/
 lib/
 .vertx/
 .DS_Store

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

@@ -38,10 +38,9 @@
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
 import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
 import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
-import org.apache.flink.streaming.api.operators.async.AsyncWaitOperator;
 import org.apache.flink.streaming.runtime.tasks.ProcessingTimeCallback;
 import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
-import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
+import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -53,10 +52,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
 
 /**
  * All interfaces inherit naming rules: type + "AsyncReqRow" such as == "MysqlAsyncReqRow
@@ -114,7 +110,7 @@ private void initMetric() {
 
 
     protected Object convertTimeIndictorTypeInfo(Integer index, Object obj) {
-        boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(index).getClass());
+        boolean isTimeIndicatorTypeInfo = DataTypes.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(index).getClass());
 
         //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
         if (obj instanceof LocalDateTime && isTimeIndicatorTypeInfo) {
@@ -257,19 +253,15 @@ protected ScheduledFuture<?> registerTimer(Tuple2<Boolean,Row> input, ResultFutu
         long timeoutTimestamp = sideInfo.getSideTableInfo().getAsyncTimeout() + getProcessingTimeService().getCurrentProcessingTime();
         return getProcessingTimeService().registerTimer(
                 timeoutTimestamp,
-                new ProcessingTimeCallback() {
-                    @Override
-                    public void onProcessingTime(long timestamp) throws Exception {
-                        timeout(input, resultFuture);
-                    }
-                });
+                timestamp -> timeout(input, resultFuture));
     }
 
     protected void registerTimerAndAddToHandler(Tuple2<Boolean,Row> input, ResultFuture<Tuple2<Boolean,Row>> resultFuture)
             throws InvocationTargetException, IllegalAccessException {
         ScheduledFuture<?> timeFuture = registerTimer(input, resultFuture);
         // resultFuture 是ResultHandler 的实例
         Method setTimeoutTimer = ReflectionUtils.getDeclaredMethod(resultFuture, "setTimeoutTimer", ScheduledFuture.class);
+        setTimeoutTimer.setAccessible(true);
         setTimeoutTimer.invoke(resultFuture, timeFuture);
     }