Uploaded image for project: 'Flink'
  1. Flink
  2. FLINK-10160

Index out of bound when serializing operator state

    XMLWordPrintableJSON

    Details

    • Type: Bug
    • Status: Closed
    • Priority: Major
    • Resolution: Duplicate
    • Affects Version/s: 1.4.2
    • Fix Version/s: None
    • Component/s: None
    • Labels:
      None
    • Environment:

      Cloudera Hadoop 2.6.5

      JVM: Java HotSpot(TM) 64-Bit Server VM - Oracle Corporation - 1.8/25.121-b13

      Scala 2.11

      Description

      I am getting unexpected randomly happend error when checkpoint state is serialized to state backend (In-Memory). 

      Source code usage:

      
      env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
      
      env.registerType(classOf[ImpressionMetric])
      env.registerType(classOf[RecommendationMetric])
      env.registerType(classOf[AdRequestMetric])
      env.registerType(classOf[AdRequestStatusMetric])
      
      val impressionsStream = env
          .addSource(impsSource)
          .map(imp => ImpressionMetric(imp))
          .map(imp => imp.asInstanceOf[DomainMetric])
          .name("Impressions")
      
      
      val recommendationsStream = env
          .addSource(recommendationsSource)
          .map(rcmd => RecommendationMetric(rcmd))
          .map(rcmd => rcmd.asInstanceOf[DomainMetric])
          .name("Recommendations")
      
      
      val adRequestsStream = env
          .addSource(adRequestsSource)
          .flatMap(new MapToAdRequestDomainMetrics())
          .name("Ad requests")
      
      

      Log with error, longer version in log.txt

      2018-08-16 12:49:37,663 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 1 @ 1534416577551
      2018-08-16 12:49:38,100 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed checkpoint 1 (1404248 bytes in 545 ms).
      2018-08-16 12:57:54,217 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 2 @ 1534417074216
      2018-08-16 12:57:54,368 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed checkpoint 2 (2199656 bytes in 152 ms).
      2018-08-16 13:07:54,217 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 3 @ 1534417674216
      2018-08-16 13:17:54,218 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Checkpoint 3 expired before completing.
      2018-08-16 13:17:54,220 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 4 @ 1534418274220
      2018-08-16 13:27:54,220 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Checkpoint 4 expired before completing.
      2018-08-16 13:27:54,222 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 5 @ 1534418874221
      2018-08-16 13:29:36,640 WARN org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Received late message for now expired checkpoint attempt 3 from e4ce857f2ef5dd4cc75a48b6fdb7b694 of job a98b29dcef6ad8eac28f8290034a0590.
      2018-08-16 13:29:36,642 WARN org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Received late message for now expired checkpoint attempt 3 from 1db1e9c995bae21d6c81e828e11f50a6 of job a98b29dcef6ad8eac28f8290034a0590.
      2018-08-16 13:29:36,661 WARN org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Received late message for now expired checkpoint attempt 4 from 0f62f47cdd04acf58cf8aa6c11288950 of job a98b29dcef6ad8eac28f8290034a0590.
      2018-08-16 13:29:36,662 WARN org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Received late message for now expired checkpoint attempt 4 from e4ce857f2ef5dd4cc75a48b6fdb7b694 of job a98b29dcef6ad8eac28f8290034a0590.
      2018-08-16 13:29:36,663 WARN org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Received late message for now expired checkpoint attempt 3 from 0f62f47cdd04acf58cf8aa6c11288950 of job a98b29dcef6ad8eac28f8290034a0590.
      2018-08-16 13:29:36,967 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: Custom Source -> Ad requests (1/2) (1db1e9c995bae21d6c81e828e11f50a6) switched from RUNNING to FAILED.
      java.lang.Exception: Error while triggering checkpoint 4 for Source: Custom Source -> Ad requests (1/2)
      at org.apache.flink.runtime.taskmanager.Task$2.run(Task.java:1210)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
      at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
      at java.lang.Thread.run(Thread.java:745)
      Caused by: java.lang.Exception: Could not perform checkpoint 4 for operator Source: Custom Source -> Ad requests (1/2).
      at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:544)
      at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.triggerCheckpoint(SourceStreamTask.java:111)
      at org.apache.flink.runtime.taskmanager.Task$2.run(Task.java:1199)
      ... 5 more
      Caused by: java.lang.Exception: Could not complete snapshot 4 for operator Source: Custom Source -> Ad requests (1/2).
      at org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:378)
      at org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.checkpointStreamOperator(StreamTask.java:1089)
      at org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.executeCheckpointing(StreamTask.java:1038)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.checkpointState(StreamTask.java:671)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:607)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:538)
      ... 7 more
      Caused by: java.lang.ArrayIndexOutOfBoundsException: -1
      at com.esotericsoftware.kryo.util.IntArray.add(IntArray.java:61)
      at com.esotericsoftware.kryo.Kryo.readReferenceOrNull(Kryo.java:800)
      at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:655)
      at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:189)
      at org.apache.flink.api.java.typeutils.runtime.TupleSerializer.copy(TupleSerializer.java:104)
      at org.apache.flink.api.java.typeutils.runtime.TupleSerializer.copy(TupleSerializer.java:30)
      at org.apache.flink.runtime.state.ArrayListSerializer.copy(ArrayListSerializer.java:74)
      at org.apache.flink.runtime.state.DefaultOperatorStateBackend$PartitionableListState.<init>(DefaultOperatorStateBackend.java:448)
      at org.apache.flink.runtime.state.DefaultOperatorStateBackend$PartitionableListState.deepCopy(DefaultOperatorStateBackend.java:460)
      at org.apache.flink.runtime.state.DefaultOperatorStateBackend.snapshot(DefaultOperatorStateBackend.java:220)
      at org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:363)
      ... 12 more
      

        Attachments

        1. log.txt
          29 kB
          Zdenek

          Issue Links

            Activity

              People

              • Assignee:
                Unassigned
                Reporter:
                tisonet Zdenek
              • Votes:
                0 Vote for this issue
                Watchers:
                2 Start watching this issue

                Dates

                • Created:
                  Updated:
                  Resolved: