Merge branch 'feat_1.8_asyncException_mergeTest' into 1.8_release_3.10.x_mergedTest_new

Author: dapeng

kudu/kudu-side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncSideInfo.java

@@ -4,6 +4,9 @@
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.BaseSideInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
+import com.dtstack.flink.sql.side.kudu.table.KuduSideTableInfo;
+import com.dtstack.flink.sql.util.ParseUtils;
+import com.google.common.collect.Lists;
 import org.apache.calcite.sql.SqlBasicCall;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlKind;
@@ -21,6 +24,22 @@ public KuduAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldI
 
     @Override
     public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInfo) {
+        KuduSideTableInfo kuduSideTableInfo = (KuduSideTableInfo) sideTableInfo;
+
+        String sideTableName = joinInfo.getSideTableName();
+
+        SqlNode conditionNode = joinInfo.getCondition();
+
+        List<SqlNode> sqlNodeList = Lists.newArrayList();
+        ParseUtils.parseAnd(conditionNode, sqlNodeList);
+
+        for (SqlNode sqlNode : sqlNodeList) {
+            dealOneEqualCon(sqlNode, sideTableName);
+        }
+
+        sqlCondition = "select ${selectField} from ${tableName} ";
+        sqlCondition = sqlCondition.replace("${tableName}", kuduSideTableInfo.getTableName()).replace("${selectField}", sideSelectFields);
+        System.out.println("---------side_exe_sql-----\n" + sqlCondition);
     }
 
     @Override

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

@@ -155,13 +155,13 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
                 }
             });
             try {
-                latch.wait();
+                latch.await();
             } catch (InterruptedException e) {
                 LOG.error("", e);
             }
             if(!finishFlag.get()){
                 try {
-                    Thread.sleep(100);
+                    Thread.sleep(3000);
                 } catch (Exception e){
                     LOG.error("", e);
                 }