Flink中MapValue使用报错求助：无法实例化自定义HashMapValue类

问题根源 #

你遇到的AssertionError是Java泛型类型擦除机制导致的：自定义的HashMapValue是泛型类，运行时JVM会擦除泛型参数占位符<K,V>，Flink的ReflectionUtil在反射获取MapValue父类的具体泛型类型时，只能拿到抽象的K和V，而非你实例化时指定的IntValue、BooleanValue，触发断言校验失败。

解决方案一：实现非泛型的MapValue子类 #

Flink的MapValue设计要求子类必须明确具体的泛型参数类型，因此需要针对你要存储的键值类型，实现非泛型的子类：

public class IntBooleanHashMapValue extends MapValue<IntValue, BooleanValue> {
    public IntBooleanHashMapValue() {
        super();
    }
}

实例化时直接使用：

IntBooleanHashMapValue hashMapValue = new IntBooleanHashMapValue();

这种方式下，Flink的反射工具可以明确获取到父类MapValue的具体泛型参数，正常完成序列化/反序列化逻辑。

解决方案二：使用Flink原生MapState（更推荐） #

如果你的需求是在状态中存储键值对，更推荐直接使用Flink状态API提供的MapState，无需手动实现MapValue子类，Flink会自动处理状态的序列化：

// 在KeyedStream的Operator/ProcessFunction中使用
MapStateDescriptor<Integer, Boolean> mapStateDesc = 
    new MapStateDescriptor<>(
        "user-state",  // 状态名称
        Integer.class, // 键类型
        Boolean.class  // 值类型
    );
MapState<Integer, Boolean> mapState = getRuntimeContext().getMapState(mapStateDesc);

// 操作MapState
mapState.put(1, true);
Boolean value = mapState.get(1);

这种方式不仅简化了代码，还能利用Flink状态管理的快照、容错等原生特性。

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