問(wèn)題現(xiàn)象

一個(gè)使用10秒滾動(dòng)窗口的任務(wù)在平穩(wěn)運(yùn)行一段時(shí)間之后出現(xiàn)了頻繁的重啟。在TaskManager日志中能看到以下文本：

2019-03-17 16:05:28,854 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  - RECEIVED SIGNAL 15: SIGTERM. Shutting down as requested.

原因定位

• 首先可以看到是YarnTaskExecutorRunner收到了SIGTERM信號(hào), 因?yàn)槭遣渴鹪赮arn上，所以基本可以定位到是Yarn因?yàn)槭裁丛驈腛S的層面將這個(gè)進(jìn)程給Kill掉的。
  • 代碼上也可以根據(jù)這個(gè)日志可以定位到Flink的SignalHandler，下圖可以看到Handler的構(gòu)造調(diào)用過(guò)程。不管是YarnSessionClusterEntrypoint還是YarnTaskExecutorRunner的主函數(shù)都會(huì)注冊(cè)，并且會(huì)在接收到OS的"TERM", "HUP", "INT"信號(hào)是打出日志。

    private static class Handler implements sun.misc.SignalHandler {
        private final Logger LOG;
        private final sun.misc.SignalHandler prevHandler;
        Handler(String name, Logger LOG) {
            this.LOG = LOG;
            prevHandler = Signal.handle(new Signal(name), this);
        }
        /**
         * Handle an incoming signal.
         *
         * @param signal    The incoming signal
         */
        @Override
        public void handle(Signal signal) {
            LOG.info("RECEIVED SIGNAL {}: SIG{}. Shutting down as requested.",
                signal.getNumber(),
                signal.getName());
            prevHandler.handle(signal);
        }
    }
  

  Handler構(gòu)造器的調(diào)用
• 接著可以觀察這個(gè)TaskManager所在機(jī)器的Yarn的NodeManager的日志，grep出和這個(gè)容器相關(guān)的日志，可以看到最后如下。TaskManager內(nèi)存超出了物理內(nèi)存的限制。但是從GC日志來(lái)看，連Full GC都很少發(fā)生。

  2019-03-19 16:48:10,647 INFO   org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainersMonitorImpl: Memory usage of ProcessTree 10265 for container-id container_1541225469893_4985_01_000005: 3.4 GB of 4 GB physical memory used; 5.4 GB of 8.4 GB virtual memory used
  

• 因?yàn)殚_(kāi)了taskmanager.memory.off-heap=true選項(xiàng)，所以Flink內(nèi)部也會(huì)使用一些堆外的內(nèi)存。還有就是RocksDB也會(huì)直接通過(guò)malloc分配內(nèi)存。

堆外內(nèi)存排查（大型繞彎路現(xiàn)場(chǎng)，想知道直接原因可以直接跳到最后）

• 在開(kāi)啟堆外內(nèi)存優(yōu)化時(shí)，F(xiàn)link的MemoryManager和NetworkBufferPool會(huì)使用ByteBuffer.allocateDirect方法來(lái)創(chuàng)建DirectByteBuffer，以此來(lái)使用堆外內(nèi)存。但是通過(guò)heap dump，可以看到DirectByteBuffer數(shù)量及其有限，因?yàn)槭褂玫氖悄J(rèn)的taskmanager.memory.segment-size，也就是32KB，所以占用的堆外內(nèi)存也只有幾百兆，而我預(yù)留了兩點(diǎn)幾G的堆外內(nèi)存，顯然不是這個(gè)引起的。這時(shí)問(wèn)題排查一度陷入了死胡同。

  Class Name               | Objects | Shallow Heap | Retained Heap
  ------------------------------------------------------------------
  java.nio.DirectByteBuffer|   9,470 |      606,080 |    >= 606,096
  ------------------------------------------------------------------
  

• 還有一個(gè)可疑的點(diǎn)是RocksDB，但是很難去排查它到底占用了多少內(nèi)存。
• 在大佬的幫助下，在性能環(huán)境上安裝了Jemalloc來(lái)代替原來(lái)的malloc，關(guān)于jemalloc的安裝參考文檔
• 并且在flink-conf.yaml中配置如下參數(shù)，將其注入到container的系統(tǒng)環(huán)境變量中，使其生效。這樣可以定期把memory profile dump出來(lái)，進(jìn)行分析, 發(fā)現(xiàn)最后malloc最多的是rocksdb的rocksdb::UncompressBlockContentsForCompressionType方法，并且最終占到了2.15G內(nèi)存的47%，TaskManager也被Yarn給kill掉。

  containerized.master.env.LD_PRELOAD: "/opt/jemalloc/lib/libjemalloc.so.2"
  containerized.master.env.MALLOC_CONF: "prof:true,lg_prof_interval:25,lg_prof_sample:17"
  

