(DEPRECATED) Apache Flink User Mailing List archive.

Error while trigger checkpoint due to Kyro Exception

classic Classic list List threaded Threaded
♦
♦
 2 messages Options
Bruce Qiu
Reply | Threaded
Open this post in threaded view
♦
♦
|

Error while trigger checkpoint due to Kyro Exception

Bruce Qiu

Hi Community,

I am using Flink 1.4.2 to do streaming processing. I fetch data from Kafka and write the parquet file to HDFS. In the previous environment, the Kafka had 192 partitions and I set the source parallelism to 192, the application works fine. But recently we had increased the Kafka paritions to 384. So I changed the source parallelism to 384. After I made this change, the application throws the exception as blow, and the checkpoint is always fail. Also I saw the backpressure is very high in the ColFlatMap stage. My application’s DAG as blow. Can someone helps me about this exception, thanks a lot.

 

DAG Stage

 

 

 

Exception stack trace:

 

java.lang.Exception: Error while triggering checkpoint 109 for Source: Custom Source (257/384)

      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:1149)

      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)

      at java.lang.Thread.run(Thread.java:748)

Caused by: java.lang.Exception: Could not perform checkpoint 109 for operator Source: Custom Source (257/384).

      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 109 for operator Source: Custom Source (257/384).

      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.pop(IntArray.java:157)

      at com.esotericsoftware.kryo.Kryo.reference(Kryo.java:822)

    at com.esotericsoftware.kryo.serializers.FieldSerializer.copy(FieldSerializer.java:625)

      at com.esotericsoftware.kryo.Kryo.copy(Kryo.java:862)

      at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:175)

      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

 

 

Regards,

Bruce
Stefan Richter
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Error while trigger checkpoint due to Kyro Exception

Stefan Richter
Hi,

this problem is fixed in Flink >= 1.4.3, see https://issues.apache.org/jira/browse/FLINK-8836.

Best,
Stefan

Am 18.08.2018 um 05:07 schrieb Bruce Qiu <[hidden email]>:

Hi Community,
I am using Flink 1.4.2 to do streaming processing. I fetch data from Kafka and write the parquet file to HDFS. In the previous environment, the Kafka had 192 partitions and I set the source parallelism to 192, the application works fine. But recently we had increased the Kafka paritions to 384. So I changed the source parallelism to 384. After I made this change, the application throws the exception as blow, and the checkpoint is always fail. Also I saw the backpressure is very high in the ColFlatMap stage. My application’s DAG as blow. Can someone helps me about this exception, thanks a lot.
 
DAG Stage
 
<image001.png>
 
 
Exception stack trace:
 
java.lang.Exception: Error while triggering checkpoint 109 for Source: Custom Source (257/384)
      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:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.Exception: Could not perform checkpoint 109 for operator Source: Custom Source (257/384).
      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 109 for operator Source: Custom Source (257/384).
      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.pop(IntArray.java:157)
      at com.esotericsoftware.kryo.Kryo.reference(Kryo.java:822)
    at com.esotericsoftware.kryo.serializers.FieldSerializer.copy(FieldSerializer.java:625)
      at com.esotericsoftware.kryo.Kryo.copy(Kryo.java:862)
      at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:175)
      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
 
 
Regards,
Bruce
<image001.png>
Free forum by Nabble Edit this page