flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Biplob Biswas <revolutioni...@gmail.com>
Subject Regarding Concurrent Modification Exception
Date Sat, 13 Feb 2016 17:55:51 GMT
Hi,
We are getting a ConcurrentModificationException, the complete stack trace
is as follows:

org.apache.flink.optimizer.CompilerException: Error translating node 'Data
> Source "at compute(ArpackSVD.java:367)
> (org.apache.flink.api.java.io.CollectionInputFormat)" : NONE [[
> GlobalProperties [partitioning=RANDOM_PARTITIONED] ]] [[ LocalProperties
> [ordering=null, grouped=null, unique=null] ]]':
> java.util.ConcurrentModificationException
> Serialization trace:
> classes (sun.misc.Launcher$AppClassLoader)
> classLoader (akka.actor.ReflectiveDynamicAccess)
> _pm (akka.actor.ActorSystemImpl)
> actorSystem (org.apache.flink.client.program.Client)
> client (org.apache.flink.client.program.ContextEnvironment)
> context (org.apache.flink.api.java.operators.MapOperator)
> matrix (flink.pca.impl.svd.ArpackSVD)
> this$0 (flink.pca.impl.svd.ArpackSVD$ArpackContext)
> at
> org.apache.flink.optimizer.plantranslate.JobGraphGenerator.preVisit(JobGraphGenerator.java:386)
> at
> org.apache.flink.optimizer.plantranslate.JobGraphGenerator.preVisit(JobGraphGenerator.java:109)
> at
> org.apache.flink.optimizer.plan.SourcePlanNode.accept(SourcePlanNode.java:86)
> at
> org.apache.flink.optimizer.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:199)
> at
> org.apache.flink.optimizer.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:199)
> at
> org.apache.flink.optimizer.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:199)
> at
> org.apache.flink.optimizer.plan.OptimizedPlan.accept(OptimizedPlan.java:128)
> at
> org.apache.flink.optimizer.plantranslate.JobGraphGenerator.compileJobGraph(JobGraphGenerator.java:188)
> at
> org.apache.flink.optimizer.plantranslate.JobGraphGenerator.compileJobGraph(JobGraphGenerator.java:166)
> at org.apache.flink.client.program.Client.getJobGraph(Client.java:534)
> at org.apache.flink.client.program.Client.runBlocking(Client.java:347)
> at org.apache.flink.client.program.Client.runBlocking(Client.java:315)
> at
> org.apache.flink.client.program.ContextEnvironment.execute(ContextEnvironment.java:70)
> at
> org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:804)
> at org.apache.flink.api.java.DataSet.collect(DataSet.java:410)
> at flink.pca.impl.svd.ArpackSVD.compute(ArpackSVD.java:379)
> at flink.pca.impl.PCA.computeSVD(PCA.java:110)
> at flink.pca.impl.PCA.project(PCA.java:35)
> at flink.pca.impl.PCASystemTest.runFlinkJob(PCASystemTest.java:90)
> at flink.pca.impl.PCASystemTest.main(PCASystemTest.java:61)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:497)
> at
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:497)
> at
> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:395)
> at org.apache.flink.client.program.Client.runBlocking(Client.java:252)
> at
> org.apache.flink.client.CliFrontend.executeProgramBlocking(CliFrontend.java:676)
> at org.apache.flink.client.CliFrontend.run(CliFrontend.java:326)
> at
> org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:978)
> at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1028)
> Caused by: com.esotericsoftware.kryo.KryoException:
> java.util.ConcurrentModificationException
> Serialization trace:
> classes (sun.misc.Launcher$AppClassLoader)
> classLoader (akka.actor.ReflectiveDynamicAccess)
> _pm (akka.actor.ActorSystemImpl)
> actorSystem (org.apache.flink.client.program.Client)
> client (org.apache.flink.client.program.ContextEnvironment)
> context (org.apache.flink.api.java.operators.MapOperator)
> matrix (flink.pca.impl.svd.ArpackSVD)
> this$0 (flink.pca.impl.svd.ArpackSVD$ArpackContext)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:82)
> at
> com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
> at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
> at
> com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
> at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
> at
> com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
> at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
> at
> com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
> at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
> at
> com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
> at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
> at
> com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
> at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
> at
> com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
> at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
> at
> com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
> at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:599)
> at
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.serialize(KryoSerializer.java:193)
> at
> org.apache.flink.api.java.io.CollectionInputFormat.writeObject(CollectionInputFormat.java:88)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:497)
> at java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:1028)
> at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1496)
> at
> java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
> at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
> at
> java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
> at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
> at
> java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
> at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
> at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
> at
> org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:307)
> at
> org.apache.flink.util.InstantiationUtil.writeObjectToConfig(InstantiationUtil.java:259)
> at
> org.apache.flink.runtime.operators.util.TaskConfig.setStubWrapper(TaskConfig.java:273)
> at
> org.apache.flink.optimizer.plantranslate.JobGraphGenerator.createDataSourceVertex(JobGraphGenerator.java:890)
> at
> org.apache.flink.optimizer.plantranslate.JobGraphGenerator.preVisit(JobGraphGenerator.java:286)
> ... 30 more
> Caused by: java.util.ConcurrentModificationException
> at java.util.Vector$Itr.checkForComodification(Vector.java:1184)
> at java.util.Vector$Itr.next(Vector.java:1137)
> at
> com.esotericsoftware.kryo.serializers.CollectionSerializer.write(CollectionSerializer.java:74)
> at
> com.esotericsoftware.kryo.serializers.CollectionSerializer.write(CollectionSerializer.java:22)
> at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
> at
> com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
> ... 73 more



Can anyone enlighten us as why is it like this or how to fix this issue? We
did a bit of google search, but all we get is some problem with serializing
broadcast variable. We use flink bulk iterations and this variable is
broadcasted to both map and reduce in one dataflow!

Thanks & Regards
Biplob Biswas

Mime
View raw message