作者：林小铂
在生产环境中，Flink 通常会部署在 YARN 或 k8s 等资源管理系统之上，进程会以容器化（YARN 容器或 docker 等容器）的方式运行，其资源会受到资源管理系统的严格限制。另一方面，Flink 运行在 JVM 之上，而 JVM 与容器化环境并不是特别适配，尤其 JVM 复杂且可控性较弱的内存模型，容易导致进程因使用资源超标而被 kill 掉，造成 Flink 应用的不稳定甚至不可用。
针对这个问题，Flink 在 1.10 版本对内存管理模块进行了重构，设计了全新的内存参数。在大多数场景下 Flink 的内存模型和默认已经足够好用，可以帮用户屏蔽进程背后的复杂内存结构，然而一旦出现内存问题，问题的排查和修复都需要比较多的领域知识，通常令普通用户望而却步。
为此，本文将解析 JVM 和 Flink 的内存模型，并总结在工作中遇到和在社区交流中了解到的造成 Flink 内存使用超出容器限制的常见原因。由于 Flink 内存使用与用户代码、部署环境、各种依赖版本等因素都有紧密关系，本文主要讨论 on YARN 部署、Oracle JDK/OpenJDK 8、Flink 1.10+ 的情况。此外，特别感谢 @宋辛童（Flink 1.10+ 新内存架构的主要作者）和 @唐云（RocksDB StateBackend 专家）在社区的答疑，令笔者受益匪浅。
JVM 内存分区
对于大多数 Java 用户而言，日常开发中与 JVM Heap 打交道的频率远大于其他 JVM 内存分区，因此常把其他内存分区统称为 Off-Heap 内存。而对于 Flink 来说，内存超标问题通常来自 Off-Heap 内存，因此对 JVM 内存模型有更深入的理解是十分必要的。
根据 JVM 8 Spec[1]，JVM 管理的内存分区如下图:
img1. JVM 8 内存模型
除了上述 Spec 规定的标准分区，在具体实现上 JVM 常常还会加入一些额外的分区供进阶功能模块使用。以 HotSopt JVM 为例，根据 Oracle NMT[5] 的标准，我们可以将 JVM 内存细分为如下区域:
● Heap: 各线程共享的内存区域，主要存放 new 操作符创建的对象，内存的释放由 GC 管理，可被用户代码或 JVM 本身使用。
● Class: 类的元数据，对应 Spec 中的 Method Area (不含 Constant Pool)，Java 8 中的 Metaspace。
● Thread: 线程级别的内存区，对应 Spec 中的 PC Register、Stack 和 Natvive Stack 三者的总和。
● Compiler: JIT (Just-In-Time) 编译器使用的内存。
● Code Cache: 用于存储 JIT 编译器生成的代码的缓存。
● GC: 垃圾回收器使用的内存。
● Symbol: 存储 Symbol (比如字段名、方法签名、Interned String) 的内存，对应 Spec 中的 Constant Pool。
● Arena Chunk: JVM 申请操作系统内存的临时缓存区。
● NMT: NMT 自己使用的内存。
● Internal: 其他不符合上述分类的内存，包括用户代码申请的 Native/Direct 内存。
● Unknown: 无法分类的内存。
理想情况下，我们可以严格控制各分区内存的上限，来保证进程总体内存在容器限额之内。但是过于严格的管理会带来会有额外使用成本且缺乏灵活度，所以在实际中为了 JVM 只对其中几个暴露给用户使用的分区提供了硬性的上限，而其他分区则可以作为整体被视为 JVM 本身的内存消耗。
具体可以用于限制分区内存的 JVM 参数如下表所示（值得注意的是，业界对于 JVM Native 内存并没有准确的定义，本文的 Native 内存指的是 Off-Heap 内存中非 Direct 的部分，与 Native Non-Direct 可以互换）。
从表中可以看到，使用 Heap、Metaspace 和 Direct 内存都是比较安全的，但非 Direct 的 Native 内存情况则比较复杂，可能是 JVM 本身的一些内部使用（比如下文会提到的 MemberNameTable），也可能是用户代码引入的 JNI 依赖，还有可能是用户代码自身通过 sun.misc.Unsafe 申请的 Native 内存。理论上讲，用户代码或第三方 lib 申请的 Native 内存需要用户来规划内存用量，而 Internal 的其余部分可以并入 JVM 本身的内存消耗。而实际上 Flink 的内存模型也遵循了类似的原则。
Flink TaskManager 内存模型
首先回顾下 Flink 1.10+ 的 TaskManager 内存模型。
img2. Flink TaskManager 内存模型
显然，Flink 框架本身不仅会包含 JVM 管理的 Heap 内存，也会申请自己管理 Off-Heap 的 Native 和 Direct 内存。在笔者看来，Flink 对于 Off-Heap 内存的管理策略可以分为三种:
● 硬限制（Hard Limit）: 硬限制的内存分区是 Self-Contained 的，Flink 会保证其用量不会超过设置的阈值（若内存不够则抛出类似 OOM 的异常），
● 软限制（Soft Limit）: 软限制意味着内存使用长期会在阈值以下，但可能短暂地超过配置的阈值。
● 预留（Reserved）: 预留意味着 Flink 不会限制分区内存的使用，只是在规划内存时预留一部分空间，但不能保证实际使用会不会超额。
结合 JVM 的内存管理来看，一个 Flink 内存分区的内存溢出会导致何种后果，判断逻辑如下:
1、若是 Flink 有硬限制的分区，Flink 会报该分区内存不足。否则进入下一步。
2、若该分区属于 JVM 管理的分区，在其实际值增长导致 JVM 分区也内存耗尽时，JVM 会报其所属的 JVM 分区的 OOM （比如 java.lang.OutOfMemoryError: Jave heap space）。否则进入下一步。
3、该分区内存持续溢出，最终导致进程总体内存超出容器内存限制。在开启严格资源控制的环境下，资源管理器（YARN/k8s 等）会 kill 掉该进程。
为直观地展示 Flink 各内存分区与 JVM 内存分区间的关系，笔者整理了如下的内存分区映射表:
img3. Flink 分区及 JVM 分区内存限制关系
根据之前的逻辑，在所有的 Flink 内存分区中，只有不是 Self-Contained 且所属 JVM 分区也没有内存硬限制参数的 JVM Overhead 是有可能导致进程被 OOM kill 掉的。作为一个预留给各种不同用途的内存的大杂烩，JVM Overhead 的确容易出问题，但同时它也可以作为一个兜底的隔离缓冲区，来缓解来自其他区域的内存问题。
举个例子，Flink 内存模型在计算 Native Non-Direct 内存时有一个 trick:
Although, native non-direct memory usage can be accounted for as a part of the framework off-heap memory or task off-heap memory, it will result in a higher JVM’s direct memory limit in this case.
虽然 Task/Framework 的 Off-Heap 分区中可能含有 Native Non-Direct 内存，而这部分内存严格来说属于 JVM Overhead，不会被 JVM -XX:MaxDirectMemorySize 参数所限制，但 Flink 还是将它算入 MaxDirectMemorySize 中。这部分预留的 Direct 内存配额不会被实际使用，所以可以留给没有上限 JVM Overhead 占用，达到为 Native Non-Direct 内存预留空间的效果。
OOM Killed 常见原因
与上文分析一致，实践中导致 OOM Killed 的常见原因基本源于 Native 内存的泄漏或者过度使用。因为虚拟内存的 OOM Killed 通过资源管理器的配置很容易避免且通常不会有太大问题，所以下文只讨论物理内存的 OOM Killed。
RocksDB Native 内存的不确定性
众所周知，RocksDB 通过 JNI 直接申请 Native 内存，并不受 Flink 的管控，所以实际上 Flink 通过设置 RocksDB 的内存参数间接影响其内存使用。然而，目前 Flink 是通过估算得出这些参数，并不是非常精确的值，其中有以下的几个原因。
首先是部分内存难以准确计算的问题。RocksDB 的内存占用有 4 个部分[6]:
● Block Cache: OS PageCache 之上的一层缓存，缓存未压缩的数据 Block。
● Indexes and filter blocks: 索引及布隆过滤器，用于优化读性能。
● Memtable: 类似写缓存。
● Blocks pinned by Iterator: 触发 RocksDB 遍历操作（比如遍历 RocksDBMapState 的所有 key）时，Iterator 在其生命周期内会阻止其引用到的 Block 和 Memtable 被释放，导致额外的内存占用[10]。
前三个区域的内存都是可配置的，但 Iterator 锁定的资源则要取决于应用业务使用模式，且没有提供一个硬限制，因此 Flink 在计算 RocksDB StateBackend 内存时没有将这部分纳入考虑。
其次是 RocksDB Block Cache 的一个 bug[8][9]，它会导致 Cache 大小无法严格控制，有可能短时间内超出设置的内存容量，相当于软限制。
对于这个问题，通常我们只要调大 JVM Overhead 的阈值，让 Flink 预留更多内存即可，因为 RocksDB 的内存超额使用只是暂时的。
glibc Thread Arena 问题
另外一个常见的问题就是 glibc 著名的 64 MB 问题，它可能会导致 JVM 进程的内存使用大幅增长，最终被 YARN kill 掉。
具体来说，JVM 通过 glibc 申请内存，而为了提高内存分配效率和减少内存碎片，glibc 会维护称为 Arena 的内存池，包括一个共享的 Main Arena 和线程级别的 Thread Arena。当一个线程需要申请内存但 Main Arena 已经被其他线程加锁时，glibc 会分配一个大约 64 MB (64 位机器)的 Thread Arena 供线程使用。这些 Thread Arena 对于 JVM 是透明的，但会被算进进程的总体虚拟内存（VIRT）和物理内存（RSS）里。
默认情况下，Arena 的最大数目是 cpu 核数 * 8，对于一台普通的 32 核服务器来说最多占用 16 GB，不可谓不可观。为了控制总体消耗内存的总量，glibc 提供了环境变量 MALLOC_ARENA_MAX 来限制 Arena 的总量，比如 Hadoop 就默认将这个值设置为 4。然而，这个参数只是一个软限制，所有 Arena 都被加锁时，glibc 仍会新建 Thread Arena 来分配内存[11]，造成意外的内存使用。
通常来说，这个问题会出现在需要频繁创建线程的应用里，比如 HDFS Client 会为每个正在写入的文件新建一个 DataStreamer 线程，所以比较容易遇到 Thread Arena 的问题。如果怀疑你的 Flink 应用遇到这个问题，比较简单的验证方法就是看进程的 pmap 是否存在很多大小为 64MB 倍数的连续 anon 段，比如下图中蓝色几个的 65536 KB 的段就很有可能是 Arena。
img4. pmap 64 MB arena
这个问题的修复办法比较简单，将 MALLOC_ARENA_MAX 设置为 1 即可，也就是禁用 Thread Arena 只使用 Main Arena。当然，这样的代价就是线程分配内存效率会降低。不过值得一提的是，使用 Flink 的进程环境变量参数（比如 containerized.taskmanager.env.MALLOC_ARENA_MAX=1）来覆盖默认的 MALLOC_ARENA_MAX 参数可能是不可行的，原因是在非白名单变量（yarn.nodemanager.env-whitelist）冲突的情况下， NodeManager 会以合并 URL 的方式来合并原有的值和追加的值，最终造成 MALLOC_ARENA_MAX="4:1" 这样的结果。
最后，还有一个更彻底的可选解决方案，就是将 glibc 替换为 Google 家的 tcmalloc 或 Facebook 家的 jemalloc [12]。除了不会有 Thread Arena 问题，内存分配性能更好，碎片更少。在实际上，Flink 1.12 的官方镜像也将默认的内存分配器从 glibc 改为 jemelloc [17]。
JDK8 Native 内存泄漏
Oracle Jdk8u152 之前的版本存在一个 Native 内存泄漏的 bug[13]，会造成 JVM 的 Internal 内存分区一直增长。
具体而言，JVM 会缓存字符串符号（Symbol）到方法（Method）、成员变量（Field）的映射对来加快查找，每对映射称为 MemberName，整个映射关系称为 MemeberNameTable，由 java.lang.invoke.MethodHandles 这个类负责。在 Jdk8u152 之前，MemberNameTable 是使用 Native 内存的，因此一些过时的 MemberName 不会被 GC 自动清理，造成内存泄漏。
要确认这个问题，需要通过 NMT 来查看 JVM 内存情况，比如笔者就遇到过线上一个 TaskManager 的超过 400 MB 的 MemeberNameTable。
img5. JDK8 MemberNameTable Native 内存泄漏
在 JDK-8013267[14] 以后，MemeberNameTable 从 Native 内存被移到 Java Heap 当中，修复了这个问题。然而，JVM 的 Native 内存泄漏问题不止一个，比如 C2 编译器的内存泄漏问题[15]，所以对于跟笔者一样没有专门 JVM 团队的用户来说，升级到最新版本的 JDK 是修复问题的最好办法。
YARN mmap 内存算法
众所周知，YARN 会根据 /proc/${pid} 下的进程信息来计算整个 container 进程树的总体内存，但这里面有一个比较特殊的点是 mmap 的共享内存。mmap 内存会全部被算进进程的 VIRT，这点应该没有疑问，但关于 RSS 的计算则有不同标准。 依据 YARN 和 Linux smaps 的计算规则，内存页（Pages）按两种标准划分: ● Private Pages: 只有当前进程映射（mapped）的 Pages ● Shared Pages: 与其他进程共享的 Pages ● Clean Pages: 自从被映射后没有被修改过的 Pages ● Dirty Pages: 自从被映射后已经被修改过的 Pages 在默认的实现里，YARN 根据 /proc/${pid}/status 来计算总内存，所有的 Shared Pages 都会被算入进程的 RSS，即便这些 Pages 同时被多个进程映射[16]，这会导致和实际操作系统物理内存的偏差，有可能导致 Flink 进程被误杀（当然，前提是用户代码使用 mmap 且没有预留足够空间）。
为此，YARN 提供 yarn.nodemanager.container-monitor.procfs-tree.smaps-based-rss.enabled 配置选项，将其设置为 true 后，YARN 将根据更准确的 /proc/${pid}/smap 来计算内存占用，其中很关键的一个概念是 PSS。简单来说，PSS 的不同点在于计算内存时会将 Shared Pages 均分给所有使用这个 Pages 的进程，比如一个进程持有 1000 个 Private Pages 和 1000 个会分享给另外一个进程的 Shared Pages，那么该进程的总 Page 数就是 1500。 回到 YARN 的内存计算上，进程 RSS 等于其映射的所有 Pages RSS 的总和。在默认情况下，YARN 计算一个 Page RSS 公式为: ``` Page RSS = Private_Clean + Private_Dirty + Shared_Clean + Shared_Dirty ``` 因为一个 Page 要么是 Private，要么是 Shared，且要么是 Clean 要么是 Dirty，所以其实上述公示右边有至少三项为 0 。而在开启 smaps 选项后，公式变为: ``` Page RSS = Min(Shared_Dirty, PSS) + Private_Clean + Private_Dirty ``` 简单来说，新公式的结果就是去除了 Shared_Clean 部分被重复计算的影响。 虽然开启基于 smaps 计算的选项会让计算更加准确，但会引入遍历 Pages 计算内存总和的开销，不如 直接取 /proc/${pid}/status 的统计数据快，因此如果遇到 mmap 的问题，还是推荐通过提高 Flink 的 JVM Overhead 分区容量来解决。
总结
本文首先介绍 JVM 内存模型和 Flink TaskManager 内存模型，然后据此分析得出进程 OOM Killed 通常源于 Native 内存泄漏，最后列举几个常见的 Native 内存泄漏原因以及处理办法，包括 RocksDB 内存占用的不确定性、glibc 的 64MB 问题、JDK8 MemberNameTable 泄露和 YARN 对 mmap 内存计算的不准确。由于笔者水平有限，不能保证全部内容均正确无误，若读者有不同意见，非常欢迎留言指教一起探讨。
