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

Author: dapeng

.gitlab-ci.yml

@@ -2,7 +2,7 @@ build:
   stage: test
   script:
     - mvn clean org.jacoco:jacoco-maven-plugin:0.7.8:prepare-agent package -Dmaven.test.failure.ignore=true -q
-    - mvn sonar:sonar -Dsonar.projectKey="dt-insight-engine/flinkStreamSQL" -Dsonar.branch.name="v1.8.0_dev" -Dsonar.login=11974c5e9a29625efa09fdc3c3fdc031efb1aab1 -Dsonar.host.url=http://172.16.100.198:9000 -Dsonar.jdbc.url=jdbc:postgresql://172.16.100.198:5432/sonar -Dsonar.java.binaries=target/sonar
+    - mvn sonar:sonar -Dsonar.projectKey="dt-insight-engine/flinkStreamSQL"  -Dsonar.login=11974c5e9a29625efa09fdc3c3fdc031efb1aab1 -Dsonar.host.url=http://172.16.100.198:9000 -Dsonar.jdbc.url=jdbc:postgresql://172.16.100.198:5432/sonar -Dsonar.java.binaries=target/sonar
     - sh ci/sonar_notify.sh
   only:
     - v1.8.0_dev

cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java

@@ -162,60 +162,13 @@ private void connCassandraDB(CassandraSideTableInfo tableInfo) {
     }
 
     @Override
-    public void asyncInvoke(CRow input, ResultFuture<CRow> resultFuture) throws Exception {
-        CRow inputCopy = new CRow(input.row(), input.change());
-        JsonArray inputParams = new JsonArray();
-        StringBuffer stringBuffer = new StringBuffer();
-        String sqlWhere = " where ";
-
-        for (int i = 0; i < sideInfo.getEqualFieldList().size(); i++) {
-            Integer conValIndex = sideInfo.getEqualValIndex().get(i);
-            Object equalObj = inputCopy.row().getField(conValIndex);
-            if (equalObj == null) {
-                dealMissKey(inputCopy, resultFuture);
-                return;
-            }
-            inputParams.add(equalObj);
-            StringBuffer sqlTemp = stringBuffer.append(sideInfo.getEqualFieldList().get(i))
-                    .append(" = ");
-            if (equalObj instanceof String) {
-                sqlTemp.append("'" + equalObj + "'")
-                        .append(" and ");
-            } else {
-                sqlTemp.append(equalObj)
-                        .append(" and ");
-            }
-
-        }
+    public void handleAsyncInvoke(Map<String, Object> inputParams, CRow input, ResultFuture<CRow> resultFuture) throws Exception {
 
         String key = buildCacheKey(inputParams);
-        sqlWhere = sqlWhere + stringBuffer.toString().substring(0, stringBuffer.lastIndexOf(" and "));
-
-        if (openCache()) {
-            CacheObj val = getFromCache(key);
-            if (val != null) {
-
-                if (ECacheContentType.MissVal == val.getType()) {
-                    dealMissKey(inputCopy, resultFuture);
-                    return;
-                } else if (ECacheContentType.MultiLine == val.getType()) {
-                    List<CRow> rowList = Lists.newArrayList();
-                    for (Object jsonArray : (List) val.getContent()) {
-                        Row row = fillData(inputCopy.row(), jsonArray);
-                        rowList.add(new CRow(row, inputCopy.change()));
-                    }
-                    resultFuture.complete(rowList);
-                } else {
-                    throw new RuntimeException("not support cache obj type " + val.getType());
-                }
-                return;
-            }
-        }
-
         //connect Cassandra
         connCassandraDB(cassandraSideTableInfo);
 
-        String sqlCondition = sideInfo.getSqlCondition() + " " + sqlWhere + "  ALLOW FILTERING ";
+        String sqlCondition = sideInfo.getSqlCondition() + " " + buildWhereCondition(inputParams) + "  ALLOW FILTERING ";
         LOG.info("sqlCondition:{}" + sqlCondition);
 
         ListenableFuture<ResultSet> resultSet = Futures.transformAsync(session,
@@ -242,18 +195,18 @@ public void onSuccess(List<com.datastax.driver.core.Row> rows) {
                     List<com.datastax.driver.core.Row> cacheContent = Lists.newArrayList();
                     List<CRow> rowList = Lists.newArrayList();
                     for (com.datastax.driver.core.Row line : rows) {
-                        Row row = fillData(inputCopy.row(), line);
+                        Row row = fillData(input.row(), line);
                         if (openCache()) {
                             cacheContent.add(line);
                         }
-                        rowList.add(new CRow(row,inputCopy.change()));
+                        rowList.add(new CRow(row, input.change()));
                     }
                     resultFuture.complete(rowList);
                     if (openCache()) {
                         putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));
                     }
                 } else {
-                    dealMissKey(inputCopy, resultFuture);
+                    dealMissKey(input, resultFuture);
                     if (openCache()) {
                         putCache(key, CacheMissVal.getMissKeyObj());
                     }
@@ -271,6 +224,24 @@ public void onFailure(Throwable t) {
         });
     }
 
