You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Prasanna kumar <pr...@gmail.com> on 2021/11/01 10:34:51 UTC
Re: Kryo Serialization issues in Flink Jobs.
Any thoughts on these ?
Thanks,
Prasanna.
On Sat, Oct 30, 2021 at 7:25 PM Prasanna kumar <
prasannakumarramani@gmail.com> wrote:
> Hi ,
>
> We have the following Flink Job that processes records from kafka based on
> the rules we get from S3 files into broadcasted state.
> Earlier we were able to spin a job with any number of task parallelism
> without any issues.
> Recently we made changes to the Broadcast state Structure and it is
> working well upto parallelism of 3.
> If we give parallelism of 4 or more , we end up getting
> serialization exceptions which result in job failure. ( Block 4 as in the
> Image)
> Also If the leader job manager dies and a new one comes up , the other
> jobs are restarted automatically but this job dies of serialization issues.
> But when we start manually with a parallelism <= 3, it is working.
>
> Programmatically this code is working when we tested with all possible
> test cases.
> How do we debug serialization issues that we face.
> I have attached the exception of logs and the code related to it.
>
> Let me know if any more details are required.
>
>
>
>
>
> *KRYO SERIALIZAER INITIALISATON*
>
> Class<?> unmodifiableCollectionsSerializer =
> Class.forName("java.util.Collections$UnmodifiableCollection");
> env.getConfig().addDefaultKryoSerializer(
> unmodifiableCollectionsSerializer,
> UnmodifiableCollectionsSerializer.class
> );
>
>
> *CONFIGSTATE INTERFACE(USED IN BROADCAST STATE)*
>
> public interface EventConfigState {
>
>
> void createOrUpdateState(String key, DataPair dataPair);
>
> List<OutputMessage> executeRule(InputMessage inputMessage);
>
> Map<String, Set<DataPair>> getCurrentState();
> }
>
>
> *DERIVED EVENT CONFIG STATE IMPLEMENTATION*
>
> public class DerivedEventConfigState implements EventConfigState {
>
> Logger logger = LoggerFactory.getLogger(DerivedEventConfigState.class);
> private Map<String, Set<DataPair>> derivedConfigMap;
>
> public DerivedEventConfigState() {
> derivedConfigMap = new HashMap<>();
> }
>
> public void createOrUpdateState(String key, DataPair dataPair) {
>
> derivedConfigMap.putIfAbsent(key, new HashSet<>());
> if (derivedConfigMap.get(key).contains(dataPair)) {
> derivedConfigMap.get(key).remove(dataPair);
> }
> derivedConfigMap.get(key).add(dataPair);
> }
>
> @Override
> public List<OutputMessage> executeRule(InputMessage inputMessage) {
>
> String key = inputMessage.getKey();
> List<OutputMessage> outputMessageList = new ArrayList<>();
>
> if (derivedConfigMap.size() == 0) {
> logger.error("DerivedEventConfigMap is empty");
> return outputMessageList;
> }
>
> if ( derivedConfigMap.get(key) == null) {
> return outputMessageList;
> }
>
> for (DataPair dataPair : derivedConfigMap.get(key)) {
> IRule rule = dataPair.getRule();
> if (rule.isSatisfied(inputMessage)) {
>
> IMessageBuilder messageBuilder =
> MessageBuilderFactory.getMessageBuilder("OutputMessage");
> OutputMessage outputMessage = messageBuilder.build(
> inputMessage,
> dataPair.getEventMessageDefinition()
> );
> outputMessageList.add(outputMessage);
> }
> }
> return outputMessageList;
> }
>
> @Override
> public Map<String, Set<DataPair>> getCurrentState() {
> return Collections.unmodifiableMap(derivedConfigMap);
> }
>
> @Override
> public String toString() {
> return "DerivedEventConfigState{"
> + "derivedConfigMap=" + derivedConfigMap
> + '}';
> }
> }
>
>
> Attached are three Exceptions thrown rando
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
> *Caused by: org.apache.flink.util.SerializedThrowable: TABLE-OP at
> java.net.URLClassLoader.findClass(Unknown Source) ~[?:?] at
> java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?] at
> org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:64)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:74)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:48)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?] at
> org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:172)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at java.lang.Class.forName0(Native
> Method) ~[?:?] at java.lang.Class.forName(Unknown Source) ~[?:?] at
> com.esotericsoftware.kryo.util.DefaultClassResolver.readName(DefaultClassResolver.java:136)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:115)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:752)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:116)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:22)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:143)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:21)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:657)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:273)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.api.common.typeutils.base.MapSerializer.copy(MapSerializer.java:111)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.<init>(HeapBroadcastState.java:69)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.deepCopy(HeapBroadcastState.java:84)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.deepCopy(HeapBroadcastState.java:40)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.DefaultOperatorStateBackendSnapshotStrategy.snapshot(DefaultOperatorStateBackendSnapshotStrategy.java:100)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.DefaultOperatorStateBackend.snapshot(DefaultOperatorStateBackend.java:234)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:213)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:162)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:371)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointStreamOperator(SubtaskCheckpointCoordinatorImpl.java:686)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.buildOperatorSnapshotFutures(SubtaskCheckpointCoordinatorImpl.java:607)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.takeSnapshotSync(SubtaskCheckpointCoordinatorImpl.java:572)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:298)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$9(StreamTask.java:1004)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:988)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:947)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] ... 14 more*
>
>
> *Caused by: org.apache.flink.util.SerializedThrowable: Encountered
> unregistered class ID: 97 Serialization trace: derivedConfigMap
> (com.org.app.model.producer.DerivedEventConfigState) at
> com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:119)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:752)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:135)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:21)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:143)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:21)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:657)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:273)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.api.common.typeutils.base.MapSerializer.copy(MapSerializer.java:111)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.<init>(HeapBroadcastState.java:69)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.deepCopy(HeapBroadcastState.java:84)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.deepCopy(HeapBroadcastState.java:40)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.DefaultOperatorStateBackendSnapshotStrategy.snapshot(DefaultOperatorStateBackendSnapshotStrategy.java:100)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.DefaultOperatorStateBackend.snapshot(DefaultOperatorStateBackend.java:234)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:213)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:162)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:371)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointStreamOperator(SubtaskCheckpointCoordinatorImpl.java:686)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.buildOperatorSnapshotFutures(SubtaskCheckpointCoordinatorImpl.java:607)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.takeSnapshotSync(SubtaskCheckpointCoordinatorImpl.java:572)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:298)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$9(StreamTask.java:1004)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:988)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:947)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] ... 14 more*
>
> *Caused by: org.apache.flink.util.SerializedThrowable: Buffer underflow.
> Serialization trace: logger
> (com.org.app.model.producer.AppEventConfigState) at
> com.esotericsoftware.kryo.io.Input.require(Input.java:181)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.io.Input.readVarInt(Input.java:355)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:109)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:99)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:657)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:273)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.api.common.typeutils.base.MapSerializer.copy(MapSerializer.java:111)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.<init>(HeapBroadcastState.java:69)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.deepCopy(HeapBroadcastState.java:84)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.deepCopy(HeapBroadcastState.java:40)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.DefaultOperatorStateBackendSnapshotStrategy.snapshot(DefaultOperatorStateBackendSnapshotStrategy.java:100)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.DefaultOperatorStateBackend.snapshot(DefaultOperatorStateBackend.java:234)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:213)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] ... 24 more*
>
> Thanks
> Prasanna,
>
Re: Kryo Serialization issues in Flink Jobs.
Posted by Timo Walther <tw...@apache.org>.
Hi Prasanna,
it could be a bug where the ExecutionConfig is not forwarded properly to
all locations where the KryoSerializer is used.
As a first step for debugging, I would recommend to create a custom
TypeInformation (most methods are not relevant except for
createTypeSerializer and getTypeClass) and instantiate the
KryoSerializer manually. You can then pass this type information to all
operators (using dataStream.returns() or for state). This ensures that
all locations use the same KryoSerializer configuration and registered
ID get lost in the stack.
You can also check locally (either in your IDE or with a remote
debugger) that all KryoSerializer instances have the same registry
entries for class ID: 97.
Where is UnmodifiableCollectionsSerializer coming from?
Regards,
Timo
On 01.11.21 11:34, Prasanna kumar wrote:
> Any thoughts on these ?
>
> Thanks,
> Prasanna.
>
> On Sat, Oct 30, 2021 at 7:25 PM Prasanna kumar
> <prasannakumarramani@gmail.com <ma...@gmail.com>>
> wrote:
>
> Hi ,
>
> We have the following Flink Job that processes records from kafka
> based on the rules we get from S3 files into broadcasted state.
> Earlier we were able to spin a job with any number of task
> parallelism without any issues.
> Recently we made changes to the Broadcast state Structure and it is
> working well upto parallelism of 3.
> If we give parallelism of 4 or more , we end up getting
> serialization exceptions which result in job failure. ( Block 4 as
> in the Image)
> Also If the leader job manager dies and a new one comes up , the
> other jobs are restarted automatically but this job dies of
> serialization issues.
> But when we start manually with a parallelism <= 3, it is working.
>
> Programmatically this code is working when we tested with all
> possible test cases.
> How do we debug serialization issues that we face.
> I have attached the exception of logs and the code related to it.
>
> Let me know if any more details are required.
>
>
>
>
>
> *KRYO SERIALIZAER INITIALISATON*
>
> Class<?> unmodifiableCollectionsSerializer =
> Class.forName("java.util.Collections$UnmodifiableCollection");
> env.getConfig().addDefaultKryoSerializer(
> unmodifiableCollectionsSerializer,
> UnmodifiableCollectionsSerializer.class
> );
>
>
> *CONFIGSTATE INTERFACE(USED IN BROADCAST STATE)*
>
> public interface EventConfigState {
>
>
> void createOrUpdateState(String key, DataPair dataPair);
>
> List<OutputMessage>executeRule(InputMessage inputMessage);
>
> Map<String, Set<DataPair>>getCurrentState();
> }
>
>
> *DERIVED EVENT CONFIG STATE IMPLEMENTATION*
>
> public class DerivedEventConfigStateimplements EventConfigState {
>
> Loggerlogger = LoggerFactory.getLogger(DerivedEventConfigState.class);
> private Map<String, Set<DataPair>>derivedConfigMap;
>
> public DerivedEventConfigState() {
> derivedConfigMap =new HashMap<>();
> }
>
> public void createOrUpdateState(String key, DataPair dataPair) {
>
> derivedConfigMap.putIfAbsent(key, new HashSet<>());
> if (derivedConfigMap.get(key).contains(dataPair)) {
> derivedConfigMap.get(key).remove(dataPair);
> }
> derivedConfigMap.get(key).add(dataPair);
> }
>
> @Override
> public List<OutputMessage>executeRule(InputMessage inputMessage) {
>
> String key = inputMessage.getKey();
> List<OutputMessage> outputMessageList =new ArrayList<>();
>
> if (derivedConfigMap.size() ==0) {
> logger.error("DerivedEventConfigMap is empty");
> return outputMessageList;
> }
>
> if (derivedConfigMap.get(key) ==null) {
> return outputMessageList;
> }
>
> for (DataPair dataPair :derivedConfigMap.get(key)) {
> IRule rule = dataPair.getRule();
> if (rule.isSatisfied(inputMessage)) {
>
> IMessageBuilder messageBuilder =
> MessageBuilderFactory.getMessageBuilder("OutputMessage");
> OutputMessage outputMessage = messageBuilder.build(
> inputMessage,
> dataPair.getEventMessageDefinition()
> );
> outputMessageList.add(outputMessage);
> }
> }
> return outputMessageList;
> }
>
> @Override
> public Map<String, Set<DataPair>>getCurrentState() {
> return Collections.unmodifiableMap(derivedConfigMap);
> }
>
> @Override
> public StringtoString() {
> return "DerivedEventConfigState{"
> +"derivedConfigMap=" +derivedConfigMap
> +'}';
> }
> }
>
>
> Attached are three Exceptions thrown rando
>
> *Caused by: org.apache.flink.util.SerializedThrowable: TABLE-OP
> at java.net.URLClassLoader.findClass(Unknown Source) ~[?:?]
> at java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?]
> at
> org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:64)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:74)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:48)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?]
> at
> org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:172)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at java.lang.Class.forName0(Native Method) ~[?:?]
> at java.lang.Class.forName(Unknown Source) ~[?:?]
> at
> com.esotericsoftware.kryo.util.DefaultClassResolver.readName(DefaultClassResolver.java:136)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:115)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:752)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:116)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:22)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:143)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:21)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:657)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:273)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.api.common.typeutils.base.MapSerializer.copy(MapSerializer.java:111)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.runtime.state.HeapBroadcastState.<init>(HeapBroadcastState.java:69)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.runtime.state.HeapBroadcastState.deepCopy(HeapBroadcastState.java:84)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.runtime.state.HeapBroadcastState.deepCopy(HeapBroadcastState.java:40)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.runtime.state.DefaultOperatorStateBackendSnapshotStrategy.snapshot(DefaultOperatorStateBackendSnapshotStrategy.java:100)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.runtime.state.DefaultOperatorStateBackend.snapshot(DefaultOperatorStateBackend.java:234)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:213)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:162)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:371)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointStreamOperator(SubtaskCheckpointCoordinatorImpl.java:686)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.buildOperatorSnapshotFutures(SubtaskCheckpointCoordinatorImpl.java:607)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.takeSnapshotSync(SubtaskCheckpointCoordinatorImpl.java:572)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:298)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$9(StreamTask.java:1004)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:988)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:947)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> ... 14 more*
>
>
> *Caused by: org.apache.flink.util.SerializedThrowable: Encountered
> unregistered class ID: 97 Serialization trace: derivedConfigMap
> (com.org.app.model.producer.DerivedEventConfigState) at
> com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:119)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:752)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:135)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:21)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:143)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:21)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:657)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:273)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.api.common.typeutils.base.MapSerializer.copy(MapSerializer.java:111)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.<init>(HeapBroadcastState.java:69)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.deepCopy(HeapBroadcastState.java:84)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.deepCopy(HeapBroadcastState.java:40)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.DefaultOperatorStateBackendSnapshotStrategy.snapshot(DefaultOperatorStateBackendSnapshotStrategy.java:100)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.DefaultOperatorStateBackend.snapshot(DefaultOperatorStateBackend.java:234)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:213)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:162)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:371)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointStreamOperator(SubtaskCheckpointCoordinatorImpl.java:686)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.buildOperatorSnapshotFutures(SubtaskCheckpointCoordinatorImpl.java:607)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.takeSnapshotSync(SubtaskCheckpointCoordinatorImpl.java:572)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:298)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$9(StreamTask.java:1004)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:988)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:947)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] ... 14 more*
>
>
> *Caused by: org.apache.flink.util.SerializedThrowable: Buffer
> underflow. Serialization trace: logger
> (com.org.app.model.producer.AppEventConfigState) at
> com.esotericsoftware.kryo.io.Input.require(Input.java:181)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.io.Input.readVarInt(Input.java:355)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:109)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:99)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:657)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:273)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.api.common.typeutils.base.MapSerializer.copy(MapSerializer.java:111)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.<init>(HeapBroadcastState.java:69)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.deepCopy(HeapBroadcastState.java:84)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.HeapBroadcastState.deepCopy(HeapBroadcastState.java:40)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.DefaultOperatorStateBackendSnapshotStrategy.snapshot(DefaultOperatorStateBackendSnapshotStrategy.java:100)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.runtime.state.DefaultOperatorStateBackend.snapshot(DefaultOperatorStateBackend.java:234)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:213)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2] ... 24 more*
>
> Thanks
> Prasanna,
>