Cannot instantiate user function, unread block data

[Metla,Sujitha]

Hello,

 

We are trying to run Beam application with Flink runner.

We are using Beam API-2.10, Kafka-0.10.2.1 and Flink-1.7.

 

Using direct-runner we made sure that the pipeline works fine end-to-end. But when we are trying to run the application using flink-runner we are facing this issue. I am not able to debug it, as the stacktrace is so generic. Can someone help me debug it?

The initial research suggested I might be using different versions of Java, I ruled out that.

 

Command Using: ./flink-1.7.2/bin/flink run -m yarn-cluster -yqu test -yjm 1024 -ytm 8GB <jar> --runner=FlinkRunner <args>

 

 

------------------------------------------------------------

 The program finished with the following exception:

 

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error.

                at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:546)

                at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421)

                at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:427)

                at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:813)

                at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:287)

                at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:213)

                at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1050)

                at org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1126)

                at java.security.AccessController.doPrivileged(Native Method)

                at javax.security.auth.Subject.doAs(Subject.java:422)

                at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1671)

                at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)

                at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1126)

Caused by: java.lang.RuntimeException: Pipeline execution failed

                at org.apache.beam.runners.flink.FlinkRunner.run(FlinkRunner.java:117)

                at org.apache.beam.sdk.Pipeline.run(Pipeline.java:313)

                at org.apache.beam.sdk.Pipeline.run(Pipeline.java:299)

                at com.cerner.healtheintent.longrecord.allergy.processing.pipeline.beam.AllergyPipeline.run(AllergyPipeline.java:257)

                at com.cerner.healtheintent.longrecord.allergy.processing.pipeline.runner.Main.main(Main.java:161)

                at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

                at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)

                at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

                at java.lang.reflect.Method.invoke(Method.java:498)

                at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529)

                ... 12 more

Caused by: org.apache.flink.client.program.ProgramInvocationException: Job failed. (JobID: 20ceb7214a568c7413080f8029a33a85)

                at org.apache.flink.client.program.rest.RestClusterClient.submitJob(RestClusterClient.java:268)

                at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:487)

                at org.apache.flink.streaming.api.environment.StreamContextEnvironment.execute(StreamContextEnvironment.java:66)

                at org.apache.beam.runners.flink.FlinkPipelineExecutionEnvironment.executePipeline(FlinkPipelineExecutionEnvironment.java:125)

                at org.apache.beam.runners.flink.FlinkRunner.run(FlinkRunner.java:114)

                ... 21 more

Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.

                at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:146)

                at org.apache.flink.client.program.rest.RestClusterClient.submitJob(RestClusterClient.java:265)

                ... 25 more

Caused by: org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot instantiate user function.

                at org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperator(StreamConfig.java:239)

                at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:369)

                at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:296)

                at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:360)

                at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:296)

                at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:360)

                at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:296)

                at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:360)

                at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:296)

                at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:360)

                at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:296)

                at org.apache.flink.streaming.runtime.tasks.OperatorChain.<init>(OperatorChain.java:133)

                at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:267)

                at org.apache.flink.runtime.taskmanager.Task.run(Task.java:704)

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

Caused by: java.lang.IllegalStateException: unread block data

                at java.io.ObjectInputStream$BlockDataInputStream.setBlockDataMode(ObjectInputStream.java:2783)

                at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1605)

                at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2287)

                at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2211)

                at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2069)

                at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573)

                at java.io.ObjectInputStream.readObject(ObjectInputStream.java:431)

                at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:566)

                at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:552)

                at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:540)

                at org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:501)

                at org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperator(StreamConfig.java:224)

                ... 14 more

 

Thank you,

Sujitha

 

CONFIDENTIALITY NOTICE This message and any included attachments are from Cerner Corporation and are intended only for the addressee. The information contained in this message is confidential and may constitute inside or non-public information under international, federal, or state securities laws. Unauthorized forwarding, printing, copying, distribution, or use of such information is strictly prohibited and may be unlawful. If you are not the addressee, please promptly delete this message and notify the sender of the delivery error by e-mail or you may call Cerner's corporate offices in Kansas City, Missouri, U.S.A at (+1) [phone].