I am running a streaming processing application in a LocalStreamEnvironment (embedded flink cluster). I processed a specific dataset using my code successfully a few times. I wanted to rerun the application yesterday after doing some modifications to the processing logic, however after about 3/4 way through the data processing it seems the flink cluster crashes without any reason. Review the condensed log - my comments inserted in sharp brackets <>:
2018-02-09 12:04:05,146 [INFO] from a.b.l.f.MultiS3FileSource in Source: General source (1/1) - inserting 266574 events
2018-02-09 12:10:55,094 [ERROR] from o.a.f.r.c.JobSubmissionClientActor in flink-akka.actor.default-dispatcher-11020 - class org.apache.flink.runtime.client.JobSubmissionClientActor received unknown message:
2018-02-09 12:10:55,245 [WARN] from o.a.f.r.c.JobSubmissionClientActor in flink-akka.actor.default-dispatcher-11019 - Discard message LeaderSessionMessage(7240d925-8573-44e8-996c-fa4658ab0463,02/09/2018 12:10:55 Process -> Detection(7/8) switched to CANCELED ) because there is currently no valid leader id known.
2018-02-09 12:10:55,268 [WARN] from o.a.f.r.c.JobSubmissionClientActor in flink-akka.actor.default-dispatcher-11019 - Discard message LeaderSessionMessage(7240d925-8573-44e8-996c-fa4658ab0463,02/09/2018 12:10:55 Enrichment-> Flat Map(7/8) switched to CANCELED ) because there is currently no valid leader id known.
... <similar messages for all the processing steps>
2018-02-09 12:10:55,509 [ERROR] from o.a.f.s.r.t.StreamTask in PartialAggregations-> Sink: CassandraSink (1/8) - Error during disposal of stream operator.
java.lang.InterruptedException: null <because its interrupting a future>
... <for all of my sinks - these are custom, not the flink cassandra connectors>
The first info message is about my sources reading data from s3 and collecting it into flink.
Afterwards the first error is produced by: https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobSubmissionClientActor.java#L137
and the warnings are produced by: https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/akka/FlinkUntypedActor.java#L115
The last error is in my code but its caused by flink trying to tear down the job, so it should not be the original cause of the error.
I can provide some extra info but Im not sure what would be relevant.
The first error seems to be what cascades the whole crash. How would it be possible for the JobSubmissionClientActor to have a null getLeaderSessionID? What kind of messages does the JobSubmissionClientActor expect if flink is running embedded? It seems to me that all the messages its able to receive are messages concerning submitting jobs. Should that even be possible in embedded mode? How can I prevent this crash?
UPDATE: I think I have misinterpreted the error log. When I ran the execution one more time i got as lightly different order of events. In the previous run I only got errors around the disposal of the stream, with no obvious cause to the stream ending, since the last error seems to not be included in my log file (it was printed out to stdout though). This error is below, the previous errors are similar as in the previous run (errors around disposal of stream).
[error] Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: JobClientActor seems to have died before the JobExecutionResult could be retrieved.[error] at org.apache.flink.runtime.client.JobClient.awaitJobResult(JobClient.java:285)
[error] at org.apache.flink.runtime.client.JobClient.submitJobAndWait(JobClient.java:387)
[error] at org.apache.flink.runtime.minicluster.FlinkMiniCluster.submitJobAndWait(FlinkMiniCluster.scala:565)
[error] at org.apache.flink.runtime.minicluster.FlinkMiniCluster.submitJobAndWait(FlinkMiniCluster.scala:539)
[error] at org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:108)
[error] at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1501)
[error] at org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.scala:629)
[error] at a.b.l.flink.FlinkIngestPrototype$.run(FlinkIngestPrototype.scala:90)
[error] at a.b.l.flink.FlinkIngestPrototype$.main(FlinkIngestPrototype.scala:43)
[error] at a.b.l.flink.FlinkIngestPrototype.main(FlinkIngestPrototype.scala)
[error] Caused by: java.util.concurrent.TimeoutException: Futures timed out after [10000 milliseconds]
[error] at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:223)
[error] at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:227)
[error] at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:190)
[error] at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53)
[error] at scala.concurrent.Await$.result(package.scala:190)
[error] at scala.concurrent.Await.result(package.scala)
[error] at org.apache.flink.runtime.client.JobClient.awaitJobResult(JobClient.java:273)
[error] ... 9 more
Ive traced the execution failure to the following:
The JobClient object pings the job client actor whether its done yet and if its not then it simply pings him if he is alive. The alive ping is: https://github.com/apache/flink/blob/62a777bc8ddfb4e34d7beaf7091a90b0bcc70c51/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java#L273
This ping times out and sends a poison pill to the job actor and that results in all the different disposal errors.
I have ran into some issues with futures before where they would get interrupted with shorter timeouts in a non-deterministic fashion. Ive debugged the issue somewhat and I think that was because of some very long GC pauses (or something similar). A illustration of how the timeouts are synchronized with the GC pauses: https://i.stack.imgur.com/IZBab.jpg. I think this could be the cause of this timeout aswell. This is my GC configuration:
"-XX:-UseParallelGC",
"-XX:-UseConcMarkSweepGC",
"-XX:+UseG1GC",
which according to most sources should result in very short GC pauses (less than a second). Anybody has any experience with getting very long GC pauses in flink? Could this be a issue somehow connected to HW? I am running the application on a EC2 AWS instance.