背压
数据涌入的速度大于处理速度。在source operator中，可通过Kafka解决。在任务间的operator有如下机制应对：
Local exchange：task1和2在同一个工作节点，那么buffer pool可以直接交给下一个任务，但下一个任务task2消费buffer pool中的信息速度减慢时，当前任务task1填充buffer pool的速度也会减慢。
Remote exchange：TM保证每个task至少有一个incoming和一个outgoing缓冲区。当下游receiver的处理速度低于上有的sender的发送速度，receiver的incoming缓冲区就会开始积累数据（需要空闲的buffer来放从TCP连接中接收的数据），当挤满后就不再接收数据。上游sender利用netty水位机制，当网络中的缓冲数据过多时暂停发送。
Data Transfer in Flink
TM负责数据在tasks间的转移，转移之前会存储到buffer（这又变回micro-batches）。每个TM有32KB的网络buffer用于接收和发送数据。如果sender和receiver在不同进程，那么会通过操作系统的网络栈来通信。每对TM保持permanent TCP连接来交换数据。每个sender任务能够给所有receiving任务发送数据，反之，所有receiver任务能够接收所有sender任务的数据。TM保证每个任务都至少有一个incoming和outgoing的buffer，并增加额外的缓冲区分配约束来避免死锁。
如果sender和receiver任务在同一个TM进程，sender会序列化结果数据到buffer，如果满了就放到队列。receiver任务通过队列得到数据并进行反序列化。这样的好处是解耦任务并允许在任务中使用可变对象，从而减少了对象实例化和垃圾收集。一旦数据被序列化，就能安全地修改。而缺点是计算消耗大，在一些条件下能够把task穿起来，避免序列化。(C10)
Flow Control with Back Pressure
receiver放到缓冲区的数据变为队列，sender将要发送的数据变为队列，最后sender减慢发送速度。
Event Time Processing
event time处理的数据必须有时间戳（Long unix timestamp）并定义了watermarks。watermark是一种特殊的records holding a timestamp long value。它必须是递增的（防止倒退），有一个timestamp t（下图的5），暗示所有接下来的数据都会大于这个值。后来的，小于这个值，就被视为迟来数据，Flink有其他机制处理。
Watermarks and Event Time
WM在Flink是一种特殊的record，它会被operator tasks接收和释放。
tasks有时间服务来维持timers（timers注册到时间服务上），在time-window task中，timers分别记录了各个window的结束时间。当任务获得一个watermark时，task会根据这个watermark的timestamp更新内部的event-time clock。任务内部的时间服务确定所有timers时间是否小于watermark的timestamp，如果大于则触发call-back算子来释放记录并返回结果。最后task还会将更新的event-time clock的WM进行广播。（结合下图理解）
只有ProcessFunction可以读取和修改timestamp或者watermark（The ProcessFunction can read the timestamp of a currently processed record, request the current event-time of the operator, and register timers）。下面是PF的行为。
当收到WM大于所有目前拥有的WM，就会把event-time clock更新为所有WM中最小的那个，并广播这个最小的WM。即便是多个streams输入，机制也一样，只是增加Paritition WM数量。这种机制要求获得的WM必须是累加的，而且task必须有新的WM接收，否则clock就不会更新，task的timers就不会被触发。另外，当多个streams输入时，timers会被WM比较离散的stream主导，从而使更密集的stream的state不断积累。
Timestamp Assignment and Watermark Generation
当streaming application消化流时产生。Flink有三种方式产生：
上面两种state的存在方式有两种：raw和managed，一般都是用后者，也推荐用后者（更好的内存管理、不需造轮子）。
State Backends
state backend决定了state如何被存储、访问和维持。它的主要职责是本地state管理和checkpoint state到远程。在管理方面，可选择将state存储到内存还是磁盘。checkpoint方面在C8详细介绍。
MemoryStateBackend, FsStateBackend, RocksDBStateBackend适合越来越大的state。都支持异步checkpoint，其中RocksDB还支持incremental的checkpoint。
注意：As RocksDB’s JNI bridge API is based on byte[], the maximum supported size per key and per value is 2^31 bytes each. IMPORTANT: states that use merge operations in RocksDB (e.g. ListState) can silently accumulate value sizes > 2^31 bytes and will then fail on their next retrieval. This is currently a limitation of RocksDB JNI.
当operator的state很大时，复制整个state并发送到远程storage会很费时。而RocksDB state backend支持asynchronous and incremental的checkpoints。当触发checkpoint时，backend会快照所有本地state的修改（直至上一次checkpoint），然后马上让task继续执行。后台线程异步发送快照到远程storage。
Recovery from Consistent Checkpoints
上图队列中的7和6之所以能恢复，取决于数据源是否resettable，如Kafka，不会因为发送信息就把信息删除。这才能实现处理过程的exactly-once state consistency（严格来讲，数据还是被重复处理，但是在读档后重复的）。但是下游系统有可能接收到多个结果。这方面，Flink提供sink算子实现output的exactly-once，例如给checkpoint提交records释放记录。另一个方法是idempotent updates，详细看C7。
Savepoints
checkpoints加上一些额外的元数据，功能也是在checkpoint的基础上丰富。不同于checkpoints，savepoint不会被Flink自动创造（由用户或者外部scheduler触发创造）和销毁。savepoint可以重启不同但兼容的作业，从而：
修复bugs进而修复错误的结果，也可用于A/B test或者what-if场景。
调整并发度
迁移作业到其他集群、新版Flink
也可以用于暂停作业，通过savepoint查看作业情况。
参考 Stream Processing with Apache Flink by Vasiliki Kalavri; Fabian Hueske