• 使用/opt/jemalloc/bin/jeprof --show_bytes /opt/java/bin/java jeprof.xxx 來(lái)分析dump文件，使用top來(lái)顯示對(duì)調(diào)用malloc最多的方法
• 對(duì)比前后dump文件如下

  Using local file /opt/java/bin/java.
  Using local file jeprof.18091.9812.i9812.heap.
  Welcome to jeprof!  For help, type 'help'.
  (jeprof) top
  Total: 1107642087 B
  884271340  79.8%  79.8% 884271340  79.8% os::malloc@921040
  150994944  13.6%  93.5% 150994944  13.6% rocksdb::Arena::AllocateNewBlock
  51761789   4.7%  98.1% 51761789   4.7% rocksdb::UncompressBlockContentsForCompressionType
   5242880   0.5%  98.6%  5242880   0.5% init
   5184828   0.5%  99.1%  5184828   0.5% updatewindow
   4204536   0.4%  99.5%  4204536   0.4% readCEN
   1643018   0.1%  99.6%  1643018   0.1% std::basic_string::_Rep::_S_create
   1346886   0.1%  99.7%  1346886   0.1% inflateInit2_
    917840   0.1%  99.8%  1181009   0.1% rocksdb::LRUCacheShard::Insert
    393336   0.0%  99.8% 52155125   4.7% rocksdb::BlockBasedTable::GetTableProperties
  

  (jeprof) top
  Total: 2259309361 B
  1062208712  47.0%  47.0% 1062208712  47.0% rocksdb::UncompressBlockContentsForCompressionType
  884120659  39.1%  86.1% 884120659  39.1% os::malloc@921040
  285348930  12.6%  98.8% 285348930  12.6% rocksdb::Arena::AllocateNewBlock
   5451379   0.2%  99.0%  5451379   0.2% std::basic_string::_Rep::_S_create
   5242880   0.2%  99.3%  5242880   0.2% init
   5036690   0.2%  99.5%  5036690   0.2% updatewindow
   4204536   0.2%  99.7%  4204536   0.2% readCEN
   2621559   0.1%  99.8%  2621559   0.1% rocksdb::WritableFileWriter::Append
   1346886   0.1%  99.8%  1346886   0.1% inflateInit2_
    524472   0.0%  99.9%   788155   0.0% rocksdb::LRUCacheShard::Insert
  

jeprof.pdf

• 在搜索這個(gè)方法后，發(fā)現(xiàn)有這個(gè)issue的github issue鏈接，實(shí)際上也不是Memory Leak，在默認(rèn)配置下，rocksdb會(huì)為所有flush的sst文件在內(nèi)存中保留索引，索引會(huì)隨著文件數(shù)越來(lái)越多而占用更多的內(nèi)存空間，如果限制內(nèi)存中索引的消耗，會(huì)導(dǎo)致經(jīng)常需要去sst文件中獲取元信息來(lái)搜索，大量增加io消耗(這塊不是特別熟悉，有可能說(shuō)的有點(diǎn)問(wèn)題)，那為什么RocksDB文件會(huì)不停膨脹？

最終問(wèn)題定位(走完彎路)

• RocksDB文件不斷膨脹，可以從checkpoint的大小來(lái)看出來(lái)，將incremental checkpoint關(guān)閉后，發(fā)現(xiàn)每次Checkpoint大小都在遞增，但是用戶代碼的邏輯實(shí)際是使用一個(gè)10s的滾動(dòng)窗口，不應(yīng)該會(huì)出現(xiàn)這樣的情況。
• 之后在flink窗口算子中加了幾行日志，如下所示，以ClarkTest開(kāi)頭

    @Override
    public void onProcessingTime(InternalTimer<K, W> timer) throws Exception {
        triggerContext.key = timer.getKey();
        triggerContext.window = timer.getNamespace();

        MergingWindowSet<W> mergingWindows;

        if (windowAssigner instanceof MergingWindowAssigner) {
            mergingWindows = getMergingWindowSet();
            W stateWindow = mergingWindows.getStateWindow(triggerContext.window);
            if (stateWindow == null) {
                // Timer firing for non-existent window, this can only happen if a
                // trigger did not clean up timers. We have already cleared the merging
                // window and therefore the Trigger state, however, so nothing to do.
                return;
            } else {
                windowState.setCurrentNamespace(stateWindow);
            }
        } else {
            windowState.setCurrentNamespace(triggerContext.window);
            mergingWindows = null;
        }

        TriggerResult triggerResult = triggerContext.onProcessingTime(timer.getTimestamp());

        int randomInt = random.nextInt(1000);
        if (triggerResult.isFire()) {
            ACC contents = windowState.get();
            if (randomInt == 1) {
                LOG.info("ClarkTest: Window state namespace: " + triggerContext.window + " and key " + triggerContext.key);
                LOG.info("ClarkTest: Window state value is going to fire is null ? " + (windowState.get() == null));
            }
            if (contents != null) {
                emitWindowContents(triggerContext.window, contents);
            }
        }

        if (triggerResult.isPurge()) {
            if (randomInt == 1) {
                LOG.info("ClarkTest: Window state get purged. ");
            }
            windowState.clear();
        }

        if (!windowAssigner.isEventTime() && isCleanupTime(triggerContext.window, timer.getTimestamp())) {
            windowState.setCurrentNamespace(triggerContext.window);
            if (randomInt == 1) {
                LOG.info("ClarkTest: Window State namespace before cleaning: " + triggerContext.window + " and key " + triggerContext.key);
                LOG.info("ClarkTest: Window state value before clear is null ? " + (windowState.get() == null));
            }
            clearAllState(triggerContext.window, windowState, mergingWindows);
            if (randomInt == 1) {
                LOG.info("ClarkTest: Window state value after clear is null ? " + (windowState.get() == null));
            }

        }

        if (mergingWindows != null) {
            // need to make sure to update the merging state in state
            mergingWindows.persist();
        }
    }

• 發(fā)現(xiàn)每次在emitWindowContents之前window state的結(jié)果都不為null，但是在clean up之前，window state的結(jié)果已經(jīng)變成了null。說(shuō)明在這兩段邏輯之間出了什么問(wèn)題。通過(guò)將key打印出來(lái)，發(fā)現(xiàn)前后key有所變化，所以最后確定是用戶代碼的process function中改變了keyby的key的值導(dǎo)致窗口狀態(tài)無(wú)法清理
• 最后總結(jié)就是在keyby的時(shí)候key一定要是不變量，不然有可能導(dǎo)致?tīng)顟B(tài)無(wú)法清理。還有就是在分布式系統(tǒng)中，大量使用不變量是規(guī)避風(fēng)險(xiǎn)的最佳途徑之一。