A clarification.. In 1.5 with custom heartbeats are there additional configurations we should be concerned about ?
On Fri, May 25, 2018 at 10:17 AM, Steven Wu <stevenz...@gmail.com> wrote: > Till, thanks for the follow-up. looking forward to 1.5 :) > > On Fri, May 25, 2018 at 2:11 AM, Till Rohrmann <trohrm...@apache.org> > wrote: > >> Hi Steven, >> >> we don't have `jobmanager.exit-on-fatal-akka-error` because then the JM >> would also be killed if a single TM gets quarantined. This is also not a >> desired behaviour. >> >> With Flink 1.5 the problem with quarantining should be gone since we >> don't rely anymore on Akka's death watch and instead use our own heartbeats. >> >> Cheers, >> Till >> >> On Mon, May 14, 2018 at 1:07 AM, Steven Wu <stevenz...@gmail.com> wrote: >> >>> Till, >>> >>> thanks for the clarification. yes, that situation is undesirable either. >>> >>> In our case, restarting jobmanager could also recover the job from akk >>> association lock-out. it was actually the issue (high GC pause) on >>> jobmanager side that caused the akka failure. >>> >>> do we have sth like "jobmanager.exit-on-fatal-akka-error: true"? does >>> it make sense to terminate jobmanager in this case? >>> >>> Thanks, >>> Steven >>> >>> On Sun, May 13, 2018 at 1:12 PM, Till Rohrmann <trohrm...@apache.org> >>> wrote: >>> >>>> Hi Steven, >>>> >>>> the reason why we did not turn on this feature per default was that in >>>> case of a true JM failure, all of the TMs will think that they got >>>> quarantined which triggers their shut down. Depending on how many container >>>> restarts you have left on Yarn, for example, this can lead to a situation >>>> where Flink is not able to recover the job even though it needed to only >>>> restart the JM container. >>>> >>>> Cheers, >>>> Till >>>> >>>> On Wed, Apr 25, 2018 at 10:39 PM, Steven Wu <stevenz...@gmail.com> >>>> wrote: >>>> >>>>> Till, >>>>> >>>>> We ran into the same issue. It started with high GC pause that caused >>>>> jobmanager to lose zk conn and leadership and caused jobmanager to >>>>> quarantine taskmanager in akka. Once quarantined, akka association btw >>>>> jobmanager and taskmanager is locked forever. >>>>> >>>>> Your suggestion of " taskmanager.exit-on-fatal-akka-error: true" >>>>> worked. taskmanager exited and replacement taskmanager joined the cluster >>>>> afterwards. I am wondering why is this not defaulted to "true". Any >>>>> downside? >>>>> >>>>> Thanks, >>>>> Steven >>>>> >>>>> On Sat, Feb 24, 2018 at 7:02 AM, ashish pok <ashish...@yahoo.com> >>>>> wrote: >>>>> >>>>>> @Jelmer, this is Till's las response on the issue. >>>>>> >>>>>> -- Ashish >>>>>> >>>>>> On Mon, Feb 5, 2018 at 5:56 AM, Till Rohrmann >>>>>> <trohrm...@apache.org> wrote: >>>>>> Hi, >>>>>> >>>>>> this sounds like a serious regression wrt Flink 1.3.2 and we should >>>>>> definitely find out what's causing this problem. Given from what I see in >>>>>> the logs, the following happens: >>>>>> >>>>>> For some time the JobManager seems to no longer receive heartbeats >>>>>> from the TaskManager. This could be, for example, due to long GC pauses >>>>>> or >>>>>> heavy load which starves the ActorSystem's threads which are responsible >>>>>> for sending the heartbeats. Due to this, the TM's ActorSystem is >>>>>> quarantined which effectively renders them useless because the JM will >>>>>> henceforth ignore all messages from these systems. The only way to >>>>>> resolve >>>>>> this problem is to restart the ActorSystem. By >>>>>> setting taskmanager.exit-on-fatal-akka-error to true in >>>>>> flink-conf.yaml, a quarantined TM will shut down. If you run the Flink >>>>>> cluster on Yarn, then a new substitute TM will be started if you have >>>>>> still >>>>>> some container restarts left. That way, the system should be able to >>>>>> recover. >>>>>> >>>>>> Additionally you could try to play around >>>>>> with akka.watch.heartbeat.interval and akka.watch.heartbeat.pause >>>>>> which control the heartbeat interval and the acceptable pause. By >>>>>> increasing the latter, the system should tolerate longer GC pauses and >>>>>> period of high load. >>>>>> >>>>>> However, this only addresses the symptoms of the problem and I'd like >>>>>> to find out what's causing the problem. In order to further debug the >>>>>> problem, it would be really helpful to obtain the logs of the JobManager >>>>>> and the TaskManagers on DEBUG log level and with >>>>>> taskmanager.debug.memory.startLogThread >>>>>> set to true. Additionally it would be interesting to see whats happening >>>>>> on >>>>>> the TaskManagers when you observe high load. So obtaining a profiler dump >>>>>> via VisualVM would be great. And last but not least, it also helps to >>>>>> learn >>>>>> more about the job you're running. What kind of connectors is it using? >>>>>> Are >>>>>> you using Flink's metric system? How is the Flink cluster deployed? Which >>>>>> other libraries are you using in your job? >>>>>> >>>>>> Thanks a lot for your help! >>>>>> >>>>>> Cheers, >>>>>> Till >>>>>> >>>>>> On Tue, Jan 30, 2018 at 8:59 PM, Cliff Resnick <cre...@gmail.com> >>>>>> wrote: >>>>>> >>>>>> I've seen a similar issue while running successive Flink SQL batches >>>>>> on 1.4. In my case, the Job Manager would fail with the log output about >>>>>> unreachability (with an additional statement about something going >>>>>> "horribly wrong"). Under workload pressure, I reverted to 1.3.2 where >>>>>> everything works perfectly, but we will try again soon on 1.4. When we >>>>>> do I >>>>>> will post the actual log output. >>>>>> >>>>>> This was on YARN in AWS, with akka.ask.timeout = 60s. >>>>>> >>>>>> On Wed, Jan 24, 2018 at 9:57 PM, Ashish Pokharel <ashish...@yahoo.com >>>>>> > wrote: >>>>>> >>>>>> I haven’t gotten much further with this. It doesn’t look like GC >>>>>> related - at least GC counters were not that atrocious. However, my main >>>>>> concern was once the load subsides why aren’t TM and JM connecting again? >>>>>> That doesn’t look normal. I could definitely tell JM was listening on the >>>>>> port and from logs it does appear TM is trying to message JM that is >>>>>> still >>>>>> alive. >>>>>> >>>>>> Thanks, Ashish >>>>>> >>>>>> On Jan 23, 2018, at 12:31 PM, Lasse Nedergaard < >>>>>> lassenederga...@gmail.com> wrote: >>>>>> >>>>>> Hi. >>>>>> >>>>>> Did you find a reason for the detaching ? >>>>>> I sometimes see the same on our system running Flink 1.4 on dc/os. I >>>>>> have enabled taskmanager.Debug.memory.start logthread for debugging. >>>>>> >>>>>> Med venlig hilsen / Best regards >>>>>> Lasse Nedergaard >>>>>> >>>>>> >>>>>> Den 20. jan. 2018 kl. 12.57 skrev Kien Truong < >>>>>> duckientru...@gmail.com>: >>>>>> >>>>>> Hi, >>>>>> >>>>>> You should enable and check your garbage collection log. >>>>>> >>>>>> We've encountered case where Task Manager disassociated due to long >>>>>> GC pause. >>>>>> >>>>>> >>>>>> Regards, >>>>>> >>>>>> Kien >>>>>> On 1/20/2018 1:27 AM, ashish pok wrote: >>>>>> >>>>>> Hi All, >>>>>> >>>>>> We have hit some load related issues and was wondering if any one has >>>>>> some suggestions. We are noticing task managers and job managers being >>>>>> detached from each other under load and never really sync up again. As a >>>>>> result, Flink session shows 0 slots available for processing. Even >>>>>> though, >>>>>> apps are configured to restart it isn't really helping as there are no >>>>>> slots available to run the apps. >>>>>> >>>>>> >>>>>> Here are excerpt from logs that seemed relevant. (I am trimming out >>>>>> rest of the logs for brevity) >>>>>> >>>>>> *Job Manager:* >>>>>> 2018-01-19 12:38:00,423 INFO org.apache.flink.runtime.jobma >>>>>> nager.JobManager - Starting JobManager (Version: 1.4.0, >>>>>> Rev:3a9d9f2, Date:06.12.2017 @ 11:08:40 UTC) >>>>>> >>>>>> 2018-01-19 12:38:00,792 INFO org.apache.flink.runtime.jobma >>>>>> nager.JobManager - Maximum heap size: 16384 MiBytes >>>>>> 2018-01-19 12:38:00,794 INFO org.apache.flink.runtime.jobma >>>>>> nager.JobManager - Hadoop version: 2.6.5 >>>>>> 2018-01-19 12:38:00,794 INFO org.apache.flink.runtime.jobma >>>>>> nager.JobManager - JVM Options: >>>>>> 2018-01-19 12:38:00,794 INFO org.apache.flink.runtime.jobma >>>>>> nager.JobManager - -Xms16384m >>>>>> 2018-01-19 12:38:00,794 INFO org.apache.flink.runtime.jobma >>>>>> nager.JobManager - -Xmx16384m >>>>>> 2018-01-19 12:38:00,795 INFO org.apache.flink.runtime.jobma >>>>>> nager.JobManager - -XX:+UseG1GC >>>>>> >>>>>> 2018-01-19 12:38:00,908 INFO org.apache.flink.configuration >>>>>> .GlobalConfiguration - Loading configuration property: >>>>>> jobmanager.rpc.port, 6123 >>>>>> 2018-01-19 12:38:00,908 INFO org.apache.flink.configuration >>>>>> .GlobalConfiguration - Loading configuration property: >>>>>> jobmanager.heap.mb, 16384 >>>>>> >>>>>> >>>>>> 2018-01-19 12:53:34,671 WARN akka.remote.RemoteWatcher >>>>>> - Detected unreachable: >>>>>> [akka.tcp://flink@<jm-host>:37 >>>>>> 840] >>>>>> 2018-01-19 12:53:34,676 INFO org.apache.flink.runtime.jobma >>>>>> nager.JobManager - Task manager >>>>>> akka.tcp://flink@<jm-host>:378 >>>>>> 40/user/taskmanager terminated. >>>>>> >>>>>> -- So once Flink session boots up, we are hitting it with pretty >>>>>> heavy load, which typically results in the WARN above >>>>>> >>>>>> *Task Manager:* >>>>>> 2018-01-19 12:38:01,002 INFO org.apache.flink.runtime.taskm >>>>>> anager.TaskManager - Starting TaskManager (Version: 1.4.0, >>>>>> Rev:3a9d9f2, Date:06.12.2017 @ 11:08:40 UTC) >>>>>> 2018-01-19 12:38:01,367 INFO org.apache.flink.runtime.taskm >>>>>> anager.TaskManager - Hadoop version: 2.6.5 >>>>>> 2018-01-19 12:38:01,367 INFO org.apache.flink.runtime.taskm >>>>>> anager.TaskManager - JVM Options: >>>>>> 2018-01-19 12:38:01,367 INFO org.apache.flink.runtime.taskm >>>>>> anager.TaskManager - -Xms16384M >>>>>> 2018-01-19 12:38:01,367 INFO org.apache.flink.runtime.taskm >>>>>> anager.TaskManager - -Xmx16384M >>>>>> 2018-01-19 12:38:01,367 INFO org.apache.flink.runtime.taskm >>>>>> anager.TaskManager - -XX:MaxDirectMemorySize=83886 07T >>>>>> 2018-01-19 12:38:01,367 INFO org.apache.flink.runtime.taskm >>>>>> anager.TaskManager - -XX:+UseG1GC >>>>>> >>>>>> 2018-01-19 12:38:01,392 INFO org.apache.flink.configuration >>>>>> .GlobalConfiguration - Loading configuration property: >>>>>> jobmanager.rpc.port, 6123 >>>>>> 2018-01-19 12:38:01,392 INFO org.apache.flink.configuration >>>>>> .GlobalConfiguration - Loading configuration property: >>>>>> jobmanager.heap.mb, 16384 >>>>>> >>>>>> >>>>>> 2018-01-19 12:54:48,626 WARN akka.remote.RemoteWatcher >>>>>> - Detected unreachable: >>>>>> [akka.tcp://flink@<jm-host>:61 >>>>>> 23] >>>>>> 2018-01-19 12:54:48,690 INFO akka.remote.Remoting >>>>>> - Quarantined address >>>>>> [akka.tcp://flink@<jm-host>:61 >>>>>> 23] is still unreachable or has not been restarted. Keeping it >>>>>> quarantined. >>>>>> 018-01-19 12:54:48,774 WARN akka.remote.Remoting >>>>>> - Tried to associate with unreachable remote >>>>>> address [akka.tcp://flink@<tm-host>:61 >>>>>> 23]. Address is now gated for 5000 ms, all messages to this address >>>>>> will be delivered to dead letters. Reason: [The remote system has a UID >>>>>> that has been quarantined. Association aborted.] >>>>>> 2018-01-19 12:54:48,833 WARN akka.remote.Remoting >>>>>> - Tried to associate with unreachable remote >>>>>> address [akka.tcp://flink@<tm-host>:61 >>>>>> 23]. Address is now gated for 5000 ms, all messages to this address >>>>>> will be delivered to dead letters. Reason: [The remote system has >>>>>> quarantined this system. No further associations to the remote system are >>>>>> possible until this system is restarted.] >>>>>> <bunch of ERRORs on operations not shutdown properly - assuming >>>>>> because JM is unreachable> >>>>>> >>>>>> 2018-01-19 12:56:51,244 INFO org.apache.flink.runtime.taskm >>>>>> anager.TaskManager - Trying to register at JobManager >>>>>> akka.tcp://flink@<jm-host>:612 >>>>>> 3/user/jobmanager (attempt 10, timeout: 30000 milliseconds) >>>>>> 2018-01-19 12:56:51,253 WARN akka.remote.Remoting >>>>>> - Tried to associate with unreachable remote >>>>>> address [akka.tcp://flink@<jm-host>:61 >>>>>> 23]. Address is now gated for 5000 ms, all messages to this address >>>>>> will be delivered to dead letters. Reason: [The remote system has >>>>>> quarantined this system. No further associations to the remote system are >>>>>> possible until this system is restarted.] >>>>>> >>>>>> So bottom line is, JM and TM couldn't communicate under load, which >>>>>> is obviously not good. I tried to bump up akka.tcp.timeout as well but it >>>>>> didnt help either. So my question here is after all processing is halted >>>>>> and there is no new data being picked up, shouldn't this environment >>>>>> self-heal? Any other things I can be looking at other than extending >>>>>> timeouts? >>>>>> >>>>>> Thanks, >>>>>> >>>>>> Ashish >>>>>> >>>>>> >>>>>> >>>>>> >>>>>> >>>>>> >>>>>> >>>>>> >>>>>> >>>>>> >>>>> >>>> >>> >> >
