Flink: Why does a break statement in loop breaking the Flink Job

Question

There is a richCoFlatmap in which a class is instantiated for every input event. The class has some class variables and method definitions. The class also has calls to those method definition using in if constructs based on certain conditions. The if blocks calling the method definitions are in a for loop. In some cased If a condition is met I want the loop to exist and not call the other functions. For this when I use break the. Instead of exiting the loop and going back to the RichCoFlatMap the Flink job is exited and terminated. The code is Flink using Scala

I would like to know if there is something I am missing or Flink is not break statement friendly.
And is there any way to achieve this without failing the job.

Below is the error for reference

at akka.actor.ActorCell.invoke(ActorCell.scala:561)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
    at akka.dispatch.Mailbox.run(Mailbox.scala:225)
    at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
    ... 4 more
Caused by: scala.util.control.BreakControl

Answer 1

Most certainly the issue is that You haven't used the breakable when calling the break. This is one of the differences between Java and Scala.

Scala handles break differently than Java and it needs a special construct called the breakable block.

You can find more info here/