(DEPRECATED) Apache Flink Mailing List archive.

JobManager not reachable

Classic

List

Threaded

13 messages Options

Matthias J. Sax-2

JobManager not reachable

Hi,

I was just playing arround with Flink. After submitting my job, it runs
for multiple minutes, until I get the following Exception in one if the
TaskManager logs and the job fails.

I have no clue what's going on...

-Matthias

> 18:43:23,567 WARN akka.remote.RemoteWatcher - Detected unreachable: [akka.tcp://flink@192.168.127.11:6123]
> 18:43:23,864 WARN akka.remote.ReliableDeliverySupervisor - Association with remote system [akka.tcp://flink@192.168.127.11:6123] has failed, address is now gated for [5000] ms. Reason is: [Disassociated].
> 18:43:23,866 INFO org.apache.flink.runtime.taskmanager.TaskManager - TaskManager akka://flink/user/taskmanager disconnects from JobManager akka.tcp://flink@192.168.127.11:6123/user/jobmanager: JobManager is no longer reachable
> 18:43:23,867 INFO org.apache.flink.runtime.taskmanager.TaskManager - Cancelling all computations and discarding all cached data.
> 18:43:23,870 INFO org.apache.flink.runtime.taskmanager.Task - Attempting to fail task externally Keyed Aggregation -> Sink: Unnamed (1/28)
> 18:43:23,870 INFO org.apache.flink.runtime.taskmanager.Task - Keyed Aggregation -> Sink: Unnamed (1/28) switched to FAILED with exception.
> java.lang.Exception: TaskManager akka://flink/user/taskmanager disconnects from JobManager akka.tcp://flink@192.168.127.11:6123/user/jobmanager: JobManager is no longer reachable
> at org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
> at org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
> at scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> at scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> at scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> at org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
> at scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> at scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> at scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> at org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
> at org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
> at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
> at org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> at org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> at akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
> at akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
> at akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
> 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)
> 18:43:23,879 INFO org.apache.flink.runtime.taskmanager.Task - Triggering cancellation of task code Keyed Aggregation -> Sink: Unnamed (1/28) (4b1f979a60c30f84ee60553730a4e99a).
> 18:43:23,880 INFO org.apache.flink.runtime.taskmanager.Task - Attempting to fail task externally Source: Custom Source -> Flat Map (1/28)
> 18:43:23,880 INFO org.apache.flink.runtime.taskmanager.Task - Source: Custom Source -> Flat Map (1/28) switched to FAILED with exception.
> java.lang.Exception: TaskManager akka://flink/user/taskmanager disconnects from JobManager akka.tcp://flink@192.168.127.11:6123/user/jobmanager: JobManager is no longer reachable
> at org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
> at org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
> at scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> at scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> at scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> at org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
> at scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> at scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> at scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> at org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
> at org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
> at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
> at org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> at org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> at akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
> at akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
> at akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
> 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)

signature.asc (836 bytes) Download Attachment

Stephan Ewen-2

Re: JobManager not reachable

Can you see is there is anything unusual in the JobManager logs?
Am 11.10.2015 18:56 schrieb "Matthias J. Sax" <[hidden email]>:

> Hi,
>
> I was just playing arround with Flink. After submitting my job, it runs
> for multiple minutes, until I get the following Exception in one if the
> TaskManager logs and the job fails.
>
> I have no clue what's going on...
>
> -Matthias
>
>
> > 18:43:23,567 WARN akka.remote.RemoteWatcher
> - Detected unreachable: [akka.tcp://flink@192.168.127.11:6123]
> > 18:43:23,864 WARN akka.remote.ReliableDeliverySupervisor
> - Association with remote system [akka.tcp://
> flink@192.168.127.11:6123] has failed, address is now gated for [5000]
> ms. Reason is: [Disassociated].
> > 18:43:23,866 INFO org.apache.flink.runtime.taskmanager.TaskManager
> - TaskManager akka://flink/user/taskmanager disconnects from
> JobManager akka.tcp://flink@192.168.127.11:6123/user/jobmanager:
> JobManager is no longer reachable
> > 18:43:23,867 INFO org.apache.flink.runtime.taskmanager.TaskManager
> - Cancelling all computations and discarding all cached data.
> > 18:43:23,870 INFO org.apache.flink.runtime.taskmanager.Task
> - Attempting to fail task externally Keyed Aggregation -> Sink:
> Unnamed (1/28)
> > 18:43:23,870 INFO org.apache.flink.runtime.taskmanager.Task
> - Keyed Aggregation -> Sink: Unnamed (1/28) switched to FAILED
> with exception.
> > java.lang.Exception: TaskManager akka://flink/user/taskmanager
> disconnects from JobManager akka.tcp://
> flink@192.168.127.11:6123/user/jobmanager: JobManager is no longer
> reachable
> > at
> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
> > at
> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
> > at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> > at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> > at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> > at
> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
> > at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> > at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> > at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> > at
> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
> > at
> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
> > at
> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
> > at
> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
> > at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> > at
> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
> > at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> > at
> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
> > at akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
> > at akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
> > at akka.actor.ActorCell.invoke(ActorCell.scala:486)
> > at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
> > at akka.dispatch.Mailbox.run(Mailbox.scala:221)
> > at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
> > 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)
> > 18:43:23,879 INFO org.apache.flink.runtime.taskmanager.Task
> - Triggering cancellation of task code Keyed Aggregation -> Sink:
> Unnamed (1/28) (4b1f979a60c30f84ee60553730a4e99a).
> > 18:43:23,880 INFO org.apache.flink.runtime.taskmanager.Task
> - Attempting to fail task externally Source: Custom Source -> Flat
> Map (1/28)
> > 18:43:23,880 INFO org.apache.flink.runtime.taskmanager.Task
> - Source: Custom Source -> Flat Map (1/28) switched to FAILED with
> exception.
> > java.lang.Exception: TaskManager akka://flink/user/taskmanager
> disconnects from JobManager akka.tcp://
> flink@192.168.127.11:6123/user/jobmanager: JobManager is no longer
> reachable
> > at
> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
> > at
> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
> > at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> > at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> > at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> > at
> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
> > at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> > at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> > at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> > at
> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
> > at
> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
> > at
> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
> > at
> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
> > at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> > at
> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
> > at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> > at
> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
> > at akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
> > at akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
> > at akka.actor.ActorCell.invoke(ActorCell.scala:486)
> > at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
> > at akka.dispatch.Mailbox.run(Mailbox.scala:221)
> > at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
> > 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)
>
>

Ufuk Celebi-2

Re: JobManager not reachable

> On 11 Oct 2015, at 23:54, Stephan Ewen <[hidden email]> wrote:
>
> Can you see is there is anything unusual in the JobManager logs?

Ping. :)

Matthias J. Sax-2

Re: JobManager not reachable

In reply to this post by Stephan Ewen-2

No. See log below.

Btw: the job is not cleaned up properly. Some task remain in state
"Canceling".

The program I execute is "Streaming WordCount" example with my own
source function. This custom source (see below), reads a local (small)
file, bufferes each line in an internal buffer, and emits the buffered
lines over and over again (for 5 minutes). (A copy of the file is
located on every worker machine of course.) This is of course no common
patttern; I just do this, because I have only a small file and want the
job to run longer.

As the job is running for multiple minutes until it fails, I assume that
this uncommon pattern should actually not cauase any problem. I just
report it, to make sure it is not the problem (hope you can confirm).

-Matthias

> private static final class SimpleFileSource implements ParallelSourceFunction<String> {
> private static final long serialVersionUID = -5727198990308179551L;
>
> volatile boolean running = true;
> @Override
> public void run(
> org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<String> ctx)
> throws Exception {
>
> BufferedReader reader = new BufferedReader(new FileReader(
> "/data/mjsax/hamlet.txt"));
> ArrayList<String> buffer = new ArrayList<String>();
> int counter = 0;
> String line;
>
> long start = System.nanoTime();
> while (running && (line = reader.readLine()) != null) {
> buffer.add(line);
> ctx.collect(line);
> }
> reader.close();
>
> while (running && (System.nanoTime() - start) < 5L * 60 * 1000 * 1000 * 1000) {
> ctx.collect(buffer.get(counter));
> counter = ++counter % buffer.size();
> }
> }
>
> @Override
> public void cancel() {
> running = false;
> }
> }

JobManager log:

> 22:19:16,618 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Keyed Aggregation -> Sink: Unnamed (e18d7fb12ea2b5b88dda07e79a6c62ba) switched from CANCELING to FAILED
> 22:23:01,084 INFO org.apache.flink.runtime.instance.InstanceManager - Unregistered task manager akka.tcp://flink@192.168.127.62:35242/user/taskmanager. Number of registered task managers 18. Number of available slots 216.

On 10/11/2015 11:54 PM, Stephan Ewen wrote:

> Can you see is there is anything unusual in the JobManager logs?
> Am 11.10.2015 18:56 schrieb "Matthias J. Sax" <[hidden email]>:
>
>> Hi,
>>
>> I was just playing arround with Flink. After submitting my job, it runs
>> for multiple minutes, until I get the following Exception in one if the
>> TaskManager logs and the job fails.
>>
>> I have no clue what's going on...
>>
>> -Matthias
>>
>>
>>> 18:43:23,567 WARN akka.remote.RemoteWatcher
>> - Detected unreachable: [akka.tcp://flink@192.168.127.11:6123]
>>> 18:43:23,864 WARN akka.remote.ReliableDeliverySupervisor
>> - Association with remote system [akka.tcp://
>> flink@192.168.127.11:6123] has failed, address is now gated for [5000]
>> ms. Reason is: [Disassociated].
>>> 18:43:23,866 INFO org.apache.flink.runtime.taskmanager.TaskManager
>> - TaskManager akka://flink/user/taskmanager disconnects from
>> JobManager akka.tcp://flink@192.168.127.11:6123/user/jobmanager:
>> JobManager is no longer reachable
>>> 18:43:23,867 INFO org.apache.flink.runtime.taskmanager.TaskManager
>> - Cancelling all computations and discarding all cached data.
>>> 18:43:23,870 INFO org.apache.flink.runtime.taskmanager.Task
>> - Attempting to fail task externally Keyed Aggregation -> Sink:
>> Unnamed (1/28)
>>> 18:43:23,870 INFO org.apache.flink.runtime.taskmanager.Task
>> - Keyed Aggregation -> Sink: Unnamed (1/28) switched to FAILED
>> with exception.
>>> java.lang.Exception: TaskManager akka://flink/user/taskmanager
>> disconnects from JobManager akka.tcp://
>> flink@192.168.127.11:6123/user/jobmanager: JobManager is no longer
>> reachable
>>> at
>> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
>>> at
>> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
>>> at
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
>>> at
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
>>> at
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
>>> at
>> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
>>> at
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
>>> at
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
>>> at
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
>>> at
>> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
>>> at
>> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
>>> at
>> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
>>> at
>> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
>>> at
>> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
>>> at
>> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
>>> at akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
>>> at akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
>>> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
>>> 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)
>>> 18:43:23,879 INFO org.apache.flink.runtime.taskmanager.Task
>> - Triggering cancellation of task code Keyed Aggregation -> Sink:
>> Unnamed (1/28) (4b1f979a60c30f84ee60553730a4e99a).
>>> 18:43:23,880 INFO org.apache.flink.runtime.taskmanager.Task
>> - Attempting to fail task externally Source: Custom Source -> Flat
>> Map (1/28)
>>> 18:43:23,880 INFO org.apache.flink.runtime.taskmanager.Task
>> - Source: Custom Source -> Flat Map (1/28) switched to FAILED with
>> exception.
>>> java.lang.Exception: TaskManager akka://flink/user/taskmanager
>> disconnects from JobManager akka.tcp://
>> flink@192.168.127.11:6123/user/jobmanager: JobManager is no longer
>> reachable
>>> at
>> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
>>> at
>> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
>>> at
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
>>> at
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
>>> at
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
>>> at
>> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
>>> at
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
>>> at
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
>>> at
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
>>> at
>> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
>>> at
>> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
>>> at
>> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
>>> at
>> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
>>> at
>> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
>>> at
>> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
>>> at akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
>>> at akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
>>> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
>>> 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)
>>
>>
>

signature.asc (836 bytes) Download Attachment

Matthias J. Sax-2

Re: JobManager not reachable

One thing I forgot the add. I also have a Storm-WordCount job (build via
FlinkTopologyBuilder) that uses the same
"buffer-file-and-emit-over-and-over-again-pattern" in a spout. This job
run just fine and stops regularly after 5 minutes.

-Matthias

On 10/14/2015 10:42 PM, Matthias J. Sax wrote:

> No. See log below.
>
> Btw: the job is not cleaned up properly. Some task remain in state
> "Canceling".
>
> The program I execute is "Streaming WordCount" example with my own
> source function. This custom source (see below), reads a local (small)
> file, bufferes each line in an internal buffer, and emits the buffered
> lines over and over again (for 5 minutes). (A copy of the file is
> located on every worker machine of course.) This is of course no common
> patttern; I just do this, because I have only a small file and want the
> job to run longer.
>
> As the job is running for multiple minutes until it fails, I assume that
> this uncommon pattern should actually not cauase any problem. I just
> report it, to make sure it is not the problem (hope you can confirm).
>
> -Matthias
>
>
>> private static final class SimpleFileSource implements ParallelSourceFunction<String> {
>> private static final long serialVersionUID = -5727198990308179551L;
>>
>> volatile boolean running = true;
>> @Override
>> public void run(
>> org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<String> ctx)
>> throws Exception {
>>
>> BufferedReader reader = new BufferedReader(new FileReader(
>> "/data/mjsax/hamlet.txt"));
>> ArrayList<String> buffer = new ArrayList<String>();
>> int counter = 0;
>> String line;
>>
>> long start = System.nanoTime();
>> while (running && (line = reader.readLine()) != null) {
>> buffer.add(line);
>> ctx.collect(line);
>> }
>> reader.close();
>>
>> while (running && (System.nanoTime() - start) < 5L * 60 * 1000 * 1000 * 1000) {
>> ctx.collect(buffer.get(counter));
>> counter = ++counter % buffer.size();
>> }
>> }
>>
>> @Override
>> public void cancel() {
>> running = false;
>> }
>> }
>
>
> JobManager log:
>
>> 22:19:16,618 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Keyed Aggregation -> Sink: Unnamed switched to FAILED with
>>> exception.
>>>> java.lang.Exception: TaskManager akka://flink/user/taskmanager
>>> disconnects from JobManager akka.tcp://
>>> flink@192.168.127.11:6123/user/jobmanager: JobManager is no longer
>>> reachable
>>>> at
>>> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
>>>> at
>>> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
>>>> at
>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
>>>> at
>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
>>>> at
>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
>>>> at
>>> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
>>>> at
>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
>>>> at
>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
>>>> at
>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
>>>> at
>>> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
>>>> at
>>> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
>>>> at
>>> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
>>>> at
>>> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
>>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
>>>> at
>>> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
>>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
>>>> at
>>> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
>>>> at akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
>>>> at akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
>>>> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
>>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
>>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
>>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
>>>> 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)
>>>
>>>
>>
>

signature.asc (836 bytes) Download Attachment

Stephan Ewen

Re: JobManager not reachable

From what the logs show, the TaskManager does not send pings any more for a
long time and is then considered failed and the tasks running on that
TaskManager are considered failed as well. So far, nothing unusual...

Question is, why is it considered failed? Is this a reproducible problem?
Or a one time occurrence? Does the user code accumulate so many objects to
put very heavy pressure on the GC?

Stephan

On Wed, Oct 14, 2015 at 10:47 PM, Matthias J. Sax <[hidden email]> wrote:

> One thing I forgot the add. I also have a Storm-WordCount job (build via
> FlinkTopologyBuilder) that uses the same
> "buffer-file-and-emit-over-and-over-again-pattern" in a spout. This job
> run just fine and stops regularly after 5 minutes.
>
> -Matthias
>
>
> On 10/14/2015 10:42 PM, Matthias J. Sax wrote:
> > No. See log below.
> >
> > Btw: the job is not cleaned up properly. Some task remain in state
> > "Canceling".
> >
> > The program I execute is "Streaming WordCount" example with my own
> > source function. This custom source (see below), reads a local (small)
> > file, bufferes each line in an internal buffer, and emits the buffered
> > lines over and over again (for 5 minutes). (A copy of the file is
> > located on every worker machine of course.) This is of course no common
> > patttern; I just do this, because I have only a small file and want the
> > job to run longer.
> >
> > As the job is running for multiple minutes until it fails, I assume that
> > this uncommon pattern should actually not cauase any problem. I just
> > report it, to make sure it is not the problem (hope you can confirm).
> >
> > -Matthias
> >
> >
> >> private static final class SimpleFileSource implements
> ParallelSourceFunction<String> {
> >> private static final long serialVersionUID =
> -5727198990308179551L;
> >>
> >> volatile boolean running = true;
> >> @Override
> >> public void run(
> >>
> org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<String>
> ctx)
> >> throws Exception {
> >>
> >> BufferedReader reader = new BufferedReader(new
> FileReader(
> >> "/data/mjsax/hamlet.txt"));
> >> ArrayList<String> buffer = new ArrayList<String>();
> >> int counter = 0;
> >> String line;
> >>
> >> long start = System.nanoTime();
> >> while (running && (line = reader.readLine()) !=
> null) {
> >> buffer.add(line);
> >> ctx.collect(line);
> >> }
> >> reader.close();
> >>
> >> while (running && (System.nanoTime() - start) < 5L
> * 60 * 1000 * 1000 * 1000) {
> >> ctx.collect(buffer.get(counter));
> >> counter = ++counter % buffer.size();
> >> }
> >> }
> >>
> >> @Override
> >> public void cancel() {
> >> running = false;
> >> }
> >> }
> >
> >
> > JobManager log:
> >
> >> 22:19:16,618 INFO
> org.apache.flink.runtime.executiongraph.ExecutionGraph - Keyed
> Aggregation -> Sink: Unnamed switched to FAILED
> with
> >>> exception.
> >>>> java.lang.Exception: TaskManager akka://flink/user/taskmanager
> >>> disconnects from JobManager akka.tcp://
> >>> flink@192.168.127.11:6123/user/jobmanager: JobManager is no longer
> >>> reachable
> >>>> at
> >>>
> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
> >>>> at
> >>>
> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
> >>>> at
> >>>
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> >>>> at
> >>>
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> >>>> at
> >>>
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> >>>> at
> >>>
> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
> >>>> at
> >>>
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> >>>> at
> >>>
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> >>>> at
> >>>
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> >>>> at
> >>>
> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
> >>>> at
> >>>
> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
> >>>> at
> >>> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
> >>>> at
> >>>
> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
> >>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> >>>> at
> >>>
> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
> >>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> >>>> at
> >>>
> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
> >>>> at
> akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
> >>>> at
> akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
> >>>> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
> >>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
> >>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
> >>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
> >>>> 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)
> >>>
> >>>
> >>
> >
>
>

Matthias J. Sax-2

Re: JobManager not reachable

The problem is reproducible (it happens on each run).

I doubt that GC is an issue here (at least from an UDF point of view),
because I read the file once and keep a String object for each line.
This objects are kept to the very end; the UDF does not release them
until it returns from "run()" method.

The file is also small (10K) so main memory consumption should be a
problem either.

Of course, the UDF emits on a very high rate and I am not sure if Flink
creates a lots of short living object (Streaming does not reuse object
yet...) such that GC might become an issue here?

-Matthias

On 10/15/2015 11:44 AM, Stephan Ewen wrote:

> From what the logs show, the TaskManager does not send pings any more for a
> long time and is then considered failed and the tasks running on that
> TaskManager are considered failed as well. So far, nothing unusual...
>
> Question is, why is it considered failed? Is this a reproducible problem?
> Or a one time occurrence? Does the user code accumulate so many objects to
> put very heavy pressure on the GC?
>
> Stephan
>
>
> On Wed, Oct 14, 2015 at 10:47 PM, Matthias J. Sax <[hidden email]> wrote:
>
>> One thing I forgot the add. I also have a Storm-WordCount job (build via
>> FlinkTopologyBuilder) that uses the same
>> "buffer-file-and-emit-over-and-over-again-pattern" in a spout. This job
>> run just fine and stops regularly after 5 minutes.
>>
>> -Matthias
>>
>>
>> On 10/14/2015 10:42 PM, Matthias J. Sax wrote:
>>> No. See log below.
>>>
>>> Btw: the job is not cleaned up properly. Some task remain in state
>>> "Canceling".
>>>
>>> The program I execute is "Streaming WordCount" example with my own
>>> source function. This custom source (see below), reads a local (small)
>>> file, bufferes each line in an internal buffer, and emits the buffered
>>> lines over and over again (for 5 minutes). (A copy of the file is
>>> located on every worker machine of course.) This is of course no common
>>> patttern; I just do this, because I have only a small file and want the
>>> job to run longer.
>>>
>>> As the job is running for multiple minutes until it fails, I assume that
>>> this uncommon pattern should actually not cauase any problem. I just
>>> report it, to make sure it is not the problem (hope you can confirm).
>>>
>>> -Matthias
>>>
>>>
>>>> private static final class SimpleFileSource implements
>> ParallelSourceFunction<String> {
>>>> private static final long serialVersionUID =
>> -5727198990308179551L;
>>>>
>>>> volatile boolean running = true;
>>>> @Override
>>>> public void run(
>>>>
>> org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<String>
>> ctx)
>>>> throws Exception {
>>>>
>>>> BufferedReader reader = new BufferedReader(new
>> FileReader(
>>>> "/data/mjsax/hamlet.txt"));
>>>> ArrayList<String> buffer = new ArrayList<String>();
>>>> int counter = 0;
>>>> String line;
>>>>
>>>> long start = System.nanoTime();
>>>> while (running && (line = reader.readLine()) !=
>> null) {
>>>> buffer.add(line);
>>>> ctx.collect(line);
>>>> }
>>>> reader.close();
>>>>
>>>> while (running && (System.nanoTime() - start) < 5L
>> * 60 * 1000 * 1000 * 1000) {
>>>> ctx.collect(buffer.get(counter));
>>>> counter = ++counter % buffer.size();
>>>> }
>>>> }
>>>>
>>>> @Override
>>>> public void cancel() {
>>>> running = false;
>>>> }
>>>> }
>>>
>>>
>>> JobManager log:
>>>
>>>> 22:19:16,618 INFO
>> org.apache.flink.runtime.executiongraph.ExecutionGraph - Keyed
>> Aggregation -> Sink: Unnamed switched to FAILED
>> with
>>>>> exception.
>>>>>> java.lang.Exception: TaskManager akka://flink/user/taskmanager
>>>>> disconnects from JobManager akka.tcp://
>>>>> flink@192.168.127.11:6123/user/jobmanager: JobManager is no longer
>>>>> reachable
>>>>>> at
>>>>>
>> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
>>>>>> at
>>>>>
>> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
>>>>>> at
>>>>>
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
>>>>>> at
>>>>>
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
>>>>>> at
>>>>>
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
>>>>>> at
>>>>>
>> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
>>>>>> at
>>>>>
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
>>>>>> at
>>>>>
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
>>>>>> at
>>>>>
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
>>>>>> at
>>>>>
>> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
>>>>>> at
>>>>>
>> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
>>>>>> at
>>>>> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
>>>>>> at
>>>>>
>> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
>>>>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
>>>>>> at
>>>>>
>> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
>>>>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
>>>>>> at
>>>>>
>> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
>>>>>> at
>> akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
>>>>>> at
>> akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
>>>>>> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
>>>>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
>>>>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
>>>>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
>>>>>> 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)
>>>>>
>>>>>
>>>>
>>>
>>
>>
>

signature.asc (836 bytes) Download Attachment

Till Rohrmann

Re: JobManager not reachable

To verify wether GC is a problem you can enable logging of memory usage of
the JVM via taskmanager.debug.memory.startLogThread: true. The interval of
the logging is configured via taskmanager.debug.memory.logIntervalMs.

On Thu, Oct 15, 2015 at 12:00 PM, Matthias J. Sax <[hidden email]> wrote:

> The problem is reproducible (it happens on each run).
>
> I doubt that GC is an issue here (at least from an UDF point of view),
> because I read the file once and keep a String object for each line.
> This objects are kept to the very end; the UDF does not release them
> until it returns from "run()" method.
>
> The file is also small (10K) so main memory consumption should be a
> problem either.
>
> Of course, the UDF emits on a very high rate and I am not sure if Flink
> creates a lots of short living object (Streaming does not reuse object
> yet...) such that GC might become an issue here?
>
>
> -Matthias
>
>
>
> On 10/15/2015 11:44 AM, Stephan Ewen wrote:
> > From what the logs show, the TaskManager does not send pings any more
> for a
> > long time and is then considered failed and the tasks running on that
> > TaskManager are considered failed as well. So far, nothing unusual...
> >
> > Question is, why is it considered failed? Is this a reproducible problem?
> > Or a one time occurrence? Does the user code accumulate so many objects
> to
> > put very heavy pressure on the GC?
> >
> > Stephan
> >
> >
> > On Wed, Oct 14, 2015 at 10:47 PM, Matthias J. Sax <[hidden email]>
> wrote:
> >
> >> One thing I forgot the add. I also have a Storm-WordCount job (build via
> >> FlinkTopologyBuilder) that uses the same
> >> "buffer-file-and-emit-over-and-over-again-pattern" in a spout. This job
> >> run just fine and stops regularly after 5 minutes.
> >>
> >> -Matthias
> >>
> >>
> >> On 10/14/2015 10:42 PM, Matthias J. Sax wrote:
> >>> No. See log below.
> >>>
> >>> Btw: the job is not cleaned up properly. Some task remain in state
> >>> "Canceling".
> >>>
> >>> The program I execute is "Streaming WordCount" example with my own
> >>> source function. This custom source (see below), reads a local (small)
> >>> file, bufferes each line in an internal buffer, and emits the buffered
> >>> lines over and over again (for 5 minutes). (A copy of the file is
> >>> located on every worker machine of course.) This is of course no common
> >>> patttern; I just do this, because I have only a small file and want the
> >>> job to run longer.
> >>>
> >>> As the job is running for multiple minutes until it fails, I assume
> that
> >>> this uncommon pattern should actually not cauase any problem. I just
> >>> report it, to make sure it is not the problem (hope you can confirm).
> >>>
> >>> -Matthias
> >>>
> >>>
> >>>> private static final class SimpleFileSource implements
> >> ParallelSourceFunction<String> {
> >>>> private static final long serialVersionUID =
> >> -5727198990308179551L;
> >>>>
> >>>> volatile boolean running = true;
> >>>> @Override
> >>>> public void run(
> >>>>
> >>
> org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<String>
> >> ctx)
> >>>> throws Exception {
> >>>>
> >>>> BufferedReader reader = new BufferedReader(new
> >> FileReader(
> >>>> "/data/mjsax/hamlet.txt"));
> >>>> ArrayList<String> buffer = new
> ArrayList<String>();
> >>>> int counter = 0;
> >>>> String line;
> >>>>
> >>>> long start = System.nanoTime();
> >>>> while (running && (line = reader.readLine()) !=
> >> null) {
> >>>> buffer.add(line);
> >>>> ctx.collect(line);
> >>>> }
> >>>> reader.close();
> >>>>
> >>>> while (running && (System.nanoTime() - start) <
> 5L
> >> * 60 * 1000 * 1000 * 1000) {
> >>>> ctx.collect(buffer.get(counter));
> >>>> counter = ++counter % buffer.size();
> >>>> }
> >>>> }
> >>>>
> >>>> @Override
> >>>> public void cancel() {
> >>>> running = false;
> >>>> }
> >>>> }
> >>>
> >>>
> >>> JobManager log:
> >>>
> >>>> 22:19:16,618 INFO
> >> org.apache.flink.runtime.executiongraph.ExecutionGraph - Keyed
> >> Aggregation -> Sink: Unnamed switched to
> FAILED
> >> with
> >>>>> exception.
> >>>>>> java.lang.Exception: TaskManager akka://flink/user/taskmanager
> >>>>> disconnects from JobManager akka.tcp://
> >>>>> flink@192.168.127.11:6123/user/jobmanager: JobManager is no longer
> >>>>> reachable
> >>>>>> at
> >>>>>
> >>
> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
> >>>>>> at
> >>>>>
> >>
> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
> >>>>>> at
> >>>>>
> >>
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> >>>>>> at
> >>>>>
> >>
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> >>>>>> at
> >>>>>
> >>
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> >>>>>> at
> >>>>>
> >>
> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
> >>>>>> at
> >>>>>
> >>
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> >>>>>> at
> >>>>>
> >>
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> >>>>>> at
> >>>>>
> >>
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> >>>>>> at
> >>>>>
> >> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
> >>>>>> at
> >>>>>
> >> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
> >>>>>> at
> >>>>> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
> >>>>>> at
> >>>>>
> >>
> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
> >>>>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> >>>>>> at
> >>>>>
> >>
> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
> >>>>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> >>>>>> at
> >>>>>
> >>
> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
> >>>>>> at
> >> akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
> >>>>>> at
> >> akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
> >>>>>> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
> >>>>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
> >>>>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
> >>>>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
> >>>>>> 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)
> >>>>>
> >>>>>
> >>>>
> >>>
> >>
> >>
> >
>
>

Stephan Ewen

Re: JobManager not reachable

Does not quite sound like GC is an issue.

Hmmm, what else can make the failure detector kick in unexpectedly?

On Thu, Oct 15, 2015 at 12:05 PM, Till Rohrmann <[hidden email]>
wrote:

> To verify wether GC is a problem you can enable logging of memory usage of
> the JVM via taskmanager.debug.memory.startLogThread: true. The interval of
> the logging is configured via taskmanager.debug.memory.logIntervalMs.
>
>
> On Thu, Oct 15, 2015 at 12:00 PM, Matthias J. Sax <[hidden email]>
> wrote:
>
> > The problem is reproducible (it happens on each run).
> >
> > I doubt that GC is an issue here (at least from an UDF point of view),
> > because I read the file once and keep a String object for each line.
> > This objects are kept to the very end; the UDF does not release them
> > until it returns from "run()" method.
> >
> > The file is also small (10K) so main memory consumption should be a
> > problem either.
> >
> > Of course, the UDF emits on a very high rate and I am not sure if Flink
> > creates a lots of short living object (Streaming does not reuse object
> > yet...) such that GC might become an issue here?
> >
> >
> > -Matthias
> >
> >
> >
> > On 10/15/2015 11:44 AM, Stephan Ewen wrote:
> > > From what the logs show, the TaskManager does not send pings any more
> > for a
> > > long time and is then considered failed and the tasks running on that
> > > TaskManager are considered failed as well. So far, nothing unusual...
> > >
> > > Question is, why is it considered failed? Is this a reproducible
> problem?
> > > Or a one time occurrence? Does the user code accumulate so many objects
> > to
> > > put very heavy pressure on the GC?
> > >
> > > Stephan
> > >
> > >
> > > On Wed, Oct 14, 2015 at 10:47 PM, Matthias J. Sax <[hidden email]>
> > wrote:
> > >
> > >> One thing I forgot the add. I also have a Storm-WordCount job (build
> via
> > >> FlinkTopologyBuilder) that uses the same
> > >> "buffer-file-and-emit-over-and-over-again-pattern" in a spout. This
> job
> > >> run just fine and stops regularly after 5 minutes.
> > >>
> > >> -Matthias
> > >>
> > >>
> > >> On 10/14/2015 10:42 PM, Matthias J. Sax wrote:
> > >>> No. See log below.
> > >>>
> > >>> Btw: the job is not cleaned up properly. Some task remain in state
> > >>> "Canceling".
> > >>>
> > >>> The program I execute is "Streaming WordCount" example with my own
> > >>> source function. This custom source (see below), reads a local
> (small)
> > >>> file, bufferes each line in an internal buffer, and emits the
> buffered
> > >>> lines over and over again (for 5 minutes). (A copy of the file is
> > >>> located on every worker machine of course.) This is of course no
> common
> > >>> patttern; I just do this, because I have only a small file and want
> the
> > >>> job to run longer.
> > >>>
> > >>> As the job is running for multiple minutes until it fails, I assume
> > that
> > >>> this uncommon pattern should actually not cauase any problem. I just
> > >>> report it, to make sure it is not the problem (hope you can confirm).
> > >>>
> > >>> -Matthias
> > >>>
> > >>>
> > >>>> private static final class SimpleFileSource implements
> > >> ParallelSourceFunction<String> {
> > >>>> private static final long serialVersionUID =
> > >> -5727198990308179551L;
> > >>>>
> > >>>> volatile boolean running = true;
> > >>>> @Override
> > >>>> public void run(
> > >>>>
> > >>
> >
> org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<String>
> > >> ctx)
> > >>>> throws Exception {
> > >>>>
> > >>>> BufferedReader reader = new BufferedReader(new
> > >> FileReader(
> > >>>> "/data/mjsax/hamlet.txt"));
> > >>>> ArrayList<String> buffer = new
> > ArrayList<String>();
> > >>>> int counter = 0;
> > >>>> String line;
> > >>>>
> > >>>> long start = System.nanoTime();
> > >>>> while (running && (line = reader.readLine()) !=
> > >> null) {
> > >>>> buffer.add(line);
> > >>>> ctx.collect(line);
> > >>>> }
> > >>>> reader.close();
> > >>>>
> > >>>> while (running && (System.nanoTime() - start) <
> > 5L
> > >> * 60 * 1000 * 1000 * 1000) {
> > >>>> ctx.collect(buffer.get(counter));
> > >>>> counter = ++counter % buffer.size();
> > >>>> }
> > >>>> }
> > >>>>
> > >>>> @Override
> > >>>> public void cancel() {
> > >>>> running = false;
> > >>>> }
> > >>>> }
> > >>>
> > >>>
> > >>> JobManager log:
> > >>>
> > >>>> 22:19:16,618 INFO
> > >> org.apache.flink.runtime.executiongraph.ExecutionGraph - Keyed
> > >> Aggregation -> Sink: Unnamed switched to
> > FAILED
> > >> with
> > >>>>> exception.
> > >>>>>> java.lang.Exception: TaskManager akka://flink/user/taskmanager
> > >>>>> disconnects from JobManager akka.tcp://
> > >>>>> flink@192.168.127.11:6123/user/jobmanager: JobManager is no longer
> > >>>>> reachable
> > >>>>>> at
> > >>>>>
> > >>
> >
> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
> > >>>>>> at
> > >>>>>
> > >>
> >
> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
> > >>>>>> at
> > >>>>>
> > >>
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> > >>>>>> at
> > >>>>>
> > >>
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> > >>>>>> at
> > >>>>>
> > >>
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> > >>>>>> at
> > >>>>>
> > >>
> >
> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
> > >>>>>> at
> > >>>>>
> > >>
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> > >>>>>> at
> > >>>>>
> > >>
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> > >>>>>> at
> > >>>>>
> > >>
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> > >>>>>> at
> > >>>>>
> > >>
> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
> > >>>>>> at
> > >>>>>
> > >>
> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
> > >>>>>> at
> > >>>>> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
> > >>>>>> at
> > >>>>>
> > >>
> >
> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
> > >>>>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> > >>>>>> at
> > >>>>>
> > >>
> >
> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
> > >>>>>> at
> akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> > >>>>>> at
> > >>>>>
> > >>
> >
> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
> > >>>>>> at
> > >> akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
> > >>>>>> at
> > >> akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
> > >>>>>> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
> > >>>>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
> > >>>>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
> > >>>>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
> > >>>>>> 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)
> > >>>>>
> > >>>>>
> > >>>>
> > >>>
> > >>
> > >>
> > >
> >
> >
>