+    @Override
+    public String buildCacheKey(Map<String, Object> inputParams) {
+        StringBuilder sb = new StringBuilder();
+        for (Object ele : inputParams.values()) {
+            sb.append(ele.toString()).append("_");
+        }
+        return sb.toString();
+    }
+
+    private String buildWhereCondition(Map<String, Object> inputParams){
+        StringBuilder sb = new StringBuilder(" where ");
+        for(Map.Entry<String, Object> entry : inputParams.entrySet()){
+            Object value = entry.getValue() instanceof String ? "'" + entry.getValue() + "'" : entry.getValue();
+            sb.append(String.format("%s = %s", entry.getKey(), value));
+        }
+        return sb.toString();
+    }
+
     @Override
     public Row fillData(Row input, Object line) {
         com.datastax.driver.core.Row rowArray = (com.datastax.driver.core.Row) line;
@@ -305,14 +276,4 @@ public void close() throws Exception {
             cluster = null;
         }
     }
-
-    public String buildCacheKey(JsonArray jsonArray) {
-        StringBuilder sb = new StringBuilder();
-        for (Object ele : jsonArray.getList()) {
-            sb.append(ele.toString())
-                    .append("_");
-        }
-
-        return sb.toString();
-    }
 }

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

@@ -27,6 +27,7 @@
 
 import java.io.Serializable;
 import java.util.List;
+import java.util.Objects;
 
 /**
  * Reason:
@@ -53,15 +54,15 @@ public abstract class AbstractSideTableInfo extends AbstractTableInfo implements
 
     public static final String ASYNC_TIMEOUT_KEY = "asyncTimeout";
 
-    public static final String ASYNC_TIMEOUT_NUM_KEY = "asyncTimeoutNum";
-
+    public static final String ASYNC_FAIL_MAX_NUM_KEY = "asyncFailMaxNum";
+    
     public static final String ASYNC_REQ_POOL_KEY = "asyncPoolSize";
 
     private String cacheType = "none";
 
     private int cacheSize = 10000;
 
-    private long cacheTimeout = 60_000L;
+    private long cacheTimeout = 60 * 1000L;
 
     private int  asyncCapacity=100;
 
@@ -72,12 +73,12 @@ public abstract class AbstractSideTableInfo extends AbstractTableInfo implements
      */
     private int asyncPoolSize = 0;
 
-    private int asyncTimeoutNumLimit = Integer.MAX_VALUE;
-
     private boolean partitionedJoin = false;
 
     private String cacheMode="ordered";
 
+    private Long asyncFailMaxNum;
+
     private List<PredicateInfo>  predicateInfoes = Lists.newArrayList();
 
     public RowTypeInfo getRowTypeInfo(){
@@ -155,12 +156,12 @@ public List<PredicateInfo> getPredicateInfoes() {
         return predicateInfoes;
     }
 
-    public int getAsyncTimeoutNumLimit() {
-        return asyncTimeoutNumLimit;
+    public Long getAsyncFailMaxNum(Long defaultValue) {
+        return Objects.isNull(asyncFailMaxNum) ? defaultValue : asyncFailMaxNum;
     }
 
-    public void setAsyncTimeoutNumLimit(int asyncTimeoutNumLimit) {
-        this.asyncTimeoutNumLimit = asyncTimeoutNumLimit;
+    public void setAsyncFailMaxNum(Long asyncFailMaxNum) {
+        this.asyncFailMaxNum = asyncFailMaxNum;
     }
 
     public int getAsyncPoolSize() {
@@ -180,7 +181,7 @@ public String toString() {
                 ", asyncCapacity=" + asyncCapacity +
                 ", asyncTimeout=" + asyncTimeout +
                 ", asyncPoolSize=" + asyncPoolSize +
-                ", asyncTimeoutNumLimit=" + asyncTimeoutNumLimit +
+                ", asyncFailMaxNum=" + asyncFailMaxNum +
                 ", partitionedJoin=" + partitionedJoin +
                 ", cacheMode='" + cacheMode + '\'' +
                 '}';