I have a question about using MapValue in Flink because I need to save a map as part of the state, as you know the state needs to be deserializable/serializable, so i extend my class from MapValue because MapValue is an abstract class.

public class HashMapValue<K extends Value, V extends Value> extends MapValue<K, V> {
    public HashMapValue() {
        super();
    }
}

When i try to use this class, i can’t initiate the class

HashMapValue<IntValue, BooleanValue> hashMapValue = new HashMapValue<IntValue, BooleanValue>();

the error message is as follows:

java.lang.AssertionError
    at org.apache.flink.util.ReflectionUtil.getTemplateTypes(ReflectionUtil.java:141)
    at org.apache.flink.util.ReflectionUtil.getSuperTemplateTypes(ReflectionUtil.java:98)
    at org.apache.flink.util.ReflectionUtil.getTemplateType(ReflectionUtil.java:44)
    at org.apache.flink.util.ReflectionUtil.getTemplateType1(ReflectionUtil.java:54)
    at org.apache.flink.types.MapValue.<init>(MapValue.java:55)

I step into the code and apparently in getTemplateTypes function, it gets K as the parameter instead of IntValue, which I clearly passed in as generic type.
I can’t find online or in the flink source code anything related to MapValue as example. Can anyone help me by pointing out how to use MapValue in Flink?