Robert Metzger

Re: JobManager not reachable

Can you start flink with logging level DEBUG ?
Then we can see from the TaskManager logs when the TM became inactive.
Maybe an Akka message is causing the actor to block?

You can also monitor the GC from the TaskManager view in the web interface
(for example by looking at the total time spend for GCing).

On Thu, Oct 15, 2015 at 12:09 PM, Stephan Ewen <[hidden email]> wrote:

> Does not quite sound like GC is an issue.
>
> Hmmm, what else can make the failure detector kick in unexpectedly?
>
> On Thu, Oct 15, 2015 at 12:05 PM, Till Rohrmann <[hidden email]>
> wrote:
>
> > To verify wether GC is a problem you can enable logging of memory usage
> of
> > the JVM via taskmanager.debug.memory.startLogThread: true. The interval
> of
> > the logging is configured via taskmanager.debug.memory.logIntervalMs.
> >
> >
> > On Thu, Oct 15, 2015 at 12:00 PM, Matthias J. Sax <[hidden email]>
> > wrote:
> >
> > > The problem is reproducible (it happens on each run).
> > >
> > > I doubt that GC is an issue here (at least from an UDF point of view),
> > > because I read the file once and keep a String object for each line.
> > > This objects are kept to the very end; the UDF does not release them
> > > until it returns from "run()" method.
> > >
> > > The file is also small (10K) so main memory consumption should be a
> > > problem either.
> > >
> > > Of course, the UDF emits on a very high rate and I am not sure if Flink
> > > creates a lots of short living object (Streaming does not reuse object
> > > yet...) such that GC might become an issue here?
> > >
> > >
> > > -Matthias
> > >
> > >
> > >
> > > On 10/15/2015 11:44 AM, Stephan Ewen wrote:
> > > > From what the logs show, the TaskManager does not send pings any more
> > > for a
> > > > long time and is then considered failed and the tasks running on that
> > > > TaskManager are considered failed as well. So far, nothing unusual...
> > > >
> > > > Question is, why is it considered failed? Is this a reproducible
> > problem?
> > > > Or a one time occurrence? Does the user code accumulate so many
> objects
> > > to
> > > > put very heavy pressure on the GC?
> > > >
> > > > Stephan
> > > >
> > > >
> > > > On Wed, Oct 14, 2015 at 10:47 PM, Matthias J. Sax <[hidden email]>
> > > wrote:
> > > >
> > > >> One thing I forgot the add. I also have a Storm-WordCount job (build
> > via
> > > >> FlinkTopologyBuilder) that uses the same
> > > >> "buffer-file-and-emit-over-and-over-again-pattern" in a spout. This
> > job
> > > >> run just fine and stops regularly after 5 minutes.
> > > >>
> > > >> -Matthias
> > > >>
> > > >>
> > > >> On 10/14/2015 10:42 PM, Matthias J. Sax wrote:
> > > >>> No. See log below.
> > > >>>
> > > >>> Btw: the job is not cleaned up properly. Some task remain in state
> > > >>> "Canceling".
> > > >>>
> > > >>> The program I execute is "Streaming WordCount" example with my own
> > > >>> source function. This custom source (see below), reads a local
> > (small)
> > > >>> file, bufferes each line in an internal buffer, and emits the
> > buffered
> > > >>> lines over and over again (for 5 minutes). (A copy of the file is
> > > >>> located on every worker machine of course.) This is of course no
> > common
> > > >>> patttern; I just do this, because I have only a small file and want
> > the
> > > >>> job to run longer.
> > > >>>
> > > >>> As the job is running for multiple minutes until it fails, I assume
> > > that
> > > >>> this uncommon pattern should actually not cauase any problem. I
> just
> > > >>> report it, to make sure it is not the problem (hope you can
> confirm).
> > > >>>
> > > >>> -Matthias
> > > >>>
> > > >>>
> > > >>>> private static final class SimpleFileSource implements
> > > >> ParallelSourceFunction<String> {
> > > >>>> private static final long serialVersionUID =
> > > >> -5727198990308179551L;
> > > >>>>
> > > >>>> volatile boolean running = true;
> > > >>>> @Override
> > > >>>> public void run(
> > > >>>>
> > > >>
> > >
> >
> org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<String>
> > > >> ctx)
> > > >>>> throws Exception {
> > > >>>>
> > > >>>> BufferedReader reader = new
> BufferedReader(new
> > > >> FileReader(
> > > >>>> "/data/mjsax/hamlet.txt"));
> > > >>>> ArrayList<String> buffer = new
> > > ArrayList<String>();
> > > >>>> int counter = 0;
> > > >>>> String line;
> > > >>>>
> > > >>>> long start = System.nanoTime();
> > > >>>> while (running && (line = reader.readLine())
> !=
> > > >> null) {
> > > >>>> buffer.add(line);
> > > >>>> ctx.collect(line);
> > > >>>> }
> > > >>>> reader.close();
> > > >>>>
> > > >>>> while (running && (System.nanoTime() -
> start) <
> > > 5L
> > > >> * 60 * 1000 * 1000 * 1000) {
> > > >>>> ctx.collect(buffer.get(counter));
> > > >>>> counter = ++counter % buffer.size();
> > > >>>> }
> > > >>>> }
> > > >>>>
> > > >>>> @Override
> > > >>>> public void cancel() {
> > > >>>> running = false;
> > > >>>> }
> > > >>>> }
> > > >>>
> > > >>>
> > > >>> JobManager log:
> > > >>>
> > > >>>> 22:19:16,618 INFO
> > > >> org.apache.flink.runtime.executiongraph.ExecutionGraph -
> Keyed
> > > >> Aggregation -> Sink: Unnamed switched to
> > > FAILED
> > > >> with
> > > >>>>> exception.
> > > >>>>>> java.lang.Exception: TaskManager akka://flink/user/taskmanager
> > > >>>>> disconnects from JobManager akka.tcp://
> > > >>>>> flink@192.168.127.11:6123/user/jobmanager: JobManager is no
> longer
> > > >>>>> reachable
> > > >>>>>> at
> > > >>>>>
> > > >>
> > >
> >
> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > >
> >
> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > >
> >
> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
> > > >>>>>> at
> > > >>>>>
> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > >
> >
> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
> > > >>>>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > >
> >
> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
> > > >>>>>> at
> > akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> > > >>>>>> at
> > > >>>>>
> > > >>
> > >
> >
> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
> > > >>>>>> at
> > > >> akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
> > > >>>>>> at
> > > >> akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
> > > >>>>>> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
> > > >>>>>> at
> akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
> > > >>>>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
> > > >>>>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
> > > >>>>>> 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)
> > > >>>>>
> > > >>>>>
> > > >>>>
> > > >>>
> > > >>
> > > >>
> > > >
> > >
> > >
> >
>

