flink-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Matthias J. Sax" <mj...@informatik.hu-berlin.de>
Subject Re: Question about Infinite Streaming Job on Mini Cluster and ITCase
Date Wed, 01 Apr 2015 12:55:09 GMT
I agree.

@Marton:
The idea with the extra thread does not work, because the method
JobClient.submitJobAndWait(...) does not return regularly if
ForkableFlinkMiniCluster.shutdown() is called -- instead an exception
occurs:

> Exception in thread "Thread-8" java.lang.RuntimeException: org.apache.flink.runtime.client.JobTimeoutException:
Lost connection to job manager.
> 	at org.apache.flink.streaming.util.TestStreamEnvironment$1.run(TestStreamEnvironment.java:119)
> Caused by: org.apache.flink.runtime.client.JobTimeoutException: Lost connection to job
manager.
> 	at org.apache.flink.runtime.client.JobClient$.submitJobAndWait(JobClient.scala:228)
> 	at org.apache.flink.runtime.client.JobClient.submitJobAndWait(JobClient.scala)
> 	at org.apache.flink.streaming.util.TestStreamEnvironment$1.run(TestStreamEnvironment.java:117)
> Caused by: akka.pattern.AskTimeoutException: Recipient[Actor[akka://flink/user/jobclient#-596117797]]
had already been terminated.
> 	at akka.pattern.AskableActorRef$.ask$extension(AskSupport.scala:132)
> 	at akka.pattern.AskableActorRef$.$qmark$extension(AskSupport.scala:144)
> 	at org.apache.flink.runtime.client.JobClient$.submitJobAndWait(JobClient.scala:222)
> 	... 2 more


Thus, I cannot get an JobExecutionResult this way, either.


-Matthias


On 04/01/2015 02:36 PM, Stephan Ewen wrote:
> As a followup - I think it would be a good thing to add a way to gracefully
> stop a streaming job.
> 
> Something that sends "close" to the sources, and they quit.
> 
> We can use this for graceful shutdown wen re-partitioninig / scaling in or
> out, ...
> 
> On Wed, Apr 1, 2015 at 1:29 PM, Matthias J. Sax <
> mjsax@informatik.hu-berlin.de> wrote:
> 
>> Hi,
>>
>> I will pull the fix and try it out.
>>
>> Thanks for the hint with the extra Thread. That should work for me. But
>> you are actually right; my setup is Storm inspired. I thinks its a very
>> natural way to deploy and stop and infinite streaming job. Maybe, you
>> want to adopt to it.
>>
>> The ITCase I am writing bases on StreamingProgramTestBase, so I need the
>> JobExecutionResult because the test fails without it.
>>
>>
>> -Matthias
>>
>>
>>
>> On 04/01/2015 11:09 AM, Márton Balassi wrote:
>>> Hey Matthias,
>>>
>>> Thanks for reporting the Exception thrown, we were not preparing for this
>>> use case yet. We fixed it with Gyula, he is pushing a fix for it right
>> now:
>>> When the job is cancelled (for example due to shutting down the executor
>>> underneath) you should not see that InterruptedException as soon as this
>>> commit is in. [1]
>>>
>>> As for getting the streaming JobExecutionResult back from a detached job
>> my
>>> current best practice is what you can see in
>>> the ProcessFailureRecoveryTestBase and its streaming implementation:
>>> starting an executor in a separate thread and then joining it with the
>> main
>>> one. Would you prefer a more Storm example-ish solution? [2]
>>>
>>> [1] https://github.com/mbalassi/flink/commit/5db06d6d
>>> [2]
>>>
>> https://github.com/apache/storm/blob/master/examples/storm-starter/src/jvm/storm/starter/WordCountTopology.java#L99-104
>>>
>>> On Tue, Mar 31, 2015 at 2:54 PM, Matthias J. Sax <
>>> mjsax@informatik.hu-berlin.de> wrote:
>>>
>>>> Hi Robert,
>>>>
>>>> thanks for your answer.
>>>>
>>>> I get an InterruptedException when I call shutdown():
>>>>
>>>> java.lang.InterruptedException
>>>>         at java.lang.Object.wait(Native Method)
>>>>         at java.lang.Thread.join(Thread.java:1225)
>>>>         at java.lang.Thread.join(Thread.java:1278)
>>>>         at
>>>>
>>>>
>> org.apache.flink.streaming.io.StreamRecordWriter.close(StreamRecordWriter.java:55)
>>>>         at
>>>>
>>>>
>> org.apache.flink.streaming.api.collector.StreamOutput.close(StreamOutput.java:77)
>>>>         at
>>>>
>>>>
>> org.apache.flink.streaming.api.streamvertex.OutputHandler.flushOutputs(OutputHandler.java:204)
>>>>         at
>>>>
>>>>
>> org.apache.flink.streaming.api.streamvertex.StreamVertex.invoke(StreamVertex.java:195)
>>>>         at
>>>>
>>>>
>> org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:217)
>>>>         at java.lang.Thread.run(Thread.java:701)
>>>>
>>>>
>>>> About the JobExecutionResult:
>>>>
>>>> I added a new method to the API, that calls
>>>> JobClient.submitJobDetached(...) instead of
>>>> JobClient.submitJobAndWait(...). The "detached" version has no return
>>>> value, while the blocking one returns a JobExecutionResult that is
>>>> further returned by execute(). So I cannot get a JobExecutionResult
>>>> right now.
>>>>
>>>> It would be nice to get the JobExecutionResult when stopping the running
>>>> program via a "stop-execution"-call (is there any way to do this?).
>>>> Right now, I sleep for a certain time after calling
>>>> submitJobDetached(...) an call stop() and shutdown() later on (from
>>>> ForkableMiniCluster). The stop() call does not seem to do anything...
>>>> shutdown() works (except for the Exception I get -- as described above).
>>>>
>>>>
>>>> -Matthias
>>>>
>>>>
>>>> On 03/30/2015 09:08 PM, Robert Metzger wrote:
>>>>> Hi Matthias,
>>>>>
>>>>> the streaming folks can probably answer the questions better. But I'll
>>>>> write something to bring this message back to their attention ;)
>>>>>
>>>>> 1) Which exceptions are you seeing? Flink should be able to cleanly
>> shut
>>>>> down.
>>>>> 2) As far as I saw it, the execute() method (of the Streaming API) got
>> an
>>>>> JobExecutionResult return type in the latest master. That contains
>>>>> accumulator results.
>>>>> 3) I think the cancel() method is there for exactly that purpose. If
>> the
>>>>> job is shutting down before the cancel method, that probably a bug.
>>>>>
>>>>>
>>>>> Robert
>>>>>
>>>>>
>>>>>
>>>>> On Fri, Mar 27, 2015 at 10:07 PM, Matthias J. Sax <
>>>>> mjsax@informatik.hu-berlin.de> wrote:
>>>>>
>>>>>> Hi,
>>>>>>
>>>>>> I am trying to run an infinite streaming job (ie, one that does not
>>>>>> terminate because it is generating output date randomly on the fly).
I
>>>>>> kill this job with .stop() or .shutdown() method of
>>>>>> ForkableFlinkMiniCluster.
>>>>>>
>>>>>> I did not find any example using a similar setup. In the provided
>>>>>> examples, each job terminate automatically, because only a finite
>> input
>>>>>> is processed and the source returns after all data is emitted.
>>>>>>
>>>>>>
>>>>>> I have multiple question about my setup:
>>>>>>
>>>>>>  1) The job never terminates "clean", ie, I get some exceptions.
Is
>> this
>>>>>> behavior desired?
>>>>>>
>>>>>>  2) Is it possible to get a result back? Similar to
>>>>>> JobClient.submitJobAndWait(...)?
>>>>>>
>>>>>>  3) Is it somehow possible, to send a signal to the running job such
>>>>>> that the source can terminate regularly as if finite input would
be
>>>>>> processed? Right now, I use an while(running) loop and set 'running'
>> to
>>>>>> false in the .cancel() method.
>>>>>>
>>>>>>
>>>>>>
>>>>>> Thanks for your help!
>>>>>>
>>>>>> -Matthias
>>>>>>
>>>>>>
>>>>>>
>>>>>
>>>>
>>>>
>>>
>>
>>
> 


Mime
View raw message