Kryo Serialization fails with NPE

[Prabhu Joseph <pr...@gmail.com>]

Hi, Need some pointers in fixing an issue with Kryo Serialization failing
with NPE. Based on the stack trace, kryo fails to find the registered class
for the given input type.

Kryo fails to find the class type
*FlinkA$FlinkB$$Lambda$2568/0x00007ee2fbec64b0*
This is how the actual code looks like.

*public class FlinkA { *
* public static class FlinkB extends Monitor<FlinkB.FlinkStates> { *
*   public static class Supplier implements
java.util.function.Supplier<FlinkB>, Serializable { *
*     public FlinkB get() { *
*      FlinkB ret = new FlinkB(); *
*      ret.ready(); *
*      return ret; *
*     } *
*   } *

* public enum FlinkStates { WaitForGetPutDeleteOperations }*

When I tried to register below, it did not work. Not sure how to register
for *FlinkA$FlinkB$$Lambda$2568/0x00007ee2fbec64b0 *which kryo is looking
for.

*env.getConfig().registerKryoType(FlinkA.FlinkB.class);*



Complete Stack Trace:

*Caused by: java.lang.NullPointerException at
com.esotericsoftware.kryo.util.DefaultClassResolver.writeClass(DefaultClassResolver.java:80)*
~[lib/:?] at com.esotericsoftware.kryo.Kryo.writeClass(Kryo.java:488)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:593)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.serializers.MapSerializer.write(MapSerializer.java:95)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.serializers.MapSerializer.write(MapSerializer.java:21)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.Kryo.writeObjectOrNull(Kryo.java:577)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:68)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
~[flink-dist-1.16.0.jar:1.16.0] at
com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:599)
~[flink-dist-1.16.0.jar:1.16.0] at
org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.serialize(KryoSerializer.java:364)
~[flink-dist-1.16.0.jar:1.16.0] at
org.apache.flink.runtime.state.heap.CopyOnWriteStateMapSnapshot.writeState(CopyOnWriteStateMapSnapshot.java:147)
~[flink-dist-1.16.0.jar:1.16.0] at
org.apache.flink.runtime.state.heap.AbstractStateTableSnapshot.writeStateInKeyGroup(AbstractStateTableSnapshot.java:116)
~[flink-dist-1.16.0.jar:1.16.0] at
org.apache.flink.runtime.state.heap.CopyOnWriteStateTableSnapshot.writeStateInKeyGroup(CopyOnWriteStateTableSnapshot.java:38)
~[flink-dist-1.16.0.jar:1.16.0] at
org.apache.flink.runtime.state.heap.HeapSnapshotStrategy.lambda$asyncSnapshot$3(HeapSnapshotStrategy.java:172)
~[flink-dist-1.16.0.jar:1.16.0] at
org.apache.flink.runtime.state.SnapshotStrategyRunner$1.callInternal(SnapshotStrategyRunner.java:91)
~[flink-dist-1.16.0.jar:1.16.0] at
org.apache.flink.runtime.state.SnapshotStrategyRunner$1.callInternal(SnapshotStrategyRunner.java:88)
~[flink-dist-1.16.0.jar:1.16.0] at
org.apache.flink.runtime.state.AsyncSnapshotCallable.call(AsyncSnapshotCallable.java:78)
~[flink-dist-1.16.0.jar:1.16.0] at
java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]