print detail error log when async side load data

Author: todd5167

.gitignore

@@ -11,5 +11,5 @@ plugins/
 lib/
 .vertx/
 bin/nohup.out
-
+.DS_Store
 bin/sideSql.txt

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

@@ -256,7 +256,7 @@ public void onFailure(Throwable t) {
                         t.getMessage());
                 System.out.println("Failed to retrieve the data: " + t.getMessage());
                 cluster.closeAsync();
-                resultFuture.complete(null);
+                resultFuture.completeExceptionally(t);
             }
         });
     }

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

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql.side;
 
@@ -28,9 +28,12 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
 import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
+import org.apache.flink.streaming.api.operators.async.queue.StreamRecordQueueEntry;
 import org.apache.flink.types.Row;
 
+import java.util.Collection;
 import java.util.Collections;
+import java.util.concurrent.TimeoutException;
 
 /**
  * All interfaces inherit naming rules: type + "AsyncReqRow" such as == "MysqlAsyncReqRow
@@ -50,6 +53,18 @@ public AsyncReqRow(SideInfo sideInfo){
         this.sideInfo = sideInfo;
     }
 
+    @Override
+    public void timeout(Row input, ResultFuture<Row> resultFuture) throws Exception {
+        StreamRecordQueueEntry<Row> future = (StreamRecordQueueEntry<Row>)resultFuture;
+        try {
+            if (null == future.get()) {
+                new TimeoutException("Async function call has timed out.");
+            }
+        } catch (Exception e) {
+            throw new Exception(e);
+        }
+    }
+
     private void initCache(){
         SideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
         if(sideTableInfo.getCacheType() == null || ECacheType.NONE.name().equalsIgnoreCase(sideTableInfo.getCacheType())){

hbase/hbase-side/hbase-async-side/pom.xml

@@ -50,6 +50,7 @@
                             <artifactSet>
                                 <excludes>
                                     <exclude>org.apache.hadoop:hadoop-common</exclude>
+                                    <exclude>org.apache.hadoop:hadoop-yarn-common</exclude>
                                     <exclude>org.apache.hadoop:hadoop-auth</exclude>
                                     <exclude>org.apache.hadoop:hadoop-mapreduce-client-core</exclude>
                                     <exclude>org.slf4j:*</exclude>

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/RowKeyEqualModeDealer.java

@@ -106,7 +106,7 @@ public void asyncGetData(String tableName, String rowKeyStr, Row input, ResultFu
                     }
                 }
             }catch (Exception e){
-                resultFuture.complete(null);
+                resultFuture.completeExceptionally(e);
                 LOG.error("record:" + input);
                 LOG.error("get side record exception:", e);
             }

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

@@ -92,7 +92,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                     }
                     resultFuture.complete(rowList);
                 } else {
-                    throw new RuntimeException("not support cache obj type " + val.getType());
+                    resultFuture.completeExceptionally(new RuntimeException("not support cache obj type " + val.getType()));
                 }
                 return;
             }
@@ -110,7 +110,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
             connection.queryWithParams(sqlCondition, inputParams, rs -> {
                 if (rs.failed()) {
                     LOG.error("Cannot retrieve the data from the database", rs.cause());
-                    resultFuture.complete(null);
+                    resultFuture.completeExceptionally(rs.cause());
                     return;
                 }
 

redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java

@@ -148,7 +148,8 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                     }
                     resultFuture.complete(rowList);
                 }else{
-                    throw new RuntimeException("not support cache obj type " + val.getType());
+                    RuntimeException exception = new RuntimeException("not support cache obj type " + val.getType());
+                    resultFuture.completeExceptionally(exception);
                 }
                 return;
             }