Flink SQL UDAF com.esotericsoftware.kryo.KryoException: Encountered unregistered class ID

[forideal]

    Hi 
             I wrote a UDAF referring to this article  
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/functions/udfs.html#aggregation-functions,
 when using in-memory state, the task can run normally. However, When I chose 
rocksdb as the state backend, I encountered this error. Thank you for helping 
me see this problem.


The following is the error content:
com.esotericsoftware.kryo.KryoException: Encountered unregistered classID: 87
Serialization trace:
list (com.red.data.platform.RedConcat$ConcatString)
    at 
com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:119)
    at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
    at 
com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:99)
    at 
com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
    at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
    at 
org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:346)
    at org.apache.flink.util.InstantiationUtil.deserializeFromByt


public class RedConcat extends AggregateFunction<String, 
RedConcat.ConcatString> {

public class ConcatString {
public List<String> list = new ArrayList<>();

        public void add(String toString) {
if (list != null) {
if (list.size() < 100) {
list.add(toString);
}
            }
        }
    }

@Override
public boolean isDeterministic() {
return false;
}

@Override
public ConcatString createAccumulator() {
return new ConcatString();
}

@Override
public void open(FunctionContext context)
throws Exception {
    }




Best forideal