Error in KyroSerializer

[Pankaj Chaudhary]

Hi,

I am on Flink 1.4.2 and as part of my operator logic (i.e. RichFlatMapFunction) I am collecting the values in the Collector object.

But I am getting an error stating “Caused by: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator”

On debugging it looks like the root cause of this exception is in KyroSerializer where its try to do some copy operation. Can some one please let me know how I can get around this issue.

Below is the stack trace of the error

Caused by: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator

at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:566)

at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:524)

at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:504)

at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:830)

at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:808)

at org.apache.flink.streaming.api.operators.StreamFilter.processElement(StreamFilter.java:40)

at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:549)

... 11 more

Caused by: java.lang.ArrayIndexOutOfBoundsException: -14

at com.esotericsoftware.kryo.util.IntArray.pop(IntArray.java:157)

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

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

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

at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:547)

... 17 more

Regards,

Pankaj.

[Rong Rong]

This sounds very much related to FLINK-10160 [1]. 

Would you mind upgrading your Flink version to 1.4.3 and try again?

Thanks,

Rong

[1] https://issues.apache.org/jira/browse/FLINK-10160

On Fri, Aug 17, 2018 at 4:20 PM Pankaj Chaudhary <[hidden email]> wrote:

Hi,

I am on Flink 1.4.2 and as part of my operator logic (i.e. RichFlatMapFunction) I am collecting the values in the Collector object.

But I am getting an error stating “Caused by: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator”

On debugging it looks like the root cause of this exception is in KyroSerializer where its try to do some copy operation. Can some one please let me know how I can get around this issue.

Below is the stack trace of the error

Caused by: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator

at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:566)

at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:524)

at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:504)

at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:830)

at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:808)

at org.apache.flink.streaming.api.operators.StreamFilter.processElement(StreamFilter.java:40)

at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:549)

... 11 more

Caused by: java.lang.ArrayIndexOutOfBoundsException: -14

at com.esotericsoftware.kryo.util.IntArray.pop(IntArray.java:157)

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

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

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

at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:547)

... 17 more

Regards,

Pankaj.