private void handleQuery(SQLConnection connection, Map<String, Object> inputParams, BaseRow input, ResultFuture<BaseRow> resultFuture) {
        String key = buildCacheKey(inputParams);
        JsonArray params = new JsonArray(Lists.newArrayList(inputParams.values()));
        connection.queryWithParams(sideInfo.getSqlCondition(), params, rs -> {
            try {
                if (rs.failed()) {
                    LOG.error(
                            String.format("\nget data with sql [%s] failed! \ncause: [%s]",
                                    sideInfo.getSqlCondition(),
                                    rs.cause().getMessage()
                            )
                    );
                    dealFillDataError(input, resultFuture, rs.cause());
                    return;
                }

                List<JsonArray> cacheContent = Lists.newArrayList();

                int resultSize = rs.result().getResults().size();
                if (resultSize > 0) {
                    List<BaseRow> rowList = Lists.newArrayList();

                    for (JsonArray line : rs.result().getResults()) {
                        BaseRow row = fillData(input, line);
                        if (openCache()) {
                            cacheContent.add(line);
                        }
                        rowList.add(row);
                    }

                    if (openCache()) {
                        putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));
                    }
                    RowDataComplete.completeBaseRow(resultFuture, rowList);
                } else {
                    dealMissKey(input, resultFuture);
                    if (openCache()) {
                        putCache(key, CacheMissVal.getMissKeyObj());
                    }
                }
            } finally {
                // and close the connection
                connection.close(done -> {
                    if (done.failed()) {
                        LOG.error("sql connection close failed! " +
                                ExceptionTrace.traceOriginalCause(done.cause())
                        );
                    }
                });
            }
        });
    }
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