RichAsyncFunction Timeout

[Polarisary]

Hi ALL,

When I use RichAsyncFunction read data from hbase, it always timeout after a few minutes. but the hbase connection is not close, it also can get data in the override method timeout.

Following is the code, does somebody know why trigger timeout.

====================================================================

AsyncDataStream.unorderedWait(uidDs, new AsyncHBaseRequest(hTableName,
                HBaseConfigurationUtil.serializeConfiguration(hbaseClientConf), hbaseSchema)
        , 5, TimeUnit.MINUTES, 10)

@Override
public void timeout(Tuple1<Long> input, ResultFuture<Tuple7<Short, Long, Short, Short, Long, Integer, Long>> resultFuture) throws Exception {
    
    Row r = readHelper.parseToRow(table.get(readHelper.createGet("1349126000000001515587439bf2f217")));
    logger.error("Timeout Error, input [{}], conn {}, row [{}]", input.f0, hConnection.isClosed(), r.toString());
}

@Override
public void asyncInvoke(Tuple1<Long> input, ResultFuture<Tuple7<Short, Long, Short, Short, Long, Integer, Long>> resultFuture) throws Exception {
    FamilyFilter filter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("f1")));
    String rkStart = UserInstallAppLookupTableSource.getLeftRowkeyByUid(input.f0, 0);
    String rkEnd = UserInstallAppLookupTableSource.getLeftRowkeyByUid(input.f0, 9999999999999L);
    

[hidden email]

[vino yang]

Hi  Polarisary,

IMO, firstly, it would be better to monitor the OS and Flink/HBase metrics. For example:

• Flink and HBase cluster Network I/O metrics;
• Flink TM CPU/Memory/Backpressure metrics and so on;

You can view these metrics to find some potential reasons. If you can not figure it out, you can share these metrics with the community.

Best,

Vino

Polarisary <[hidden email]> 于2019年12月18日周三 上午11:09写道：

Hi ALL,

When I use RichAsyncFunction read data from hbase, it always timeout after a few minutes. but the hbase connection is not close, it also can get data in the override method timeout.

Following is the code, does somebody know why trigger timeout.

====================================================================

AsyncDataStream.unorderedWait(uidDs, new AsyncHBaseRequest(hTableName,
                HBaseConfigurationUtil.serializeConfiguration(hbaseClientConf), hbaseSchema)
        , 5, TimeUnit.MINUTES, 10)

@Override
public void timeout(Tuple1<Long> input, ResultFuture<Tuple7<Short, Long, Short, Short, Long, Integer, Long>> resultFuture) throws Exception {
    
    Row r = readHelper.parseToRow(table.get(readHelper.createGet("1349126000000001515587439bf2f217")));
    logger.error("Timeout Error, input [{}], conn {}, row [{}]", input.f0, hConnection.isClosed(), r.toString());
}

@Override
public void asyncInvoke(Tuple1<Long> input, ResultFuture<Tuple7<Short, Long, Short, Short, Long, Integer, Long>> resultFuture) throws Exception {
    FamilyFilter filter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("f1")));
    String rkStart = UserInstallAppLookupTableSource.getLeftRowkeyByUid(input.f0, 0);
    String rkEnd = UserInstallAppLookupTableSource.getLeftRowkeyByUid(input.f0, 9999999999999L);
    

[hidden email]

[Biao Liu]

Hi Polarisary,

It's hard to tell what happened without further detail. Just some guesses.

1. Have you completed the "resultFuture" in "asyncInvoke"? Asking this is because there is only a part of "asyncInvoke" implementation, I can't see the completion part.

2. The capacity (10) of async waiting queue is enough or not? The time of waiting queue available is also a part of the timeout calculation. It seems this behavior has been changed in master branch recently. I'm not sure if it's included or not in your version.

On Wed, 18 Dec 2019 at 11:09, Polarisary <[hidden email]> wrote:

Hi ALL,

When I use RichAsyncFunction read data from hbase, it always timeout after a few minutes. but the hbase connection is not close, it also can get data in the override method timeout.

Following is the code, does somebody know why trigger timeout.

====================================================================

AsyncDataStream.unorderedWait(uidDs, new AsyncHBaseRequest(hTableName,
                HBaseConfigurationUtil.serializeConfiguration(hbaseClientConf), hbaseSchema)
        , 5, TimeUnit.MINUTES, 10)

@Override
public void timeout(Tuple1<Long> input, ResultFuture<Tuple7<Short, Long, Short, Short, Long, Integer, Long>> resultFuture) throws Exception {
    
    Row r = readHelper.parseToRow(table.get(readHelper.createGet("1349126000000001515587439bf2f217")));
    logger.error("Timeout Error, input [{}], conn {}, row [{}]", input.f0, hConnection.isClosed(), r.toString());
}

@Override
public void asyncInvoke(Tuple1<Long> input, ResultFuture<Tuple7<Short, Long, Short, Short, Long, Integer, Long>> resultFuture) throws Exception {
    FamilyFilter filter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("f1")));
    String rkStart = UserInstallAppLookupTableSource.getLeftRowkeyByUid(input.f0, 0);
    String rkEnd = UserInstallAppLookupTableSource.getLeftRowkeyByUid(input.f0, 9999999999999L);
    

[hidden email]