till.rohrmann

Re: JobManager not reachable

And please set akka.log.lifecycle.events: true to let Akka log also its
lifecycle events.

On Thu, Oct 15, 2015 at 12:12 PM, Robert Metzger <[hidden email]>
wrote:

> Can you start flink with logging level DEBUG ?
> Then we can see from the TaskManager logs when the TM became inactive.
> Maybe an Akka message is causing the actor to block?
>
> You can also monitor the GC from the TaskManager view in the web interface
> (for example by looking at the total time spend for GCing).
>
> On Thu, Oct 15, 2015 at 12:09 PM, Stephan Ewen <[hidden email]> wrote:
>
> > Does not quite sound like GC is an issue.
> >
> > Hmmm, what else can make the failure detector kick in unexpectedly?
> >
> > On Thu, Oct 15, 2015 at 12:05 PM, Till Rohrmann <[hidden email]>
> > wrote:
> >
> > > To verify wether GC is a problem you can enable logging of memory usage
> > of
> > > the JVM via taskmanager.debug.memory.startLogThread: true. The interval
> > of
> > > the logging is configured via taskmanager.debug.memory.logIntervalMs.
> > >
> > >
> > > On Thu, Oct 15, 2015 at 12:00 PM, Matthias J. Sax <[hidden email]>
> > > wrote:
> > >
> > > > The problem is reproducible (it happens on each run).
> > > >
> > > > I doubt that GC is an issue here (at least from an UDF point of
> view),
> > > > because I read the file once and keep a String object for each line.
> > > > This objects are kept to the very end; the UDF does not release them
> > > > until it returns from "run()" method.
> > > >
> > > > The file is also small (10K) so main memory consumption should be a
> > > > problem either.
> > > >
> > > > Of course, the UDF emits on a very high rate and I am not sure if
> Flink
> > > > creates a lots of short living object (Streaming does not reuse
> object
> > > > yet...) such that GC might become an issue here?
> > > >
> > > >
> > > > -Matthias
> > > >
> > > >
> > > >
> > > > On 10/15/2015 11:44 AM, Stephan Ewen wrote:
> > > > > From what the logs show, the TaskManager does not send pings any
> more
> > > > for a
> > > > > long time and is then considered failed and the tasks running on
> that
> > > > > TaskManager are considered failed as well. So far, nothing
> unusual...
> > > > >
> > > > > Question is, why is it considered failed? Is this a reproducible
> > > problem?
> > > > > Or a one time occurrence? Does the user code accumulate so many
> > objects
> > > > to
> > > > > put very heavy pressure on the GC?
> > > > >
> > > > > Stephan
> > > > >
> > > > >
> > > > > On Wed, Oct 14, 2015 at 10:47 PM, Matthias J. Sax <
> [hidden email]>
> > > > wrote:
> > > > >
> > > > >> One thing I forgot the add. I also have a Storm-WordCount job
> (build
> > > via
> > > > >> FlinkTopologyBuilder) that uses the same
> > > > >> "buffer-file-and-emit-over-and-over-again-pattern" in a spout.
> This
> > > job
> > > > >> run just fine and stops regularly after 5 minutes.
> > > > >>
> > > > >> -Matthias
> > > > >>
> > > > >>
> > > > >> On 10/14/2015 10:42 PM, Matthias J. Sax wrote:
> > > > >>> No. See log below.
> > > > >>>
> > > > >>> Btw: the job is not cleaned up properly. Some task remain in
> state
> > > > >>> "Canceling".
> > > > >>>
> > > > >>> The program I execute is "Streaming WordCount" example with my
> own
> > > > >>> source function. This custom source (see below), reads a local
> > > (small)
> > > > >>> file, bufferes each line in an internal buffer, and emits the
> > > buffered
> > > > >>> lines over and over again (for 5 minutes). (A copy of the file is
> > > > >>> located on every worker machine of course.) This is of course no
> > > common
> > > > >>> patttern; I just do this, because I have only a small file and
> want
> > > the
> > > > >>> job to run longer.
> > > > >>>
> > > > >>> As the job is running for multiple minutes until it fails, I
> assume
> > > > that
> > > > >>> this uncommon pattern should actually not cauase any problem. I
> > just
> > > > >>> report it, to make sure it is not the problem (hope you can
> > confirm).
> > > > >>>
> > > > >>> -Matthias
> > > > >>>
> > > > >>>
> > > > >>>> private static final class SimpleFileSource implements
> > > > >> ParallelSourceFunction<String> {
> > > > >>>> private static final long serialVersionUID =
> > > > >> -5727198990308179551L;
> > > > >>>>
> > > > >>>> volatile boolean running = true;
> > > > >>>> @Override
> > > > >>>> public void run(
> > > > >>>>
> > > > >>
> > > >
> > >
> >
> org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<String>
> > > > >> ctx)
> > > > >>>> throws Exception {
> > > > >>>>
> > > > >>>> BufferedReader reader = new
> > BufferedReader(new
> > > > >> FileReader(
> > > > >>>> "/data/mjsax/hamlet.txt"));
> > > > >>>> ArrayList<String> buffer = new
> > > > ArrayList<String>();
> > > > >>>> int counter = 0;
> > > > >>>> String line;
> > > > >>>>
> > > > >>>> long start = System.nanoTime();
> > > > >>>> while (running && (line =
> reader.readLine())
> > !=
> > > > >> null) {
> > > > >>>> buffer.add(line);
> > > > >>>> ctx.collect(line);
> > > > >>>> }
> > > > >>>> reader.close();
> > > > >>>>
> > > > >>>> while (running && (System.nanoTime() -
> > start) <
> > > > 5L
> > > > >> * 60 * 1000 * 1000 * 1000) {
> > > > >>>> ctx.collect(buffer.get(counter));
> > > > >>>> counter = ++counter %
> buffer.size();
> > > > >>>> }
> > > > >>>> }
> > > > >>>>
> > > > >>>> @Override
> > > > >>>> public void cancel() {
> > > > >>>> running = false;
> > > > >>>> }
> > > > >>>> }
> > > > >>>
> > > > >>>
> > > > >>> JobManager log:
> > > > >>>
> > > > >>>> 22:19:16,618 INFO
> > > > >> org.apache.flink.runtime.executiongraph.ExecutionGraph -
> > Keyed
> > > > >> Aggregation -> Sink: Unnamed switched to
> > > > FAILED
> > > > >> with
> > > > >>>>> exception.
> > > > >>>>>> java.lang.Exception: TaskManager akka://flink/user/taskmanager
> > > > >>>>> disconnects from JobManager akka.tcp://
> > > > >>>>> flink@192.168.127.11:6123/user/jobmanager: JobManager is no
> > longer
> > > > >>>>> reachable
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > >
> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > >
> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
> > > > >>>>>> at
> > > > >>>>>
> > scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
> > > > >>>>>> at
> akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
> > > > >>>>>> at
> > > akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> > > > >>>>>> at
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
> > > > >>>>>> at
> > > > >> akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
> > > > >>>>>> at
> > > > >> akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
> > > > >>>>>> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
> > > > >>>>>> at
> > akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
> > > > >>>>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
> > > > >>>>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
> > > > >>>>>> 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)
> > > > >>>>>
> > > > >>>>>
> > > > >>>>
> > > > >>>
> > > > >>
> > > > >>
> > > > >
> > > >
> > > >
> > >
> >
>

Stephan Ewen

Re: JobManager not reachable

Blocking actor calls should not be an issue (even if they are there),
because the heartbeats go between the actor systems, rather than the
actors...

On Thu, Oct 15, 2015 at 12:14 PM, Till Rohrmann <[hidden email]>
wrote:

