public static DataStream<BaseRow> getSideJoinDataStream(DataStream inputStream, String sideType, String sqlRootDir, RowTypeInfo rowTypeInfo, JoinInfo joinInfo,
                                                            List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo, String pluginLoadMode) throws Exception {
        BaseAsyncReqRow asyncDbReq = loadAsyncReq(sideType, sqlRootDir, rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo, pluginLoadMode);

        //TODO How much should be set for the degree of parallelism? Timeout? capacity settings?
        if (ORDERED.equals(sideTableInfo.getCacheMode())){
            return AsyncDataStream.orderedWait(inputStream, asyncDbReq, -1, TimeUnit.MILLISECONDS, sideTableInfo.getAsyncCapacity())
                    .setParallelism(sideTableInfo.getParallelism());
        }else {
            return AsyncDataStream.unorderedWait(inputStream, asyncDbReq, -1, TimeUnit.MILLISECONDS, sideTableInfo.getAsyncCapacity())
                    .setParallelism(sideTableInfo.getParallelism());
        }

    }
fill data with lru cache
get data from  External data source with async operator
Date: 2018/9/18
Company: www.dtstack.com
@author xuchao
core/src/main/java/com/dtstack/flink/sql/side/operator/SideAsyncOperator.java