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 class ID: 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

        

 

[rmetzger0]

Hi Forideal,

When using RocksDB, we need to serialize the data (to store it on disk), whereas when using the memory backend, the data (in this case RedConcat.ConcatString instances) is on the heap, thus we won't run into this issue.

Are you registering your custom types in the ExecutionConfig? (If so, it increases the chances of this error to happen)

Could you share the code of RedConcat.ConcatString as well?

I would not be surprised if this is a bug in Flink. Using a UDAF with custom types is probably not a very common use case.

Best,

Robert

On Fri, Aug 14, 2020 at 12:39 PM forideal <[hidden email]> wrote:

    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 class ID: 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

        

 

[forideal]

Hi Robert Metzger,

       I am very happy to share my code,

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

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

  

    > Are you registering your custom types in the ExecutionConfig? (If so, it increases the chances of this error to happen)

    Let me describe my scenario. We have built a SQL platform based on Flink, hoping to support user-defined UDF/UDAF, hoping that users only submit SQL and do not need to customize other codes. As for the serialization problem, it does exist.

    I currently work around this problem like this

    First ：this.env.getConfig().registerTypeWithKryoSerializer(ArrayList.class, org.apache.flink.api.java.typeutils.runtime.kryo.JavaSerializer.class);

    Second：ConcatString inherits Arraylist

   

    public class ConcatString extends ArrayList<String> {

    @Override
    public boolean add(String toString) {
        if (this.size() < 1000) {
            super.add(toString);
            return true;
        }
        return false;
    }

    public List<String> getList() {
        return this;
    }

}

Best forideal

At 2020-08-14 21:46:45, "Robert Metzger" <[hidden email]> wrote:

Hi Forideal,

When using RocksDB, we need to serialize the data (to store it on disk), whereas when using the memory backend, the data (in this case RedConcat.ConcatString instances) is on the heap, thus we won't run into this issue.

Are you registering your custom types in the ExecutionConfig? (If so, it increases the chances of this error to happen)

Could you share the code of RedConcat.ConcatString as well?

I would not be surprised if this is a bug in Flink. Using a UDAF with custom types is probably not a very common use case.

Best,

Robert

On Fri, Aug 14, 2020 at 12:39 PM forideal <[hidden email]> wrote:

    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 class ID: 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

        

 

 

[Timo Walther]

Hi Forideal,

luckily these problems will belong to the past in Flink 1.12 when UDAF
are updated to the new type system [1]. Lists will be natively supported
and registering custom KryoSerializers consistently as well.

Until then, another workaround is to override getAccumulatorType() and
define the PojoTypeInfo of ConcatString manually and e.g. replace the
GenericTypeInfo<List> with a proper
org.apache.flink.api.java.typeutils.ListTypeInfo.

I hope this helps.

Regards,
Timo

[1] https://issues.apache.org/jira/browse/FLINK-15803

On 15.08.20 08:07, forideal wrote:

> Hi Robert Metzger,
>
>         I am very happy to share my code,
>
>
>      public class ConcatString {
>          public List<String>list =new ArrayList<>();
>
> public void add(String toString) {
>              if (list !=null) {
>                  if (list.size() <100) {
>                      list.add(toString);
> }
>              }
>          }
>      }
>
>      > Are you registering your custom types in the ExecutionConfig? (If
> so, it increases the chances of this error to happen)
>      Let me describe my scenario. We have built a SQL platform based on
> Flink, hoping to support user-defined UDF/UDAF, hoping that users only
> submit SQL and do not need to customize other codes. As for the
> serialization problem, it does exist.
>
>      I currently work around this problem like this
>      First
> ：this.env.getConfig().registerTypeWithKryoSerializer(ArrayList.class,
> org.apache.flink.api.java.typeutils.runtime.kryo.JavaSerializer.class);
>      Second：ConcatString inherits Arraylist
>
>
> public class ConcatString extends ArrayList<String> {
>
> @Override
> public boolean add(String toString) {
>          if (this.size() <1000) {
>              super.add(toString);
> return true;
> }
>          return false;
> }
>
>      public List<String>getList() {
>          return this;
> }
>
> }
>
>
> Best forideal
>
>
>
> At 2020-08-14 21:46:45, "Robert Metzger" <[hidden email]> wrote:
>
>     Hi Forideal,
>
>     When using RocksDB, we need to serialize the data (to store it on
>     disk), whereas when using the memory backend, the data (in this
>     case RedConcat.ConcatString instances) is on the heap, thus we won't
>     run into this issue.
>
>     Are you registering your custom types in the ExecutionConfig? (If
>     so, it increases the chances of this error to happen)
>
>     Could you share the code of RedConcat.ConcatString as well?
>
>     I would not be surprised if this is a bug in Flink. Using a UDAF
>     with custom types is probably not a very common use case.
>
>     Best,
>     Robert
>
>
>
>     On Fri, Aug 14, 2020 at 12:39 PM forideal <[hidden email]
>     <mailto:[hidden email]>> wrote:
>
>              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 class ID: 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 RedConcatextends 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 ConcatStringcreateAccumulator() {
>                  return new ConcatString();
>         }
>
>              @Override
>         public void open(FunctionContext context)
>                      throws Exception {
>              }
>
>
>
>         Best forideal
>
>
>
>