[FLINK-10160] Index out of bound when serializing operator state

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Duplicate
• Affects Version/s: 1.4.2

Environment

Cloudera Hadoop 2.6.5

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

Scala 2.11

• Assignee: Unassigned
• Reporter: Zdenek Tison
• Created: 16/Aug/18 12:48
• Updated: 28/Feb/19 13:20
• Resolved: 16/Aug/18 15:25

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

Issue Links

• duplicates: FLINK-8836 Duplicating a KryoSerializer does not duplicate registered default serializers