并行度改变引发的血案

收到电话告警kafka消费堆积，就去查看任务健康状态，查找了以下几点：

1.查看背压情况，web页面绿码通行

2.查看任务GC情况，正常

3.查看数据写入情况，外部数据库mysql指标正常

4.查看checkpoint情况，几十毫秒完成

5.topic 生产消费速度，震惊~ 生产速度double了

至此可以确认消费能力不足导致，那就使用增加资源大法，调大任务并行度，看似一起都非常完美，

一顿操作调大并行度，重启任务，wath ? 任务咋报错了呢，查看错误日志，大概说的是写入mysql唯一键冲突，内心万马鹏腾，老老实实翻开代码看一下写入逻辑：做了一个窗口聚合的业务逻辑将窗口聚合的结果写入mysql，延时数据也不能丢弃，因此加了一个sideOutput 获取延时数据，为了减轻写入的压力，会将需要输出的数据缓存在状态中，做了定时定量输出，对于窗口结果数据是按照keyBy的，key与Mysql唯一键对应，那么对于全局来说key+windowTime 肯定是唯一的，对于延时数据来说拿到的是一条条的明细数据，没有经过窗口的处理的，因此在写入mysql之前会手动做窗口分配合并然后输出。看到这里，已经发现问题的根源，缓存数据状态使用的是operator-list 类型，改变任务并行度，会导致list数据被重新分配到不同的task中，对于延时的数据很有可能就会出现在不同的task出现属于同一个key+windowTime的数据，那么在写入的时候就会出现一个先写成功，一个后写就失败了。

解决方案：

1.首先并行度不做改变，在initializeState 方法中，将获取的状态数据直接刷写到mysql中

2.延时数据在写入到缓存时，做一次窗口分配、合并操作，保证延时缓存中的数据key+windowTime是唯一的

3.最后重新调整任务并行度

至此bug解决完成，做事还是不能太嚣张啊~

回顾一下任务并行度改变对状态产生的影响：

1.对于keyed state ， flink 在状态恢复的时候会按照key group自动完成状态的重新分配

2.对于operator state来说，根据使用的不同类型状态会分为以下几种情况：list state 会以轮序的方式重新分配，例如kafka offset; union state 所有的task都会有一份全量的状态信息；broadcast state 只会被广播一份数据