Flink CEP中SharedBufferAccessor触发NullPointerException问题咨询

大家好，我最近在维护一个Flink CEP作业时遇到了一个棘手的随机异常问题，想请教下社区的各位同仁：

作业与异常基本情况 #

我们的Flink作业运行在Kubernetes集群中，使用CEP匹配的模式如下：

Pattern
  .<CustomEvent>begin(PATTERN_NAME)
  .times(threshold)
  .within(Time.milliseconds(timeWindowMilliseconds));

这个作业之前长期稳定运行，但最近一次部署后开始随机抛出CEP内部的NullPointerException，异常栈如下：

Caused by: org.apache.flink.util.WrappingRuntimeException: java.lang.NullPointerException
at org.apache.flink.cep.nfa.sharedbuffer.SharedBufferAccessor.materializeMatch(SharedBufferAccessor.java:216)
at org.apache.flink.cep.nfa.NFA.doProcess(NFA.java:423)
at org.apache.flink.cep.nfa.NFA.process(NFA.java:251)
at org.apache.flink.cep.operator.CepOperator.processEvent(CepOperator.java:404)
at org.apache.flink.cep.operator.CepOperator.lambda$onEventTime$0(CepOperator.java:315)

定位到的异常触发代码是CEP共享缓冲区的访问逻辑：

V event = sharedBuffer.getEvent(eventId).getElement();

很明显是sharedBuffer.getEvent(eventId)返回了null，进而调用getElement()时触发了NPE。

作业环境与变更背景 #

• Flink版本：1.16.0
• 数据链路：基于Keyed数据流，数据源为Confluent Kafka，采用Avro序列化，使用事件时间处理
• 状态管理：RocksDB作为Checkpoint后端，作业启动时的初始状态恢复流程看起来正常
• 最近的部署变更：
  • 作业并行度从1调整为2
  • CustomEvent的Confluent Avro Schema新增了一个可选字段，默认值为null
    • 补充：我们的CEP窗口设置为1天（timeWindowMilliseconds=86400000），所以当前作业状态中很可能还留存着旧Schema版本的事件

我的初步推测 #

因为异常是随机出现的，我怀疑是当某个Key的数据流中，状态里的旧Schema事件与新事件交互时触发了这个问题。但按理解，Avro新增可选字段属于向后兼容的变更，理论上不应该导致CEP底层共享缓冲区的事件丢失或无法读取？

有没有同学碰到过类似的问题？或者能给我一些排查方向或根因分析的建议？

备注：内容来源于stack exchange，提问作者Noah