fix bug 26199

Author: simenliuxing

cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java

@@ -18,37 +18,30 @@
 
 package com.dtstack.flink.sql.side.cassandra;
 
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.types.Row;
-import org.apache.flink.util.Collector;
-
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.ConsistencyLevel;
-import com.datastax.driver.core.HostDistance;
-import com.datastax.driver.core.PoolingOptions;
-import com.datastax.driver.core.QueryOptions;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.*;
 import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
+import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.BaseAllReqRow;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
-import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
+import com.dtstack.flink.sql.util.RowDataComplete;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.InetAddress;
 import java.sql.SQLException;
-import java.sql.Timestamp;
 import java.util.ArrayList;
 import java.util.Calendar;
 import java.util.List;
@@ -124,14 +117,14 @@ protected void reloadCache() {
 
 
     @Override
-    public void flatMap(Tuple2<Boolean,Row> input, Collector<Tuple2<Boolean,Row>> out) throws Exception {
+    public void flatMap(Tuple2<Boolean, Row> input, Collector<Tuple2<Boolean, BaseRow>> out) throws Exception {
         List<Object> inputParams = Lists.newArrayList();
         for (Integer conValIndex : sideInfo.getEqualValIndex()) {
             Object equalObj = input.f1.getField(conValIndex);
             if (equalObj == null) {
                 if (sideInfo.getJoinType() == JoinType.LEFT) {
-                    Row data = fillData(input.f1, null);
-                    out.collect(Tuple2.of(input.f0, data));
+                    Row row = fillData(input.f1, null);
+                    RowDataComplete.collectTupleRow(out, new Tuple2<>(input.f0, row));
                 }
                 return;
             }
@@ -144,7 +137,7 @@ public void flatMap(Tuple2<Boolean,Row> input, Collector<Tuple2<Boolean,Row>> ou
         if (CollectionUtils.isEmpty(cacheList)) {
             if (sideInfo.getJoinType() == JoinType.LEFT) {
                 Row row = fillData(input.f1, null);
-                out.collect(Tuple2.of(input.f0, row));
+                RowDataComplete.collectTupleRow(out, new Tuple2<>(input.f0, row));
             } else {
                 return;
             }
@@ -153,7 +146,8 @@ public void flatMap(Tuple2<Boolean,Row> input, Collector<Tuple2<Boolean,Row>> ou
         }
 
         for (Map<String, Object> one : cacheList) {
-            out.collect(Tuple2.of(input.f0, fillData(input.f1, one)));
+            Row row = fillData(input.f1, one);
+            RowDataComplete.collectTupleRow(out, new Tuple2<>(input.f0, row));
         }
 
     }

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

@@ -19,10 +19,12 @@
 
 package com.dtstack.flink.sql.side.cassandra;
 
+import com.dtstack.flink.sql.util.RowDataComplete;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
+import org.apache.flink.table.dataformat.BaseRow;
 import org.apache.flink.types.Row;
 
 import com.datastax.driver.core.Cluster;
@@ -161,7 +163,7 @@ private void connCassandraDB(CassandraSideTableInfo tableInfo) {
     }
 
     @Override
-    public void handleAsyncInvoke(Map<String, Object> inputParams, Tuple2<Boolean,Row> input, ResultFuture<Tuple2<Boolean,Row>> resultFuture) throws Exception {
+    public void handleAsyncInvoke(Map<String, Object> inputParams, Tuple2<Boolean,Row> input, ResultFuture<Tuple2<Boolean, BaseRow>> resultFuture) throws Exception {
 
         String key = buildCacheKey(inputParams);
         //connect Cassandra
@@ -200,7 +202,7 @@ public void onSuccess(List<com.datastax.driver.core.Row> rows) {
                         }
                         rowList.add(Tuple2.of(input.f0,row));
                     }
-                    resultFuture.complete(rowList);
+                    RowDataComplete.completeTupleRows(resultFuture, rowList);
                     if (openCache()) {
                         putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));
                     }

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

@@ -17,9 +17,9 @@
  */
 
 
-
 package com.dtstack.flink.sql.side;
 
+import com.dtstack.flink.sql.util.RowDataComplete;
 import org.apache.flink.api.common.functions.RichFlatMapFunction;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
@@ -44,10 +44,11 @@
  * Reason:
  * Date: 2018/9/18
  * Company: www.dtstack.com
+ *
  * @author xuchao
  */
 
-public abstract class BaseAllReqRow extends RichFlatMapFunction<Tuple2<Boolean,Row>, Tuple2<Boolean,BaseRow>> implements ISideReqRow {
+public abstract class BaseAllReqRow extends RichFlatMapFunction<Tuple2<Boolean, Row>, Tuple2<Boolean, BaseRow>> implements ISideReqRow {
 
     private static final Logger LOG = LoggerFactory.getLogger(BaseAllReqRow.class);
 
@@ -57,7 +58,7 @@ public abstract class BaseAllReqRow extends RichFlatMapFunction<Tuple2<Boolean,R
 
     private ScheduledExecutorService es;
 
-    public BaseAllReqRow(BaseSideInfo sideInfo){
+    public BaseAllReqRow(BaseSideInfo sideInfo) {
         this.sideInfo = sideInfo;
 
     }
@@ -74,7 +75,7 @@ public void open(Configuration parameters) throws Exception {
 
         //start reload cache thread
         AbstractSideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
-        es = new ScheduledThreadPoolExecutor(1,new DTThreadFactory("cache-all-reload"));
+        es = new ScheduledThreadPoolExecutor(1, new DTThreadFactory("cache-all-reload"));
         es.scheduleAtFixedRate(() -> reloadCache(), sideTableInfo.getCacheTimeout(), sideTableInfo.getCacheTimeout(), TimeUnit.MILLISECONDS);
     }
 
@@ -88,12 +89,12 @@ protected Object convertTimeIndictorTypeInfo(Integer index, Object obj) {
         return obj;
     }
 
-    protected void sendOutputRow(Tuple2<Boolean, Row> value, Object sideInput, Collector<Tuple2<Boolean, Row>> out) {
+    protected void sendOutputRow(Tuple2<Boolean, Row> value, Object sideInput, Collector<Tuple2<Boolean, BaseRow>> out) {
         if (sideInput == null && sideInfo.getJoinType() != JoinType.LEFT) {
             return;
         }
         Row row = fillData(value.f1, sideInput);
-        out.collect(Tuple2.of(value.f0, row));
+        RowDataComplete.collectTupleRow(out, Tuple2.of(value.f0, row));
     }
 
     @Override