问题描述
flinksql从kafka读取数据，异步函数加载Mysql数据进行维表关联，最后将数据写入到mysql中。任务在启动时会消费kafka数据，一段时间后不读kafka或者能够持续读kafka数据但是异步函数不发送给下游算子。
- 不读kafka数据：kafka读取线程像卡住一样，从kafka中读取不到数据，以为是网络原因，但是计算节点和工作节点在同一台机器中，于是排除网络原因。
- 持续读kafka数据，但是异步函数不下发数据：以为是设置的异步超时间超时，默认是10s，增大超时时间后依然不下发。
Jstack 排查
打印执行线程堆栈信息，虽然BLOCKED状态的线程很多，但大部分是第三方类的执行线程，都比较正常。突然发现和我们程序有关的代码阻塞线程。
方法栈信息
原来是调用我们的timeout函数出现了阻塞。
public void timeout(Row input, ResultFuture<Row> resultFuture) {
resultFuture.completeExceptionally(new TimeoutException("Async function call has timed out."));
}
阻塞原因
在flink异步函数asyncInvoke中，只处理了正常逻辑。也就是匹配上调用resultFuture.complete(rowList);但是fillData里面进行数据类型转换时很容易发生异常，当发生异常时，resultFuture并没有结果输出，从而导致整个链路阻塞。
List<Row> rowList = Lists.newArrayList();
for (Object jsonArray : (List) val.getContent()) {
Row row = fillData(input, jsonArray);
rowList.add(row);
}
resultFuture.complete(rowList);
解决以及注意事项
fillData进行try-catch捕获发生异常时调用resultFuture.completeExceptionally(exception);
在flink异步函数中，resultFuture.complete()只会被调用一次，complete一个集合需要先在填充然后一次性发送，而不是通过遍历调用多次resultFuture.complete()
使用异步Future一定要记得有输出值。
堆栈信息重点关注有没有我们自己的逻辑 。
