Flink中CoProcessFunction初始化ListState报Keyed state仅可用于键控流空指针错误

你遇到的空指针错误，本质是触发了Flink键控状态的上下文强制校验逻辑：

• 键控状态（包括ListState、ValueState等）只能在KeyedStream的上下文中使用，运行时会校验当前算子的上下文是否关联了KeyedStateStore实例。
• 你当前的流链路只对dataStream1执行了keyBy操作，与之connect的dataStream2是没有经过keyBy的非键控流，此时连接生成的流不属于KeyedStream范畴，RuntimeContext中没有可访问的键控状态存储实例，因此在open方法中调用getListState时，直接抛出对应空指针异常。

满足键控状态使用前提 #

Flink中两个流连接后要使用键控状态，必须在connect前对两条流分别按照完全相同的key规则、相同的key类型执行keyBy操作，此时生成的连接流才属于KeyedConnectedStream，可正常读写键控状态。
修正后的代码逻辑如下：

// 1. 对dataStream1按f1字段（Long类型）做keyBy
KeyedStream<Tuple2<Long, String>, Long> keyedStream1 = dataStream1.keyBy(ele -> ele.f1);
// 2. 对dataStream2按同类型的Long key做keyBy，保证key类型与第一条流完全匹配
KeyedStream<String, Long> keyedStream2 = dataStream2.keyBy(content -> extractLongKey(content));

// 3. 两个键控流连接后，即可在CoProcessFunction中正常初始化ListState
keyedStream1.connect(keyedStream2)
    .process(new CoProcessFunction<Tuple2<Long, String>, String, Object>() {
        private ListState<Tuple2<Long, String>> stream1ListState;

        @Override
        public void open(Configuration parameters) throws Exception {
            ListStateDescriptor<Tuple2<Long, String>> listStateDescriptor = new ListStateDescriptor<>(
                "type1",
                TypeInformation.of(new TypeHint<Tuple2<Long, String>>() {})
            );
            // 此时获取键控状态不会抛出异常
            stream1ListState = getRuntimeContext().getListState(listStateDescriptor);
        }

        @Override
        public void processElement1(Tuple2<Long, String> longStringTuple2, Context context, Collector<Object> collector) throws Exception {
            collector.collect(longStringTuple2);
        }

        @Override
        public void processElement2(String s, Context context, Collector<Object> collector) throws Exception {
            collector.collect(s);
        }
    })
    .print();

特殊场景适配 #

如果业务上不需要按key关联两条流，确实无法对第二条流执行keyBy操作，就不能在该场景下使用键控状态。此时需要改用算子级别的托管状态（Managed Operator State），通过实现CheckpointedFunction接口定义状态读写逻辑，适配非键控连接流的上下文。

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