[ 
https://issues.apache.org/jira/browse/FLINK-2472?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14653547#comment-14653547
 ] 

ASF GitHub Bot commented on FLINK-2472:
---------------------------------------

Github user sachingoel0101 commented on a diff in the pull request:

    https://github.com/apache/flink/pull/979#discussion_r36181722
  
    --- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClientActor.java 
---
    @@ -59,11 +87,50 @@ public JobClientActor(
                this.leaderSessionID = 
Preconditions.checkNotNull(leaderSessionID, "The leader session ID option must 
not be null.");
     
                this.sysoutUpdates = sysoutUpdates;
    +           // set this to -1 to indicate the job hasn't been created yet.
    +           this.currentJobCreatedAt = -1;
        }
        
        @Override
        protected void handleMessage(Object message) {
                
    +           // ======= Job status messages on regular intervals 
==============
    +           if(message instanceof JobManagerMessages.CurrentJobStatus){
    +                   JobStatus statusReport = 
((JobManagerMessages.CurrentJobStatus) message).status();
    +                   long timeDiff;
    +                   switch(statusReport){
    +                           case RUNNING:
    +                                   // Vincent, we happy?
    +                                   this.currentJobCreatedAt = -1;
    +                                   break;
    +                           case FINISHED:
    +                                   // Yeah! We happy!
    +                                   this.currentJobCreatedAt = -1;
    +                                   break;
    +                           case CREATED:
    +                                   // we're still at Job CREATED. Let's 
see if we're over the limit.
    +                                   timeDiff = (System.currentTimeMillis() 
- this.currentJobCreatedAt);
    +                                   if(timeDiff > 
JOB_CLIENT_JOB_STATUS_TIMEOUT){
    +                                           failWithTimeout(timeDiff);
    +                                   } // otherwise just wait a bit longer.
    +                                   break;
    +                           case RESTARTING:
    --- End diff --
    
    Isn't getting `RESTARTING` multiple times a bad thing? I'm not sure though. 
In such a case the only solution would be to lower the heartbeat period I think.


> Make the JobClientActor check periodically if the submitted Job is still 
> running and if the JobManager is still alive
> ---------------------------------------------------------------------------------------------------------------------
>
>                 Key: FLINK-2472
>                 URL: https://issues.apache.org/jira/browse/FLINK-2472
>             Project: Flink
>          Issue Type: Improvement
>            Reporter: Till Rohrmann
>            Assignee: Sachin Goel
>
> In case that the {{JobManager}} dies without notifying possibly connected 
> {{JobClientActors}} or if the job execution finishes without sending the 
> {{SerializedJobExecutionResult}} back to the {{JobClientActor}}, it might 
> happen that a {{JobClient.submitJobAndWait}} never returns.
> I propose to let the {{JobClientActor}} periodically check whether the 
> {{JobManager}} is still alive and whether the submitted job is still running. 
> If not, then the {{JobClientActor}} should return an exception to complete 
> the waiting future.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Reply via email to