问题现象
一个使用10秒滚动窗口的任务在平稳运行一段时间之后出现了频繁的重启。在TaskManager日志中能看到以下文本：
2019-03-17 16:05:28,854 INFO org.apache.flink.yarn.YarnTaskExecutorRunner - RECEIVED SIGNAL 15: SIGTERM. Shutting down as requested.
原因定位
- 首先可以看到是YarnTaskExecutorRunner收到了SIGTERM信号, 因为是部署在Yarn上，所以基本可以定位到是Yarn因为什么原因从OS的层面将这个进程给Kill掉的。
- 代码上也可以根据这个日志可以定位到Flink的SignalHandler，下图可以看到Handler的构造调用过程。不管是YarnSessionClusterEntrypoint还是YarnTaskExecutorRunner的主函数都会注册，并且会在接收到OS的"TERM", "HUP", "INT"信号是打出日志。
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构造器的调用 - 接着可以观察这个TaskManager所在机器的Yarn的NodeManager的日志，grep出和这个容器相关的日志，可以看到最后如下。TaskManager内存超出了物理内存的限制。但是从GC日志来看，连Full GC都很少发生。
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 - 因为开了taskmanager.memory.off-heap=true选项，所以Flink内部也会使用一些堆外的内存。还有就是RocksDB也会直接通过malloc分配内存。
堆外内存排查（大型绕弯路现场，想知道直接原因可以直接跳到最后）
- 在开启堆外内存优化时，Flink的MemoryManager和NetworkBufferPool会使用ByteBuffer.allocateDirect方法来创建DirectByteBuffer，以此来使用堆外内存。但是通过heap dump，可以看到DirectByteBuffer数量及其有限，因为使用的是默认的taskmanager.memory.segment-size，也就是32KB，所以占用的堆外内存也只有几百兆，而我预留了两点几G的堆外内存，显然不是这个引起的。这时问题排查一度陷入了死胡同。
Class Name | Objects | Shallow Heap | Retained Heap ------------------------------------------------------------------ java.nio.DirectByteBuffer| 9,470 | 606,080 | >= 606,096 ------------------------------------------------------------------ - 还有一个可疑的点是RocksDB，但是很难去排查它到底占用了多少内存。
- 在大佬的帮助下，在性能环境上安装了Jemalloc来代替原来的malloc，关于jemalloc的安装参考文档
- 并且在flink-conf.yaml中配置如下参数，将其注入到container的系统环境变量中，使其生效。这样可以定期把memory profile dump出来，进行分析, 发现最后malloc最多的是rocksdb的rocksdb::UncompressBlockContentsForCompressionType方法，并且最终占到了2.15G内存的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 来分析dump文件，使用top来显示对调用malloc最多的方法
- 对比前后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
- 在搜索这个方法后，发现有这个issue的github issue链接，实际上也不是Memory Leak，在默认配置下，rocksdb会为所有flush的sst文件在内存中保留索引，索引会随着文件数越来越多而占用更多的内存空间，如果限制内存中索引的消耗，会导致经常需要去sst文件中获取元信息来搜索，大量增加io消耗(这块不是特别熟悉，有可能说的有点问题)，那为什么RocksDB文件会不停膨胀？
最终问题定位(走完弯路)
- RocksDB文件不断膨胀，可以从checkpoint的大小来看出来，将incremental checkpoint关闭后，发现每次Checkpoint大小都在递增，但是用户代码的逻辑实际是使用一个10s的滚动窗口，不应该会出现这样的情况。
- 之后在flink窗口算子中加了几行日志，如下所示，以ClarkTest开头
@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();
}
}
- 发现每次在emitWindowContents之前window state的结果都不为null，但是在clean up之前，window state的结果已经变成了null。说明在这两段逻辑之间出了什么问题。通过将key打印出来，发现前后key有所变化，所以最后确定是用户代码的process function中改变了keyby的key的值导致窗口状态无法清理
- 最后总结就是在keyby的时候key一定要是不变量，不然有可能导致状态无法清理。还有就是在分布式系统中，大量使用不变量是规避风险的最佳途径之一。