> And please set akka.log.lifecycle.events: true to let Akka log also its
> lifecycle events.
>
>
> On Thu, Oct 15, 2015 at 12:12 PM, Robert Metzger <[hidden email]>
> wrote:
>
> > Can you start flink with logging level DEBUG ?
> > Then we can see from the TaskManager logs when the TM became inactive.
> > Maybe an Akka message is causing the actor to block?
> >
> > You can also monitor the GC from the TaskManager view in the web
> interface
> > (for example by looking at the total time spend for GCing).
> >
> > On Thu, Oct 15, 2015 at 12:09 PM, Stephan Ewen <[hidden email]> wrote:
> >
> > > Does not quite sound like GC is an issue.
> > >
> > > Hmmm, what else can make the failure detector kick in unexpectedly?
> > >
> > > On Thu, Oct 15, 2015 at 12:05 PM, Till Rohrmann <[hidden email]>
> > > wrote:
> > >
> > > > To verify wether GC is a problem you can enable logging of memory
> usage
> > > of
> > > > the JVM via taskmanager.debug.memory.startLogThread: true. The
> interval
> > > of
> > > > the logging is configured via taskmanager.debug.memory.logIntervalMs.
> > > >
> > > >
> > > > On Thu, Oct 15, 2015 at 12:00 PM, Matthias J. Sax <[hidden email]>
> > > > wrote:
> > > >
> > > > > The problem is reproducible (it happens on each run).
> > > > >
> > > > > I doubt that GC is an issue here (at least from an UDF point of
> > view),
> > > > > because I read the file once and keep a String object for each
> line.
> > > > > This objects are kept to the very end; the UDF does not release
> them
> > > > > until it returns from "run()" method.
> > > > >
> > > > > The file is also small (10K) so main memory consumption should be a
> > > > > problem either.
> > > > >
> > > > > Of course, the UDF emits on a very high rate and I am not sure if
> > Flink
> > > > > creates a lots of short living object (Streaming does not reuse
> > object
> > > > > yet...) such that GC might become an issue here?
> > > > >
> > > > >
> > > > > -Matthias
> > > > >
> > > > >
> > > > >
> > > > > On 10/15/2015 11:44 AM, Stephan Ewen wrote:
> > > > > > From what the logs show, the TaskManager does not send pings any
> > more
> > > > > for a
> > > > > > long time and is then considered failed and the tasks running on
> > that
> > > > > > TaskManager are considered failed as well. So far, nothing
> > unusual...
> > > > > >
> > > > > > Question is, why is it considered failed? Is this a reproducible
> > > > problem?
> > > > > > Or a one time occurrence? Does the user code accumulate so many
> > > objects
> > > > > to
> > > > > > put very heavy pressure on the GC?
> > > > > >
> > > > > > Stephan
> > > > > >
> > > > > >
> > > > > > On Wed, Oct 14, 2015 at 10:47 PM, Matthias J. Sax <
> > [hidden email]>
> > > > > wrote:
> > > > > >
> > > > > >> One thing I forgot the add. I also have a Storm-WordCount job
> > (build
> > > > via
> > > > > >> FlinkTopologyBuilder) that uses the same
> > > > > >> "buffer-file-and-emit-over-and-over-again-pattern" in a spout.
> > This
> > > > job
> > > > > >> run just fine and stops regularly after 5 minutes.
> > > > > >>
> > > > > >> -Matthias
> > > > > >>
> > > > > >>
> > > > > >> On 10/14/2015 10:42 PM, Matthias J. Sax wrote:
> > > > > >>> No. See log below.
> > > > > >>>
> > > > > >>> Btw: the job is not cleaned up properly. Some task remain in
> > state
> > > > > >>> "Canceling".
> > > > > >>>
> > > > > >>> The program I execute is "Streaming WordCount" example with my
> > own
> > > > > >>> source function. This custom source (see below), reads a local
> > > > (small)
> > > > > >>> file, bufferes each line in an internal buffer, and emits the
> > > > buffered
> > > > > >>> lines over and over again (for 5 minutes). (A copy of the file
> is
> > > > > >>> located on every worker machine of course.) This is of course
> no
> > > > common
> > > > > >>> patttern; I just do this, because I have only a small file and
> > want
> > > > the
> > > > > >>> job to run longer.
> > > > > >>>
> > > > > >>> As the job is running for multiple minutes until it fails, I
> > assume
> > > > > that
> > > > > >>> this uncommon pattern should actually not cauase any problem. I
> > > just
> > > > > >>> report it, to make sure it is not the problem (hope you can
> > > confirm).
> > > > > >>>
> > > > > >>> -Matthias
> > > > > >>>
> > > > > >>>
> > > > > >>>> private static final class SimpleFileSource implements
> > > > > >> ParallelSourceFunction<String> {
> > > > > >>>> private static final long serialVersionUID =
> > > > > >> -5727198990308179551L;
> > > > > >>>>
> > > > > >>>> volatile boolean running = true;
> > > > > >>>> @Override
> > > > > >>>> public void run(
> > > > > >>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<String>
> > > > > >> ctx)
> > > > > >>>> throws Exception
> {
> > > > > >>>>
> > > > > >>>> BufferedReader reader = new
> > > BufferedReader(new
> > > > > >> FileReader(
> > > > > >>>>
> "/data/mjsax/hamlet.txt"));
> > > > > >>>> ArrayList<String> buffer = new
> > > > > ArrayList<String>();
> > > > > >>>> int counter = 0;
> > > > > >>>> String line;
> > > > > >>>>
> > > > > >>>> long start = System.nanoTime();
> > > > > >>>> while (running && (line =
> > reader.readLine())
> > > !=
> > > > > >> null) {
> > > > > >>>> buffer.add(line);
> > > > > >>>> ctx.collect(line);
> > > > > >>>> }
> > > > > >>>> reader.close();
> > > > > >>>>
> > > > > >>>> while (running && (System.nanoTime() -
> > > start) <
> > > > > 5L
> > > > > >> * 60 * 1000 * 1000 * 1000) {
> > > > > >>>> ctx.collect(buffer.get(counter));
> > > > > >>>> counter = ++counter %
> > buffer.size();
> > > > > >>>> }
> > > > > >>>> }
> > > > > >>>>
> > > > > >>>> @Override
> > > > > >>>> public void cancel() {
> > > > > >>>> running = false;
> > > > > >>>> }
> > > > > >>>> }
> > > > > >>>
> > > > > >>>
> > > > > >>> JobManager log:
> > > > > >>>
> > > > > >>>> 22:19:16,618 INFO
> > > > > >> org.apache.flink.runtime.executiongraph.ExecutionGraph -
> > > Keyed
> > > > > >> Aggregation -> Sink: Unnamed switched
> to
> > > > > FAILED
> > > > > >> with
> > > > > >>>>> exception.
> > > > > >>>>>> java.lang.Exception: TaskManager
> akka://flink/user/taskmanager
> > > > > >>>>> disconnects from JobManager akka.tcp://
> > > > > >>>>> flink@192.168.127.11:6123/user/jobmanager: JobManager is no
> > > longer
> > > > > >>>>> reachable
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > >
> > org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > >
> > org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
> > > > > >>>>>> at
> > > > > >>>>>
> > > scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
> > > > > >>>>>> at
> > akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
> > > > > >>>>>> at
> > > > akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> > > > > >>>>>> at
> > > > > >>>>>
> > > > > >>
> > > > >
> > > >
> > >
> >
> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
> > > > > >>>>>> at
> > > > > >> akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
> > > > > >>>>>> at
> > > > > >> akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
> > > > > >>>>>> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
> > > > > >>>>>> at
> > > akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
> > > > > >>>>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
> > > > > >>>>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
> > > > > >>>>>> 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)
> > > > > >>>>>
> > > > > >>>>>
> > > > > >>>>
> > > > > >>>
> > > > > >>
> > > > > >>
> > > > > >
> > > > >
> > > > >
> > > >
> > >
> >
>

Matthias J. Sax-2

Re: JobManager not reachable

Please find the logs here

http://www2.informatik.hu-berlin.de/~saxmatti/flink-mjsax-jobmanager-0-dbis21.log
http://www2.informatik.hu-berlin.de/~saxmatti/flink-mjsax-taskmanager-0-dbis34.log

-Matthias

On 10/15/2015 12:16 PM, Stephan Ewen wrote:

