@Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        initCache();
        LOG.info("----- all cacheRef init end-----");

        //start reload cache thread
        AbstractSideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
        es = new ScheduledThreadPoolExecutor(1, new DTThreadFactory("cache-all-reload"));
        es.scheduleAtFixedRate(() -> reloadCache(), sideTableInfo.getCacheTimeout(), sideTableInfo.getCacheTimeout(), TimeUnit.MILLISECONDS);
    }
Reason:
Date: 2018/9/18
Company: www.dtstack.com

@author xuchao
core/src/main/java/com/dtstack/flink/sql/side/BaseAllReqRow.java