Hi, I'm getting this error in yarn application mode when submitting my job.
Caused by: java.lang.ClassCastException: cannot assign instance of org.apache.commons.collections.map.LinkedMap to field org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.pendingOffsetsToCommit of type org.apache.commons.collections.map.LinkedMap in instance of org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2301) ~[?:1.8.0_332] at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1431) ~[?:1.8.0_332] at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2437) ~[?:1.8.0_332] at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2355) ~[?:1.8.0_332] at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2213) ~[?:1.8.0_332] at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1669) ~[?:1.8.0_332] at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2431) ~[?:1.8.0_332] at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2355) ~[?:1.8.0_332] at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2213) ~[?:1.8.0_332] at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1669) ~[?:1.8.0_332] at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2431) ~[?:1.8.0_332] at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2355) ~[?:1.8.0_332] at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2213) ~[?:1.8.0_332] at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1669) ~[?:1.8.0_332] at java.io.ObjectInputStream.readObject(ObjectInputStream.java:503) ~[?:1.8.0_332] at java.io.ObjectInputStream.readObject(ObjectInputStream.java:461) ~[?:1.8.0_332] at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:615) ~[streamingjobs-1.13.jar:?] at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:600) ~[streamingjobs-1.13.jar:?] at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:587) ~[streamingjobs-1.13.jar:?] at org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:541) ~[streamingjobs-1.13.jar:?] at org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperatorFactory(StreamConfig.java:322) ~[flink-dist_2.12-1.13.1.jar:1.13.1] at org.apache.flink.streaming.runtime.tasks.OperatorChain.<init>(OperatorChain.java:154) ~[flink-dist_2.12-1.13.1.jar:1.13.1] at org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:548) ~[flink-dist_2.12-1.13.1.jar:1.13.1] at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:647) ~[flink-dist_2.12-1.13.1.jar:1.13.1] at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:537) ~[flink-dist_2.12-1.13.1.jar:1.13.1] at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759) ~[streamingjobs-1.13.jar:?] at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566) ~[streamingjobs-1.13.jar:?] at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_332]