> Blocking actor calls should not be an issue (even if they are there),
> because the heartbeats go between the actor systems, rather than the
> actors...
>
> On Thu, Oct 15, 2015 at 12:14 PM, Till Rohrmann <[hidden email]>
> wrote:
>
>> And please set akka.log.lifecycle.events: true to let Akka log also its
>> lifecycle events.
>>
>>
>> On Thu, Oct 15, 2015 at 12:12 PM, Robert Metzger <[hidden email]>
>> wrote:
>>
>>> Can you start flink with logging level DEBUG ?
>>> Then we can see from the TaskManager logs when the TM became inactive.
>>> Maybe an Akka message is causing the actor to block?
>>>
>>> You can also monitor the GC from the TaskManager view in the web
>> interface
>>> (for example by looking at the total time spend for GCing).
>>>
>>> On Thu, Oct 15, 2015 at 12:09 PM, Stephan Ewen <[hidden email]> wrote:
>>>
>>>> Does not quite sound like GC is an issue.
>>>>
>>>> Hmmm, what else can make the failure detector kick in unexpectedly?
>>>>
>>>> On Thu, Oct 15, 2015 at 12:05 PM, Till Rohrmann <[hidden email]>
>>>> wrote:
>>>>
>>>>> To verify wether GC is a problem you can enable logging of memory
>> usage
>>>> of
>>>>> the JVM via taskmanager.debug.memory.startLogThread: true. The
>> interval
>>>> of
>>>>> the logging is configured via taskmanager.debug.memory.logIntervalMs.
>>>>>
>>>>>
>>>>> On Thu, Oct 15, 2015 at 12:00 PM, Matthias J. Sax <[hidden email]>
>>>>> wrote:
>>>>>
>>>>>> The problem is reproducible (it happens on each run).
>>>>>>
>>>>>> I doubt that GC is an issue here (at least from an UDF point of
>>> view),
>>>>>> because I read the file once and keep a String object for each
>> line.
>>>>>> This objects are kept to the very end; the UDF does not release
>> them
>>>>>> until it returns from "run()" method.
>>>>>>
>>>>>> The file is also small (10K) so main memory consumption should be a
>>>>>> problem either.
>>>>>>
>>>>>> Of course, the UDF emits on a very high rate and I am not sure if
>>> Flink
>>>>>> creates a lots of short living object (Streaming does not reuse
>>> object
>>>>>> yet...) such that GC might become an issue here?
>>>>>>
>>>>>>
>>>>>> -Matthias
>>>>>>
>>>>>>
>>>>>>
>>>>>> On 10/15/2015 11:44 AM, Stephan Ewen wrote:
>>>>>>> From what the logs show, the TaskManager does not send pings any
>>> more
>>>>>> for a
>>>>>>> long time and is then considered failed and the tasks running on
>>> that
>>>>>>> TaskManager are considered failed as well. So far, nothing
>>> unusual...
>>>>>>>
>>>>>>> Question is, why is it considered failed? Is this a reproducible
>>>>> problem?
>>>>>>> Or a one time occurrence? Does the user code accumulate so many
>>>> objects
>>>>>> to
>>>>>>> put very heavy pressure on the GC?
>>>>>>>
>>>>>>> Stephan
>>>>>>>
>>>>>>>
>>>>>>> On Wed, Oct 14, 2015 at 10:47 PM, Matthias J. Sax <
>>> [hidden email]>
>>>>>> wrote:
>>>>>>>
>>>>>>>> One thing I forgot the add. I also have a Storm-WordCount job
>>> (build
>>>>> via
>>>>>>>> FlinkTopologyBuilder) that uses the same
>>>>>>>> "buffer-file-and-emit-over-and-over-again-pattern" in a spout.
>>> This
>>>>> job
>>>>>>>> run just fine and stops regularly after 5 minutes.
>>>>>>>>
>>>>>>>> -Matthias
>>>>>>>>
>>>>>>>>
>>>>>>>> On 10/14/2015 10:42 PM, Matthias J. Sax wrote:
>>>>>>>>> No. See log below.
>>>>>>>>>
>>>>>>>>> Btw: the job is not cleaned up properly. Some task remain in
>>> state
>>>>>>>>> "Canceling".
>>>>>>>>>
>>>>>>>>> The program I execute is "Streaming WordCount" example with my
>>> own
>>>>>>>>> source function. This custom source (see below), reads a local
>>>>> (small)
>>>>>>>>> file, bufferes each line in an internal buffer, and emits the
>>>>> buffered
>>>>>>>>> lines over and over again (for 5 minutes). (A copy of the file
>> is
>>>>>>>>> located on every worker machine of course.) This is of course
>> no
>>>>> common
>>>>>>>>> patttern; I just do this, because I have only a small file and
>>> want
>>>>> the
>>>>>>>>> job to run longer.
>>>>>>>>>
>>>>>>>>> As the job is running for multiple minutes until it fails, I
>>> assume
>>>>>> that
>>>>>>>>> this uncommon pattern should actually not cauase any problem. I
>>>> just
>>>>>>>>> report it, to make sure it is not the problem (hope you can
>>>> confirm).
>>>>>>>>>
>>>>>>>>> -Matthias
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>> private static final class SimpleFileSource implements
>>>>>>>> ParallelSourceFunction<String> {
>>>>>>>>>> private static final long serialVersionUID =
>>>>>>>> -5727198990308179551L;
>>>>>>>>>>
>>>>>>>>>> volatile boolean running = true;
>>>>>>>>>> @Override
>>>>>>>>>> public void run(
>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<String>
>>>>>>>> ctx)
>>>>>>>>>> throws Exception
>> {
>>>>>>>>>>
>>>>>>>>>> BufferedReader reader = new
>>>> BufferedReader(new
>>>>>>>> FileReader(
>>>>>>>>>>
>> "/data/mjsax/hamlet.txt"));
>>>>>>>>>> ArrayList<String> buffer = new
>>>>>> ArrayList<String>();
>>>>>>>>>> int counter = 0;
>>>>>>>>>> String line;
>>>>>>>>>>
>>>>>>>>>> long start = System.nanoTime();
>>>>>>>>>> while (running && (line =
>>> reader.readLine())
>>>> !=
>>>>>>>> null) {
>>>>>>>>>> buffer.add(line);
>>>>>>>>>> ctx.collect(line);
>>>>>>>>>> }
>>>>>>>>>> reader.close();
>>>>>>>>>>
>>>>>>>>>> while (running && (System.nanoTime() -
>>>> start) <
>>>>>> 5L
>>>>>>>> * 60 * 1000 * 1000 * 1000) {
>>>>>>>>>> ctx.collect(buffer.get(counter));
>>>>>>>>>> counter = ++counter %
>>> buffer.size();
>>>>>>>>>> }
>>>>>>>>>> }
>>>>>>>>>>
>>>>>>>>>> @Override
>>>>>>>>>> public void cancel() {
>>>>>>>>>> running = false;
>>>>>>>>>> }
>>>>>>>>>> }
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> JobManager log:
>>>>>>>>>
>>>>>>>>>> 22:19:16,618 INFO
>>>>>>>> org.apache.flink.runtime.executiongraph.ExecutionGraph -
>>>> Keyed
>>>>>>>> Aggregation -> Sink: Unnamed switched
>> to
>>>>>> FAILED
>>>>>>>> with
>>>>>>>>>>> exception.
>>>>>>>>>>>> java.lang.Exception: TaskManager
>> akka://flink/user/taskmanager
>>>>>>>>>>> disconnects from JobManager akka.tcp://
>>>>>>>>>>> flink@192.168.127.11:6123/user/jobmanager: JobManager is no
>>>> longer
>>>>>>>>>>> reachable
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> org.apache.flink.runtime.taskmanager.TaskManager.handleJobManagerDisconnect(TaskManager.scala:826)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> org.apache.flink.runtime.taskmanager.TaskManager$$anonfun$handleMessage$1.applyOrElse(TaskManager.scala:297)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:44)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>
>>> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>
>>> org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
>>>>>>>>>>>> at
>>> akka.actor.Actor$class.aroundReceive(Actor.scala:465)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> org.apache.flink.runtime.taskmanager.TaskManager.aroundReceive(TaskManager.scala:119)
>>>>>>>>>>>> at
>>>>> akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
>>>>>>>>>>>> at
>>>>>>>>>>>
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>> akka.actor.dungeon.DeathWatch$class.receivedTerminated(DeathWatch.scala:46)
>>>>>>>>>>>> at
>>>>>>>> akka.actor.ActorCell.receivedTerminated(ActorCell.scala:369)
>>>>>>>>>>>> at
>>>>>>>> akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:501)
>>>>>>>>>>>> at akka.actor.ActorCell.invoke(ActorCell.scala:486)
>>>>>>>>>>>> at
>>>> akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
>>>>>>>>>>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
>>>>>>>>>>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
>>>>>>>>>>>> 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)
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>
>

signature.asc (836 bytes) Download Attachment