Uploaded image for project: 'Flink'
  1. Flink
  2. FLINK-7830 Problematic interaction of CEP and asynchronous snapshots
  3. FLINK-7484

CaseClassSerializer.duplicate() does not perform proper deep copy

    XMLWordPrintableJSON

    Details

      Description

      I am using many CEP's and Global Window. I am getting following error sometimes and application crashes. I have checked logically there's no flow in the program. Here ItemPojo is a Pojo class and we are using java.utill.List[ItemPojo]. We are using Scala DataStream API please find attached logs.

      2017-08-17 10:04:12,814 INFO  org.apache.flink.runtime.taskmanager.Task                     - TriggerWindow(GlobalWindows(), ListStateDescriptor{serializer=org.apache.flink.api.common.typeutils.base.ListSerializer@6d36aa3c}, co.thirdwatch.trigger.TransactionTrigger@5707c1cb, WindowedStream.apply(WindowedStream.scala:582)) -> Flat Map -> Map -> Sink: Saving CSV Features Sink (1/2) (06c0d4d231bc620ba9e7924b9b0da8d1) switched from RUNNING to FAILED.
      com.esotericsoftware.kryo.KryoException: java.lang.IndexOutOfBoundsException: Index: 7, Size: 5
      Serialization trace:
      category (co.thirdwatch.pojo.ItemPojo)
      underlying (scala.collection.convert.Wrappers$SeqWrapper)
      	at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)
      	at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
      	at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
      	at com.twitter.chill.TraversableSerializer.read(Traversable.scala:43)
      	at com.twitter.chill.TraversableSerializer.read(Traversable.scala:21)
      	at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
      	at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
      	at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
      	at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:657)
      	at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:190)
      	at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:101)
      	at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:32)
      	at org.apache.flink.runtime.state.ArrayListSerializer.copy(ArrayListSerializer.java:74)
      	at org.apache.flink.runtime.state.ArrayListSerializer.copy(ArrayListSerializer.java:34)
      	at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.get(CopyOnWriteStateTable.java:279)
      	at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.get(CopyOnWriteStateTable.java:296)
      	at org.apache.flink.runtime.state.heap.HeapListState.add(HeapListState.java:77)
      	at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.processElement(WindowOperator.java:442)
      	at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:206)
      	at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:69)
      	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:263)
      	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
      	at java.lang.Thread.run(Thread.java:745)
      Caused by: java.lang.IndexOutOfBoundsException: Index: 7, Size: 5
      	at java.util.ArrayList.rangeCheck(ArrayList.java:653)
      	at java.util.ArrayList.get(ArrayList.java:429)
      	at com.esotericsoftware.kryo.util.MapReferenceResolver.getReadObject(MapReferenceResolver.java:42)
      	at com.esotericsoftware.kryo.Kryo.readReferenceOrNull(Kryo.java:805)
      	at com.esotericsoftware.kryo.Kryo.readObjectOrNull(Kryo.java:728)
      	at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:113)
      	... 22 more
      2017-08-17 10:04:12,816 INFO  org.apache.flink.runtime.taskmanager.Task                     - Freeing task resources for TriggerWindow(GlobalWindows(), ListStateDescriptor{serializer=org.apache.flink.api.common.typeutils.base.ListSerializer@6d36aa3c}, co.thirdwatch.trigger.TransactionTrigger@5707c1cb, WindowedStream.apply(WindowedStream.scala:582)) -> Flat Map -> Map -> Sink: Saving CSV Features Sink (1/2) (06c0d4d231bc620ba9e7924b9b0da8d1).
      2017-08-17 10:04:12,816 INFO  org.apache.flink.runtime.taskmanager.Task                     - Ensuring all FileSystem streams are closed for task TriggerWindow(GlobalWindows(), ListStateDescriptor{serializer=org.apache.flink.api.common.typeutils.base.ListSerializer@6d36aa3c}, co.thirdwatch.trigger.TransactionTrigger@5707c1cb, WindowedStream.apply(WindowedStream.scala:582)) -> Flat Map -> Map -> Sink: Saving CSV Features Sink (1/2) (06c0d4d231bc620ba9e7924b9b0da8d1) [FAILED]
      
      

        Attachments

          Issue Links

            Activity

              People

              • Assignee:
                aljoscha Aljoscha Krettek
                Reporter:
                shashank734 Shashank Agarwal
              • Votes:
                0 Vote for this issue
                Watchers:
                7 Start watching this issue

                Dates

                • Created:
                  Updated:
                  Resolved: