print detail error log when async side load data

simenliuxing · Jul 20, 2019 · 67990fa · 67990fa
1 parent 62ee4f9
commit 67990fa
Show file tree

Hide file tree

Showing 7 changed files with 24 additions and 7 deletions.
diff --git a/.gitignore b/.gitignore
@@ -11,5 +11,5 @@ plugins/
 lib/
 .vertx/
 bin/nohup.out
-
+.DS_Store
 bin/sideSql.txt
diff --git a/...a-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java b/...a-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);
             }
         });
     }

diff --git a/core/src/main/java/com/dtstack/flink/sql/side/AsyncReqRow.java b/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())){

diff --git a/hbase/hbase-side/hbase-async-side/pom.xml b/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>

diff --git a/...de/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/RowKeyEqualModeDealer.java b/...de/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);
             }

diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java b/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;
                 }
 

diff --git a/...ide/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java b/...ide/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;
             }