Merge branch '1.8_release_3.9.x' into hotfix_1.8_3.9.x_21763

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

Author: zoudaokoulife

clickhouse/clickhouse-side/clickhouse-all-side/pom.xml

@@ -41,7 +41,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

clickhouse/clickhouse-side/clickhouse-async-side/pom.xml

@@ -41,7 +41,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

clickhouse/clickhouse-sink/pom.xml

@@ -40,7 +40,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -244,7 +244,8 @@ private static Set<URL> registerTable(SqlTree sqlTree, StreamExecutionEnvironmen
 
                 RowTypeInfo typeInfo = new RowTypeInfo(adaptTable.getSchema().getTypes(), adaptTable.getSchema().getColumnNames());
                 DataStream adaptStream = tableEnv.toRetractStream(adaptTable, typeInfo)
-                        .map((Tuple2<Boolean, Row> f0) -> { return f0.f1; })
+                        .filter((Tuple2<Boolean, Row> f0) -> f0.f0)
+                        .map((Tuple2<Boolean, Row> f0) -> f0.f1)
                         .returns(typeInfo);
 
                 String fields = String.join(",", typeInfo.getFieldNames());

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

@@ -51,6 +51,9 @@ public abstract class AsyncReqRow extends RichAsyncFunction<Row, Row> implements
     private static final Logger LOG = LoggerFactory.getLogger(AsyncReqRow.class);
     private static final long serialVersionUID = 2098635244857937717L;
 
+    private static int TIMEOUT_LOG_FLUSH_NUM = 10;
+    private int timeOutNum = 0;
+
     protected SideInfo sideInfo;
     protected transient Counter parseErrorRecords;
 
@@ -119,14 +122,14 @@ protected void dealCacheData(String key, CacheObj missKeyObj) {
 
     @Override
     public void timeout(Row input, ResultFuture<Row> resultFuture) throws Exception {
-        StreamRecordQueueEntry<Row> future = (StreamRecordQueueEntry<Row>)resultFuture;
-        try {
-            if (null == future.get()) {
-                resultFuture.completeExceptionally(new TimeoutException("Async function call has timed out."));
-            }
-        } catch (Exception e) {
-            resultFuture.completeExceptionally(new Exception(e));
+
+        //TODO 需要添加数据指标
+        if(timeOutNum % TIMEOUT_LOG_FLUSH_NUM == 0){
+            LOG.info("Async function call has timed out. input:" + input.toString());
         }
+
+        timeOutNum++;
+        resultFuture.complete(null);
     }
 
 

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

@@ -803,8 +803,9 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
         RowTypeInfo typeInfo = new RowTypeInfo(targetTable.getSchema().getTypes(), targetTable.getSchema().getColumnNames());
 
         DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class)
-                                .map((Tuple2<Boolean, Row> f0) -> f0.f1)
-                                .returns(Row.class);
+                .filter((Tuple2<Boolean, Row> f0) -> f0.f0)
+                .map((Tuple2<Boolean, Row> f0) -> f0.f1)
+                .returns(Row.class);
 
 
         //join side table before keyby ===> Reducing the size of each dimension table cache of async