flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From miki haiat <miko5...@gmail.com>
Subject Re: Temporary failure in name resolution
Date Wed, 04 Apr 2018 06:30:30 GMT
HI ,

i checked the code again the figure out where the problem  can be

i just wondered if im implementing the Evictor correctly  ?

full code
https://gist.github.com/miko-code/6d7010505c3cb95be122364b29057237




public static class EsbTraceEvictor implements Evictor<EsbTrace, GlobalWindow> {
    org.slf4j.Logger LOG = LoggerFactory.getLogger(EsbTraceEvictor.class);
    @Override
    public void evictBefore(Iterable<TimestampedValue<EsbTrace>>
iterable, int i, GlobalWindow globalWindow, Evictor.EvictorContext
evictorContext) {

    }

    @Override
    public void evictAfter(Iterable<TimestampedValue<EsbTrace>>
elements, int i, GlobalWindow globalWindow, EvictorContext
evictorContext) {
        //change it to current procces  time
        long min5min = LocalDateTime.now().minusMinutes(5).getNano();
        LOG.info("time now -5min",min5min);
        DateTimeFormatter format = DateTimeFormatter.ISO_DATE_TIME;
        for (Iterator<TimestampedValue<EsbTrace>> iterator =
elements.iterator(); iterator.hasNext(); ) {
            TimestampedValue<EsbTrace> element = iterator.next();
            LocalDateTime el =
LocalDateTime.parse(element.getValue().getEndDate(),format);
            LOG.info("element time ",element.getValue().getEndDate());
            if (el.minusMinutes(5).getNano() <= min5min) {
                iterator.remove();
            }
        }
    }
}






On Tue, Apr 3, 2018 at 4:28 PM, Hao Sun <hasun@zendesk.com> wrote:

> Hi Timo, we do have similar issue, TM got killed by a job. Is there a way
> to monitor JVM status? If through the monitor metrics, what metric I should
> look after?
> We are running Flink on K8S. Is there a possibility that a job consumes
> too much network bandwidth, so JM and TM can not connect?
>
> On Tue, Apr 3, 2018 at 3:11 AM Timo Walther <twalthr@apache.org> wrote:
>
>> Hi Miki,
>>
>> for me this sounds like your job has a resource leak such that your
>> memory fills up and the JVM of the TaskManager is killed at some point. How
>> does your job look like? I see a WindowedStream.apply which might not be
>> appropriate if you have big/frequent windows where the evaluation happens
>> too late such that the state becomes too big.
>>
>> Regards,
>> Timo
>>
>>
>> Am 03.04.18 um 08:26 schrieb miki haiat:
>>
>> i tried to run flink on kubernetes and  as stand alone HA cluster and on
>> both cases  task manger got lost/kill after few hours/days    .
>> im using ubuntu and flink 1.4.2 .
>>
>>
>> this is part of the log , i also attaches the full log .
>>
>>>
>>> org.tlv.esb.StreamingJob$EsbTraceEvictor@20ffca60, WindowedStream.apply(WindowedStream.java:1061))
>>> -> Sink: Unnamed (1/1) (91b27853aa30be93322d9c516ec266bf) switched from
>>> RUNNING to FAILED.
>>> java.lang.Exception: TaskManager was lost/killed:
>>> 6dc6cd5c15588b49da39a31b6480b2e3 @ beam2 (dataPort=42587)
>>> at org.apache.flink.runtime.instance.SimpleSlot.
>>> releaseSlot(SimpleSlot.java:217)
>>> at org.apache.flink.runtime.instance.SlotSharingGroupAssignment.
>>> releaseSharedSlot(SlotSharingGroupAssignment.java:523)
>>> at org.apache.flink.runtime.instance.SharedSlot.
>>> releaseSlot(SharedSlot.java:192)
>>> at org.apache.flink.runtime.instance.Instance.markDead(
>>> Instance.java:167)
>>> at org.apache.flink.runtime.instance.InstanceManager.
>>> unregisterTaskManager(InstanceManager.java:212)
>>> at org.apache.flink.runtime.jobmanager.JobManager.org$
>>> apache$flink$runtime$jobmanager$JobManager$$handleTaskManagerTerminated(
>>> JobManager.scala:1198)
>>> at org.apache.flink.runtime.jobmanager.JobManager$$
>>> anonfun$handleMessage$1.applyOrElse(JobManager.scala:1096)
>>> at scala.runtime.AbstractPartialFunction.apply(
>>> AbstractPartialFunction.scala:36)
>>> at org.apache.flink.runtime.LeaderSessionMessageFilter$$
>>> anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:49)
>>> at scala.runtime.AbstractPartialFunction.apply(
>>> AbstractPartialFunction.scala:36)
>>> 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:123)
>>> at org.apache.flink.runtime.LogMessages$$anon$1.
>>> applyOrElse(LogMessages.scala:28)
>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:502)
>>> at org.apache.flink.runtime.jobmanager.JobManager.
>>> aroundReceive(JobManager.scala:122)
>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
>>> at akka.actor.dungeon.DeathWatch$class.receivedTerminated(
>>> DeathWatch.scala:46)
>>> at akka.actor.ActorCell.receivedTerminated(ActorCell.scala:374)
>>> at akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:511)
>>> at akka.actor.ActorCell.invoke(ActorCell.scala:494)
>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
>>> at akka.dispatch.Mailbox.run(Mailbox.scala:224)
>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
>>> at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
>>> at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.
>>> runTask(ForkJoinPool.java:1339)
>>> at scala.concurrent.forkjoin.ForkJoinPool.runWorker(
>>> ForkJoinPool.java:1979)
>>> at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(
>>> ForkJoinWorkerThread.java:107)
>>> 2018-04-02 13:09:01,727 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph
>>> - Job Flink Streaming esb correlate msg (0db04ff29124f59a123d4743d89473ed)
>>> switched from state RUNNING to FAILING.
>>> java.lang.Exception: TaskManager was lost/killed:
>>> 6dc6cd5c15588b49da39a31b6480b2e3 @ beam2 (dataPort=42587)
>>> at org.apache.flink.runtime.instance.SimpleSlot.
>>> releaseSlot(SimpleSlot.java:217)
>>> at org.apache.flink.runtime.instance.SlotSharingGroupAssignment.
>>> releaseSharedSlot(SlotSharingGroupAssignment.java:523)
>>> at org.apache.flink.runtime.instance.SharedSlot.
>>> releaseSlot(SharedSlot.java:192)
>>> at org.apache.flink.runtime.instance.Instance.markDead(
>>> Instance.java:167)
>>> at org.apache.flink.runtime.instance.InstanceManager.
>>> unregisterTaskManager(InstanceManager.java:212)
>>> at org.apache.flink.runtime.jobmanager.JobManager.org$
>>> apache$flink$runtime$jobmanager$JobManager$$handleTaskManagerTerminated(
>>> JobManager.scala:1198)
>>> at org.apache.flink.runtime.jobmanager.JobManager$$
>>> anonfun$handleMessage$1.applyOrElse(JobManager.scala:1096)
>>> at scala.runtime.AbstractPartialFunction.apply(
>>> AbstractPartialFunction.scala:36)
>>> at org.apache.flink.runtime.LeaderSessionMessageFilter$$
>>> anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:49)
>>> at scala.runtime.AbstractPartialFunction.apply(
>>> AbstractPartialFunction.scala:36)
>>> 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:123)
>>> at org.apache.flink.runtime.LogMessages$$anon$1.
>>> applyOrElse(LogMessages.scala:28)
>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:502)
>>> at org.apache.flink.runtime.jobmanager.JobManager.
>>> aroundReceive(JobManager.scala:122)
>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
>>> at akka.actor.dungeon.DeathWatch$class.receivedTerminated(
>>> DeathWatch.scala:46)
>>> at akka.actor.ActorCell.receivedTerminated(ActorCell.scala:374)
>>> at akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:511)
>>> at akka.actor.ActorCell.invoke(ActorCell.scala:494)
>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
>>> at akka.dispatch.Mailbox.run(Mailbox.scala:224)
>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
>>> at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
>>> at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.
>>> runTask(ForkJoinPool.java:1339)
>>> at scala.concurrent.forkjoin.ForkJoinPool.runWorker(
>>> ForkJoinPool.java:1979)
>>> at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(
>>> ForkJoinWorkerThread.java:107)
>>> 2018-04-02 13:09:01,737 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph
>>> - Source: Custom Source (1/1) (a10c25c2d3de57d33828524938fcfcc2)
>>> switched from RUNNING to CANCELING.
>>
>>
>>
>>
>>
>>

Mime
View raw message