Merge branch 'feat_async_mergedDev' into 'v1.8.0_dev'

Feat async merged dev



See merge request !250

Author: zoudaokoulife

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraOutputFormat.java

@@ -219,7 +219,7 @@ private void insertWrite(Row row) {
             }
         } catch (Exception e) {
             if(outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0){
-                LOG.error("record insert failed ..", row.toString().substring(0, 100));
+                LOG.error("record insert failed, total dirty num:{}, current record:{}", outDirtyRecords.getCount(), row.toString());
                 LOG.error("", e);
             }
 

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/side/AsyncReqRow.java

@@ -51,6 +51,9 @@ public abstract class AsyncReqRow extends RichAsyncFunction<CRow, CRow> implemen
     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,13 +122,16 @@ protected void dealCacheData(String key, CacheObj missKeyObj) {
 
     @Override
     public void timeout(CRow input, ResultFuture<CRow> resultFuture) throws Exception {
-        StreamRecordQueueEntry<CRow> future = (StreamRecordQueueEntry<CRow>)resultFuture;
-        try {
-            if (null == future.get()) {
-                resultFuture.completeExceptionally(new TimeoutException("Async function call has timed out."));
-            }
-        } catch (Exception e) {
-            resultFuture.completeExceptionally(new Exception(e));
+
+        if(timeOutNum % TIMEOUT_LOG_FLUSH_NUM == 0){
+            LOG.info("Async function call has timed out. input:{}, timeOutNum:{}",input.toString(), timeOutNum);
+        }
+
+        timeOutNum ++;
+        if(timeOutNum > sideInfo.getSideTableInfo().getAsyncTimeoutNumLimit()){
+            resultFuture.completeExceptionally(new Exception("Async function call timedoutNum beyond limit."));
+        } else {
+            resultFuture.complete(null);
         }
     }
 

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

@@ -53,6 +53,8 @@ public abstract class SideTableInfo extends TableInfo implements Serializable {
 
     public static final String ASYNC_TIMEOUT_KEY = "asyncTimeout";
 
+    public static final String ASYNC_TIMEOUT_NUM_KEY = "asyncTimeoutNum";
+
     private String cacheType = "none";//None or LRU or ALL
 
     private int cacheSize = 10000;
@@ -63,6 +65,8 @@ public abstract class SideTableInfo extends TableInfo implements Serializable {
 
     private int  asyncTimeout=10000;
 
+    private int asyncTimeoutNumLimit = Integer.MAX_VALUE;
+
     private boolean partitionedJoin = false;
 
     private String cacheMode="ordered";
@@ -143,4 +147,13 @@ public void setPredicateInfoes(List<PredicateInfo> predicateInfoes) {
     public List<PredicateInfo> getPredicateInfoes() {
         return predicateInfoes;
     }
+
+    public int getAsyncTimeoutNumLimit() {
+        return asyncTimeoutNumLimit;
+    }
+
+    public void setAsyncTimeoutNumLimit(int asyncTimeoutNumLimit) {
+        this.asyncTimeoutNumLimit = asyncTimeoutNumLimit;
+    }
+
 }

core/src/main/java/com/dtstack/flink/sql/table/AbsSideTableParser.java

@@ -108,6 +108,12 @@ protected void parseCacheProp(SideTableInfo sideTableInfo, Map<String, Object> p
                 }
                 sideTableInfo.setAsyncTimeout(asyncTimeout);
             }
+            if(props.containsKey(SideTableInfo.ASYNC_TIMEOUT_NUM_KEY.toLowerCase())){
+                Integer asyncTimeoutNum = MathUtil.getIntegerVal(props.get(SideTableInfo.ASYNC_TIMEOUT_NUM_KEY.toLowerCase()));
+                if (asyncTimeoutNum > 0){
+                    sideTableInfo.setAsyncTimeoutNumLimit(asyncTimeoutNum);
+                }
+            }
         }
     }
 }

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java

@@ -142,7 +142,7 @@ public void writeRecord(Tuple2 record) throws IOException {
             outRecords.inc();
         } catch (KuduException e) {
             if(outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0){
-                LOG.error("record insert failed ..{}", row.toString().substring(0, 100));
+                LOG.error("record insert failed, total dirty record:{} current row:{}", outDirtyRecords.getCount(), row.toString());
                 LOG.error("", e);
             }
             outDirtyRecords.inc();