Hello, I am using Flink 1.7.2. I wrote a small application which uses KeyedProcessFunction to maintain application state. The state value object is using Guava's (version 18.0.50) ImmutableMap to create a copy of the map instance as part of its constructor.
I am using "de.javakaffee:kryo-serializers:0.45" to take handle the guava ImmutableMap serialization and configured the serializer using ` env.getConfig().registerTypeWithKryoSerializer(ImmutableMap.class, ImmutableMapSerializer.class)`. I am still seeing an issue with ImmutableMap serialization and I am not sure if it is using the serializer that I have configured? (may be I am not configuring it correctly??). Using standard java Map works properly. Appreciate any inputs? I have provided the error stack trace and the code snippet below. Regards Vijay Error StackTrace ``` Serialization trace: itemMap (com.flink.SensorState) start (com.flink.ValueObject) at com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:133) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:122) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:712) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:114) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:122) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:793) ~[kryo-5.0.0-RC1.jar:na] at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:315) ~[flink-core-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.heap.StateTableByKeyGroupReaders.lambda$createV2PlusReader$0(StateTableByKeyGroupReaders.java:74) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.KeyGroupPartitioner$PartitioningResultKeyGroupReader.readMappingsInKeyGroup(KeyGroupPartitioner.java:297) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.readKeyGroupStateData(HeapKeyedStateBackend.java:492) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.readStateHandleStateData(HeapKeyedStateBackend.java:453) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restorePartitionedState(HeapKeyedStateBackend.java:410) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restore(HeapKeyedStateBackend.java:358) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restore(HeapKeyedStateBackend.java:104) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:151) ~[flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:123) ~[flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:284) [flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:135) [flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:250) [flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:738) [flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:289) [flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.taskmanager.Task.run(Task.java:704) [flink-runtime_2.12-1.7.2.jar:1.7.2] at java.lang.Thread.run(Thread.java:748) [na:1.8.0_191] Caused by: java.lang.UnsupportedOperationException: null at org.apache.flink.shaded.guava18.com.google.common.collect.ImmutableMap.put(ImmutableMap.java:326) ~[flink-shaded-guava-18.0-5.0.jar:18.0-5.0] at com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:230) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:42) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:712) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:114) ~[kryo-5.0.0-RC1.jar:na] ... 22 common frames omitted ``` ``` final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().enableCheckpointing(CHECKPOINT_INTERVAL); env.setParallelism(PARALLELISM); env.getConfig().registerTypeWithKryoSerializer(ImmutableMap.class, ImmutableMapSerializer.class); int totalRecords = 1000; env.addSource(new StateDataProducer(totalRecords)) .keyBy(0) .process((KeyedProcessFunction) new VOBookmarker()) .print(); try { env.execute(); } catch (Exception e) { e.printStackTrace(); } ``` ``` class SensorState implements Serializable { private final int sensorId; private final Map<String,String> itemMap; public SensorState(int sensorId, Map<String,String> itemMap) { this.sensorId = sensorId; this.itemMap = ImmutableMap.copyOf(itemMap); //this.itemMap = itemMap; } public int getSensorId() { return sensorId; } public Map<String, String> getItemMap() { return Collections.unmodifiableMap(itemMap); } } public class ValueObject implements Serializable { private final int sensorId; private SensorState start; private SensorState end; public ValueObject(int sensorId) { this.sensorId = sensorId; } public int getSensorId() { return sensorId; } public void setStartState(SensorState start) { this.start = start; } public void setEndState(SensorState end) { this.end = end; } } ``` -- Sent from: http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/