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

Author: dapeng

mongo/mongo-side/mongo-side-core/src/main/java/com/dtstack/flink/sql/side/mongo/utils/MongoUtil.java

@@ -21,6 +21,7 @@
 import com.dtstack.flink.sql.side.PredicateInfo;
 import com.mongodb.BasicDBObject;
 import org.apache.commons.lang3.StringUtils;
+import org.bson.Document;
 
 import java.util.Arrays;
 import java.util.stream.Collectors;
@@ -32,25 +33,27 @@
  */
 public class MongoUtil {
     public static BasicDBObject buildFilterObject(PredicateInfo info) {
+
+        String value = info.getCondition().replaceAll("'", "");
         switch (info.getOperatorName()) {
             case "=":
-                return new BasicDBObject("$eq", info.getCondition());
+                return new BasicDBObject("$eq", value);
             case ">":
-                return new BasicDBObject("$gt", info.getCondition());
+                return new BasicDBObject("$gt", value);
             case ">=":
-                return new BasicDBObject("$gte", info.getCondition());
+                return new BasicDBObject("$gte", value);
             case "<":
-                return new BasicDBObject("$lt", info.getCondition());
+                return new BasicDBObject("$lt", value);
             case "<=":
-                return new BasicDBObject("$lte", info.getCondition());
+                return new BasicDBObject("$lte", value);
             case "<>":
-                return new BasicDBObject("$ne", info.getCondition());
+                return new BasicDBObject("$ne", value);
             case "IN":
-                Object[] values = Arrays.stream(StringUtils.split(info.getCondition(), ",")).map(String::trim)
+                Object[] values = Arrays.stream(StringUtils.split(value, ",")).map(String::trim)
                         .collect(Collectors.toList()).toArray();
                 return new BasicDBObject("$in", values);
             case "NOT IN":
-                return new BasicDBObject("$nin", StringUtils.split(info.getCondition(), ","));
+                return new BasicDBObject("$nin", StringUtils.split(value, ","));
             case "IS NOT NULL":
                 return new BasicDBObject("$exists", true);
             case "IS NULL":

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

@@ -137,8 +137,8 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
                         if(failCounter.getAndIncrement() % 1000 == 0){
                             LOG.error("getConnection error", conn.cause());
                         }
-                        if(failCounter.get() >= sideInfo.getSideTableInfo().getConnectRetryMaxNum(3)){
-                            dealFillDataError(input, resultFuture, conn.cause());
+                        if(failCounter.get() >= sideInfo.getSideTableInfo().getConnectRetryMaxNum(100)){
+                            resultFuture.completeExceptionally(conn.cause());
                             finishFlag.set(true);
                         }
                         conn.result().close();
@@ -155,12 +155,18 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
                     latch.countDown();
                 }
             });
-                        //主线程阻塞
             try {
                 latch.wait();
             } catch (InterruptedException e) {
                 LOG.error("", e);
             }
+            if(!finishFlag.get()){
+                try {
+                    Thread.sleep(100);
+                } catch (Exception e){
+                    LOG.error("", e);
+                }
+            }
         }
     }