flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sachingoel0101 <...@git.apache.org>
Subject [GitHub] flink pull request: [FLINK-2472]Make JobClientActor poll JobManage...
Date Mon, 24 Aug 2015 16:05:13 GMT
Github user sachingoel0101 commented on a diff in the pull request:

    https://github.com/apache/flink/pull/979#discussion_r37768306
  
    --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClientActor.java
---
    @@ -49,27 +57,127 @@
     	// Actor which submits a job to the JobManager via this actor
     	private ActorRef submitter;
     
    +	// timeout for a message from the job manager
    +	private static FiniteDuration JOB_CLIENT_JOB_MANAGER_TIMEOUT;
    +
    +	// heartbeat interval for pinging the job manager for job status
    +	private static FiniteDuration JOB_CLIENT_HEARTBEAT_INTERVAL;
    +
    +	// initial time delay before starting pinging job manager over regular intervals
    +	private static FiniteDuration JOB_CLIENT_INITIAL_PING_DELAY;
    +
    +	// maximum waiting time for a job to go to running status (milliseconds)
    +	private static long JOB_CLIENT_JOB_STATUS_TIMEOUT;
    +
    +	// time at which the current job was created
    +	private long currentJobCreatedAt;
    +
    +	// current job id
    +	private JobID currentJobId;
    +
    +	// scheduler to ping JobManager after a time interval
    +	private Cancellable scheduler;
    +
    +	// maintain when we got our last ping from the Job Manager.
    +	private long jobManagerPinged = 0;
    +
    +	// maintain the last time we got a terminal state message
    +	private long terminalStateAt = 0;
    +
     	public JobClientActor(
     			ActorRef jobManager,
     			Logger logger,
     			boolean sysoutUpdates,
    -			Option<UUID> leaderSessionID) {
    +			Option<UUID> leaderSessionID,
    +			Configuration config) {
     		this.jobManager = Preconditions.checkNotNull(jobManager, "The JobManager ActorRef must
not be null.");
     		this.logger = Preconditions.checkNotNull(logger, "The logger must not be null.");
     		this.leaderSessionID = Preconditions.checkNotNull(leaderSessionID, "The leader session
ID option must not be null.");
     
     		this.sysoutUpdates = sysoutUpdates;
    +		// set this to 0 to indicate the job hasn't been created yet.
    +		this.currentJobCreatedAt = 0;
    +		this.terminalStateAt = 0;
    +		parseTimes(config);
     	}
     	
     	@Override
     	protected void handleMessage(Object message) {
    -		
    +
    +		// first see if the message was from the Job Manager
    +		if(getContext().sender() == jobManager){
    +			this.jobManagerPinged = System.currentTimeMillis();
    +		}
    +
    +		// ======= 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?
    +					resetTimeouts();
    +					break;
    +				case FINISHED:
    +					// Yeah! We happy!
    +					resetTimeouts();
    +					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:
    +					if(this.currentJobCreatedAt == 0){
    --- End diff --
    
    Will do.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

Mime
View raw message