final protected void doAsyncQueryData(
            Map<String, Object> inputParams,
            BaseRow input,
            ResultFuture<BaseRow> resultFuture,
            SQLClient rdbSqlClient,
            AtomicLong failCounter,
            AtomicBoolean finishFlag,
            CountDownLatch latch) {
        rdbSqlClient.getConnection(conn -> {
            try {
                String errorMsg;
                Integer retryMaxNum = sideInfo.getSideTableInfo().getConnectRetryMaxNum(3);
                int logPrintTime = retryMaxNum / errorLogPrintNum == 0 ?
                        retryMaxNum : retryMaxNum / errorLogPrintNum;
                if (conn.failed()) {
                    connectionStatus.set(false);
                    errorMsg = ExceptionTrace.traceOriginalCause(conn.cause());
                    if (failCounter.getAndIncrement() % logPrintTime == 0) {
                        LOG.error("getConnection error. cause by " + errorMsg);
                    }
                    LOG.error(String.format("retry ... current time [%s]", failCounter.get()));
                    if (failCounter.get() >= retryMaxNum) {
                        resultFuture.completeExceptionally(
                                new SuppressRestartsException(conn.cause())
                        );
                        finishFlag.set(true);
                    }
                    return;
                }
                connectionStatus.set(true);
                registerTimerAndAddToHandler(input, resultFuture);

                handleQuery(conn.result(), inputParams, input, resultFuture);
                finishFlag.set(true);
            } catch (Exception e) {
                dealFillDataError(input, resultFuture, e);
            } finally {
                latch.countDown();
            }
        });
    }
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