Flink throws InvalidClassException: scala.collection.mutable.WrappedArray$ofRef

[regadas]

This issue happens on serde when Java serialVersionUID is checked. Scala 2.12.x doesn't extensively set @SerialVersionUID across collections which means that in order for this to work in Flink one needs to use the same Scala version. Flink's 1.10.1 which is shipped with Beam 2.23.0 is compiled with 2.12.7.

Fixed it in https://github.com/scala/scala/pull/9166 but it will take effect after 2.12.13

2020-08-13 09:51:17
org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
	at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:110)
	at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:76)
	at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:192)
	at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:186)
	at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:180)
	at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:496)
	at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:380)
	at sun.reflect.GeneratedMethodAccessor35.invoke(Unknown Source)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:284)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:199)
	at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152)
	at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
	at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
	at scala.PartialFunction.applyOrElse(PartialFunction.scala:123)
	at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122)
	at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
	at akka.actor.Actor.aroundReceive(Actor.scala:517)
	at akka.actor.Actor.aroundReceive$(Actor.scala:515)
	at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
	at akka.actor.ActorCell.invoke(ActorCell.scala:561)
	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
	at akka.dispatch.Mailbox.run(Mailbox.scala:225)
	at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
	at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
	at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
	at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
	at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
Caused by: java.lang.RuntimeException: Could not load deserializer from the configuration.
	at org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory.readParametersFromConfig(RuntimeSerializerFactory.java:85)
	at org.apache.flink.runtime.operators.util.TaskConfig.getTypeSerializerFactory(TaskConfig.java:1141)
	at org.apache.flink.runtime.operators.util.TaskConfig.getOutputSerializer(TaskConfig.java:549)
	at org.apache.flink.runtime.operators.BatchTask.getOutputCollector(BatchTask.java:1232)
	at org.apache.flink.runtime.operators.BatchTask.initOutputs(BatchTask.java:1313)
	at org.apache.flink.runtime.operators.BatchTask.initOutputs(BatchTask.java:1030)
	at org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:270)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:708)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:533)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.InvalidClassException: scala.collection.mutable.WrappedArray$ofRef; local class incompatible: stream classdesc serialVersionUID = 3456489343829468865, local class serialVersionUID = 1028182004549731694
	at java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:699)
	at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1964)
	at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1830)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2121)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
	at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
	at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
	at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
	at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
	at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
	at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
	at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
	at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
	at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
	at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
	at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
	at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
	at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
	at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
	at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
	at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
	at java.io.ObjectInputStream.readArray(ObjectInputStream.java:2054)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1635)
	at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
	at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
	at java.io.ObjectInputStream.readObject(ObjectInputStream.java:483)
	at java.io.ObjectInputStream.readObject(ObjectInputStream.java:441)
	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:576)
	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:562)
	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:550)
	at org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:511)
	at org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory.readParametersFromConfig(RuntimeSerializerFactory.java:78)
	... 9 more