@Override
    public void handleAsyncInvoke(Map<String, Object> inputParams, BaseRow input, ResultFuture<BaseRow> resultFuture) throws Exception {
        SQLClient sqlClient = clientShare ? rdbSqlClientPool.get(url) : this.rdbSqlClient;
        AtomicLong networkLogCounter = new AtomicLong(0L);
        //network is unhealthy
        while (!connectionStatus.get()) {
            if (networkLogCounter.getAndIncrement() % 1000 == 0) {
                LOG.info("network unhealthy to block task");
            }
            Thread.sleep(100);
        }
        Map<String, Object> params = formatInputParam(inputParams);
        executor.execute(() -> connectWithRetry(params, input, resultFuture, sqlClient));
    }
Date: 2018/11/26
Company: www.dtstack.com

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