Flink :- No JobSubmissionResult returned, please make sure you called ExecutionEnvironment.execute()

Image courtesy — https://flink.apache.org

Last week I was working on one of my Apache Flink based project, in which I added several more sub-tasks to the stream. And while submitting the job through flink CLI I got No JobSubmissionResult returned error.

We know that No JobSubmissionResult occurs also when we forget to call ExecutionEnvironment.execute(). But in my case I had called env.execute(“Stream Job”) still it failed when I tried to deploy using Flink CLI using command, although when I tried to deploy the task jars to the web interface, task got deployed successfully.

./flink run /opt/flink-jobs/stock-ohlc-job.jar

I noticed below in the stack trace, which threw the job not found exception.

ERROR akka.remote.EndpointWriter - Transient association error (association remains live)
akka.remote.OversizedPayloadException: Discarding oversized payload sent to Actor[akka.tcp://flink@172.XX.XX.XX:6123/user/jobmanager#467971694]: max allowed size 10485760 bytes, actual size of encoded class org.apache.flink.runtime.messages.JobManagerMessages$LeaderSessionMessage was 13846328 bytes.
Caused by: org.apache.flink.runtime.client.JobExecutionException: Couldn't retrieve the JobExecutionResult from the JobManager.
at org.apache.flink.runtime.client.JobClient.awaitJobResult(JobClient.java:300)
at org.apache.flink.runtime.client.JobClient.submitJobAndWait(JobClient.java:387)
at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:481)
... 19 more
Caused by: org.apache.flink.runtime.client.JobClientActorSubmissionTimeoutException: Job submission to the JobManager timed out. You may increase 'akka.client.timeout' in case the JobManager needs more time to configure and confirm the job submission.
at org.apache.flink.runtime.client.JobSubmissionClientActor.handleCustomMessage(JobSubmissionClientActor.java:127)
at org.apache.flink.runtime.client.JobClientActor.handleMessage(JobClientActor.java:251)
at org.apache.flink.runtime.akka.FlinkUntypedActor.handleLeaderSessionID(FlinkUntypedActor.java:92)
at org.apache.flink.runtime.akka.FlinkUntypedActor.onReceive(FlinkUntypedActor.java:71)
at akka.actor.UntypedActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:165)
at akka.actor.Actor$class.aroundReceive(Actor.scala:502)
at akka.actor.UntypedActor.aroundReceive(UntypedActor.scala:95)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
at akka.actor.ActorCell.invoke(ActorCell.scala:495)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
at akka.dispatch.Mailbox.run(Mailbox.scala:224)
at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
2018

We can see part of the stacktrace in bold, its suggesting that it discarded some part of the payload, I researched on internet and came to know with following solution.

I added below line to flink-conf.yml

Solution :- increase Akka Frame Size in flink-conf.yml

## Akka
akka.framesize: 209715200b

Reference link : http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/flink-akka-OversizedPayloadException-error-td12218.html

Although stacktrace also suggested about increasing ‘akka.client.timeout’ bit it didn’t work in my case.