[
https://issues.apache.org/jira/browse/FLINK-2800?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14945146#comment-14945146
]
Stefano Bortoli commented on FLINK-2800:
----------------------------------------
Ok, I have some hints. Executing the cross alone did not create any problem,
but adding the following join cause the race condition.
{code}
DataSet<Tuple2<String, BSONObject>> input1 = env.createFromElements(....)
DataSet<Tuple2<String, BSONObject>> input2 = env.createFromElements(....)
DataSet<Tuple2<String, String>> out = input1.cross(input2).combineGroup(....);
DataSet<Tuple3<String, String, BSONObject>> union =
out.join(input1).where(0).equalTo(0).projectFirst(0, 1).projectSecond(1);
{code}
is causing the race condition with the Kryo serialize, as shown by the
exception below.
{quote}
2015-10-06 16:54:34 ERROR RegularPactTask:1179 - Error in task code:
Cross(Cross at main(FlinkMongoHadoop2Cross.java:147)) (1/4)
java.lang.IndexOutOfBoundsException: Index: 98, Size: 73
at java.util.ArrayList.rangeCheck(ArrayList.java:635)
at java.util.ArrayList.get(ArrayList.java:411)
at
com.esotericsoftware.kryo.util.MapReferenceResolver.getReadObject(MapReferenceResolver.java:42)
at com.esotericsoftware.kryo.Kryo.readReferenceOrNull(Kryo.java:805)
at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:759)
at
com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:135)
at
com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:21)
at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
at
org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:211)
at
org.apache.flink.api.java.typeutils.runtime.TupleSerializer.deserialize(TupleSerializer.java:127)
at
org.apache.flink.api.java.typeutils.runtime.TupleSerializer.deserialize(TupleSerializer.java:30)
at
org.apache.flink.runtime.operators.resettable.AbstractBlockResettableIterator.getNextRecord(AbstractBlockResettableIterator.java:180)
at
org.apache.flink.runtime.operators.resettable.BlockResettableMutableObjectIterator.next(BlockResettableMutableObjectIterator.java:111)
at
org.apache.flink.runtime.operators.CrossDriver.runBlockedOuterSecond(CrossDriver.java:309)
at
org.apache.flink.runtime.operators.CrossDriver.run(CrossDriver.java:162)
at
org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:489)
at
org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:354)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:579)
at java.lang.Thread.run(Thread.java:745)
{quote}
> kryo serialization problem
> --------------------------
>
> Key: FLINK-2800
> URL: https://issues.apache.org/jira/browse/FLINK-2800
> Project: Flink
> Issue Type: Bug
> Components: Type Serialization System
> Affects Versions: 1.0
> Environment: linux ubuntu 12.04 LTS, Java 7
> Reporter: Stefano Bortoli
>
> Performing a cross of two dataset of POJOs I have got the exception below.
> The first time I run the process, there was no problem. When I run it the
> second time, I have got the exception. My guess is that it could be a race
> condition related to the reuse of the Kryo serializer object. However, it
> could also be "a bug where type registrations are not properly forwarded to
> all Serializers", as suggested by Stephan.
> ------------------------------------------------------------------------
> 2015-10-01 18:18:21 INFO JobClient:161 - 10/01/2015 18:18:21 Cross(Cross at
> main(FlinkMongoHadoop2LinkPOI2CDA.java:160))(3/4) switched to FAILED
> com.esotericsoftware.kryo.KryoException: Encountered unregistered class ID:
> 114
> at
> com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:119)
> at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
> at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:752)
> at
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:210)
> at
> org.apache.flink.api.java.typeutils.runtime.TupleSerializer.deserialize(TupleSerializer.java:127)
> at
> org.apache.flink.api.java.typeutils.runtime.TupleSerializer.deserialize(TupleSerializer.java:30)
> at
> org.apache.flink.runtime.operators.resettable.AbstractBlockResettableIterator.getNextRecord(AbstractBlockResettableIterator.java:180)
> at
> org.apache.flink.runtime.operators.resettable.BlockResettableMutableObjectIterator.next(BlockResettableMutableObjectIterator.java:111)
> at
> org.apache.flink.runtime.operators.CrossDriver.runBlockedOuterSecond(CrossDriver.java:309)
> at
> org.apache.flink.runtime.operators.CrossDriver.run(CrossDriver.java:162)
> at
> org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:489)
> at
> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:354)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:581)
> at java.lang.Thread.run(Thread.java:745)
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
