升级Flink Scala版本至2.12时更新RocksDB状态触发NPE

问题根源 #

从堆栈日志可以明确，Kryo序列化Scala PriorityQueue时，其内部依赖的匿名Ordering类出现空指针。原因是Scala 2.11与2.12在Ordering的匿名实现（比如Ordering.Long的衍生类）的序列化逻辑上存在差异，升级到2.12后，原有Kryo序列化无法正确处理这些类的内部引用，导致序列化失败抛出NPE。

可行解决方案 #

1. 改用显式可序列化的Ordering实现 #

避免依赖Scala默认的隐式Ordering，自定义实现可序列化的Ordering类：

// 自定义可序列化的Long排序器（示例为降序，可根据业务调整）
class SerializableLongOrdering extends Ordering[Long] with Serializable {
  override def compare(a: Long, b: Long): Int = b.compareTo(a)
}

// 创建PriorityQueue时显式传入该排序器
val eventQueue = mutable.PriorityQueue.empty[Long](new SerializableLongOrdering())

此方案从根源上解决了匿名Ordering的序列化问题，是最直接的修复方式。

2. 替换为Flink原生状态结构 #

如果业务逻辑允许，放弃使用Scala的PriorityQueue，改用Flink提供的ListState或MapState：

• 从状态中取出所有元素后，在内存中完成排序逻辑
• 处理完成后将结果写回状态
  这种方式完全规避了Scala集合的序列化兼容性问题，更贴合Flink的状态管理设计。

3. 自定义Kryo序列化规则（复杂度较高） #

针对Scala的Ordering匿名类注册自定义Kryo序列化器，或者配置Kryo忽略导致空指针的字段：

// 在Flink环境配置中注册自定义序列化器
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().registerTypeWithKryoSerializer(scala.math.Ordering.class, new CustomOrderingSerializer());

此方案需要编写自定义Kryo序列化逻辑，仅在前两种方案无法实施时考虑。

验证流程 #

1. 替换PriorityQueue的Ordering实现为自定义可序列化类
2. 使用Scala 2.12重新编译打包任务
3. 启动Flink任务并跳过历史状态，观察是否仍出现NPE

内容的提问来源于stack exchange，提问作者ashwin gupta