flink版本：1.14.3

1. savepoint保存逐步增大，单个任务一次savepoint10G以上。
2. taskmanager日志打印ttl到期
   INFO org.apache.flink.table.runtime.operators.sink.SinkUpsertMaterializer [] - The state is cleared because of state ttl. This will result in incorrect result. You can increase the state ttl to avoid this.
3. taskmanager 在任务启动后内存逐步增高，RES逐步逼近最大内存限制，即使停掉所有任务后，内存依然无法释放。

排查过程及解决方案

state.backend=rocksdb
state.backend.incremental: true
1.2.3 调大 checkpointInterval间隔：900000(15min)，调大的主要目的减少快照的资源消耗。
 
2.1排查过程：这个报错非常具有迷惑性，一开始看到报错后便调大了set table.exec.state.ttl的值
 这个参数，但是依然报错，后来把值设为0（这意味着它永远不会清除状态），这个报错依然持续。
 查阅flink1.14.3源码发现，该报错是动态表在更新时，回撤流无法找到对应主键导致的，这个也是flink非确定性更新（NDU)问题，这个问题也会导致状态膨胀；排查发现flink sql中只在sink端指定了主键，未在source端指定主键。
 2.2解决方案：source端和sink端都需要指定主键 PRIMARY KEY (id) NOT ENFORCED
 
3.1 排查过程：通过arthas的dashboard命令查看taskmanager jvm情况，发现jvm远小于taskmanager 进程的RES，且g1_eden_space占heap98%以上，而g1_old_gen只占1%左右，初步排除jvm无法回收老年代的问题；基本就是state backbend的问题，state backend 的内存回收不受flink控制。
 3.2解决方案：
 3.2.1 在flink-conf.yaml中调整RocksDB相关参数
 
#指定rocksdb日志输出路径
state.backend.rocksdb.log.dir: /tmp/rocksdb_log
#RocksDB文件的最大大小
state.backend.rocksdb.log.max-file-size: 25mb
#RocksDB应保留用于信息记录的最大文件数
state.backend.rocksdb.log.file-num: 4
#RocksDB日志级别
state.backend.rocksdb.log.level: INFO_LEVEL
#文件块大小
state.backend.rocksdb.block.blocksize: 4kb
#RocksDB中数据块的缓存量
state.backend.rocksdb.block.cache-size: 128mb
#监控驻留在块缓存中的条目的内存大小
state.backend.rocksdb.metrics.block-cache-usage: true
RocksDB的参数调整可以根据实际情况来，上述调整会导致RocksDB在内存中数据量相对减小，但是也会增大io和磁盘刷写频率，这个看取舍。
 
taskmanager oom的原因很多，有可能是代码的问题，有可能是流量激增的问题，也有可能是任务状态的问题，排查时需要仔细判断；分享一些大佬的文章关于参数调优和错误排查的文章。
 ps:flink官方文档中的一些默认参数配置也在不断优化，比如1.14.4的state.backend.rocksdb.log.file-num默认参数和1.16.0的默认参数不一致，这个也是社区在不断调优的表现，所以最好按新版的参数来调整默认参数。
很久没写过源码走读类型的文章了。最近在做业务需求时用Flink的State TTL非常多，今天就来探索一下吧。
从Flink 1.6版本开始，社区为状态引入了TTL（time-to-live，生存时间）机制，支持Keyed State的自动过期，有效解决了状态数据在无干预情况下无限增长导致OOM的问题。State TTL的用法很简单，官方文档中给出的示例代码如下。
StateTtlConfi...
'connector' = 'phoenix-jdbc', 
'driver'='org.apache.phoenix.jdbc.PhoenixDriver',
'org.apache.flink.connector.phoenix.schema.isNamespaceMappingEnabled'='true',
'org.apache.flink.connector.phoenix.schema.mapSystemTablesToNamespace'='true',
'url' = 'jdbc:phoenix:192.168.71.00:2181/hbase', 
'table-name' = 'ODS.TEST1'
有问题随时给我私信~
2、java.util.concurrent.TimeoutException: Heartbeat of TaskManager with id 【containerID】 timed out.
这时我们可以观察TaskMana
				
增加有关checkpoint的优化参数
SET execution.checkpointing.timeout = 30min; --checkpoint超时时长
SET execution.checkpointing.min-pause = 1200s; --checkpoint间隔时间
SET execution.checkpointing.externalized-checkpoint-retention = DELETE_ON_CANCELLATION; --作业失败删除checkpoint
				
flink反压现象模拟
flink网络栈
TaskManager 传输数据时，不同的 TaskManager 上的两个 Subtask 间通常根据 key 的数量有多个 Channel，这些 Channel 会复用同一个 TaskManager 级别的 TCP 链接，并且共享接收端 Subtask 级别的 Buffer Pool。
在接收端，每个 Channel 在初始阶段会被分配固定数量的 Exclusive Buffer，这些 Buffer 会被用于存储接受到的数据，交给 Operator 使用后再次被释放。Channel 接收端空闲的 Buffer 数量称为 Credit，Credit 
				
Flink的RocksDBStateBackend一些使用经验
一、idea依赖
RocksDBStateBackend是Flink中内置的第三方状态管理器，和其他两种（MemoryStateBackend和FsStateBackend）不同，使用时需添加相关依赖：
  <dependency>
         <groupId>org.apache.flink</groupId>
         <artifactId>flink-statebacken
                    CSDN-Ada助手: 
                    恭喜您写了第三篇博客，标题看起来非常有用！检查错误并提供解决方案对于正在学习 Flink SQL 的人来说绝对是非常有价值的。接下来，我建议您可以尝试撰写一些关于 Flink SQL 实际应用的案例，这样读者可以更好地了解如何在实际工作中应用所学到的知识。再次感谢您的分享！
CSDN 正在通过评论红包奖励优秀博客，请看红包流：https://bbs.csdn.net/?type=4&header=0&utm_source=csdn_ai_ada_blog_reply3，我们会奖励持续创作和学习的博主，请看：https://bbs.csdn.net/forums/csdnnews?typeId=116148&utm_source=csdn_ai_ada_blog_reply3