在单元测试中检查Flink算子状态的方法

在Flink的Test Harness中，你可以直接通过状态描述符（StateDescriptor）获取对应的状态实例，进而读取具体的状态值，以下是针对不同状态类型的实现方式：

1. 读取键控状态（Keyed State） #

针对KeyedOneInputStreamOperatorTestHarness这类键控算子测试工具，核心是用getKeyedState方法，传入目标键和对应的状态描述符：

示例：ValueState #

假设你的算子中定义了如下ValueState：

private ValueStateDescriptor<Integer> countStateDesc = new ValueStateDescriptor<>("count", Integer.class);
private ValueState<Integer> countState;

// 算子open方法中初始化
countState = getRuntimeContext().getState(countStateDesc);

在单元测试中读取状态值：

// 先触发算子处理数据，让状态完成更新
harness.processElement(new StreamRecord<>(yourInputData, 1000L));

// 获取指定key对应的状态实例
ValueState<Integer> targetState = harness.getKeyedState("target-key", countStateDesc);
// 读取状态值
Integer actualCount = targetState.value();

// 断言验证是否符合预期
assertEquals(expectedCount, actualCount);

示例：ListState #

如果是ListState，读取后可以转成集合进行验证：

ListState<String> targetListState = harness.getKeyedState("target-key", listStateDesc);
List<String> actualValues = StreamSupport.stream(targetListState.get().spliterator(), false)
                                         .collect(Collectors.toList());
assertEquals(expectedValueList, actualValues);

示例：MapState #

对于MapState，可以直接通过键读取对应值，或者遍历所有条目：

MapState<String, Integer> targetMapState = harness.getKeyedState("target-key", mapStateDesc);
// 读取单个键对应的值
Integer actualValue = targetMapState.get("map-entry-key");
assertEquals(expectedValue, actualValue);

// 遍历所有条目验证
Map<String, Integer> actualMap = new HashMap<>();
for (Map.Entry<String, Integer> entry : targetMapState.entries()) {
    actualMap.put(entry.getKey(), entry.getValue());
}
assertEquals(expectedMap, actualMap);

2. 读取算子状态（Operator State） #

如果是非键控的Operator State，使用getOperatorState方法获取状态实例，读取逻辑和键控状态一致：

ListState<String> operatorState = harness.getOperatorState(operatorStateDesc);
// 后续读取验证逻辑同ListState示例

注意事项 #

• 必须先触发算子的处理逻辑（比如processElement、processWatermark），确保状态已经完成更新后再读取
• 测试用的状态描述符必须和算子中定义的完全一致（名称、类型序列化器都要匹配）
• 如果状态配置了TTL，测试时需要通过harness.setProcessingTime()推进时间，避免状态被意外清理

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