Override this method to create the proper jobClient and the thread that
sends jobTracker heartbeat.
963021
@Override
  protected RunningJob createRunningJob(JobID jobId, TaskInProgress tip)
      throws IOException {
    CoronaSessionInfo info = (CoronaSessionInfo)(tip.getExtensible());
    // JobClient will be set by JobTrackerReporter thread later
    RunningJob rJob = new RunningJob(jobId, null, info);
    JobTrackerReporter reporter = new JobTrackerReporter(rJob,
        info.getJobTrackerAddr(), info.getSecondaryTracker(),
        info.getSessionHandle());
    reporter.setName("JobTrackerReporter for " + jobId);
    // Start the heartbeat to the jobtracker
    reporter.start();
    jobTrackerReporters.put(jobId, reporter);
    return rJob;
  }
5357
private void runHeartbeatLoop() {
		final long interval = GlobalConfiguration.getInteger(
						ConfigConstants.TASK_MANAGER_HEARTBEAT_INTERVAL_KEY,
						ConfigConstants.DEFAULT_TASK_MANAGER_HEARTBEAT_INTERVAL);

		while (!shutdownStarted.get()) {
			// send heart beat
			try {
				LOG.debug("heartbeat");
				this.jobManager.sendHeartbeat(this.localInstanceConnectionInfo, this.hardwareDescription);
			} catch (IOException e) {
				if (shutdownStarted.get()) {
					break;
				} else {
					LOG.error("Sending the heart beat caused an exception: " + e.getMessage(), e);
				}
			}
			
			// sleep until the next heart beat
			try {
				Thread.sleep(interval);
			}
			catch (InterruptedException e) {
				if (!shutdownStarted.get()) {
					LOG.error("TaskManager heart beat loop was interrupted without shutdown.");
				}
			}
		}
	}
635963
private synchronized JobMonitor getJobMonitor(JobStatus status) {
    UUID jobId = status.getJobId();
    JobMonitor monitor = monitors.get(jobId);
    if (monitor == null) {
      monitor = new JobMonitor(factory.createProgressMonitor(status.getDescription()));
      monitors.put(jobId, monitor);
    }
    return monitor;
  }
1196270
@Override
    public WSJobInstance createJobInstance(String appName, String jobXMLName, String submitter, String jsl, String correlationId) {

        JobInstanceEntity retMe = null;

        retMe = getPersistenceManagerService().createJobInstance(appName,
                                                                 jobXMLName,
                                                                 jsl,
                                                                 submitter,
                                                                 new Date());

        publishEvent(retMe, BatchEventsPublisher.TOPIC_INSTANCE_SUBMITTED, correlationId);

        return retMe;
    }
871110
public void startListening()
    {
        // Register progress listener
        this.observationManager.addListener(new WrappedThreadEventListener(this.progress));

        // Isolate log for the job status
        this.logListener = new LoggerListener(LoggerListener.class.getName() + '_' + hashCode(), this.logs);
        if (isIsolated()) {
            this.loggerManager.pushLogListener(this.logListener);
        } else {
            this.observationManager.addListener(new WrappedThreadEventListener(this.logListener));
        }
    }
997100
private void runJobLauncherLoop() throws JobException {
    try {
      while (true) {
        currentJobLauncher = this.jobScheduler.buildJobLauncher(jobProps);
        // in "run once" case, job scheduler will remove current job from the scheduler
        boolean isEarlyStopped = this.jobScheduler.runJob(jobProps, jobListener, currentJobLauncher);
        boolean isRetriggerEnabled = this.isRetriggeringEnabled();
        if (isEarlyStopped && isRetriggerEnabled) {
          log.info("Job {} will be re-triggered.", jobProps.getProperty(ConfigurationKeys.JOB_NAME_KEY));
        } else {
          break;
        }
        currentJobLauncher = null;
      }
    } catch (Exception e) {
      log.error("Failed to run job {}", jobProps.getProperty(ConfigurationKeys.JOB_NAME_KEY), e);
      throw new JobException("Failed to run job " + jobProps.getProperty(ConfigurationKeys.JOB_NAME_KEY), e);
    }
  }
1083886
@Override
  public JobMasterClientContext build() {
    if (mMasterInquireClient == null) {
      mMasterInquireClient = MasterInquireClient.Factory.createForJobMaster(mContext.getConf());
    }
    return new JobMasterClientContext(mContext, mMasterInquireClient);
  }
224627
def handle_heartbeat(queue_name):
    """"""
    task_id = request.form.get('task_id', type=str)
    message = request.form.get('message', type=str)
    index = request.form.get('index', type=int)
    try:
        work_queue.heartbeat(
            queue_name,
            task_id,
            request.form.get('owner', request.remote_addr, type=str),
            message,
            index)
    except work_queue.Error, e:
        return utils.jsonify_error(e)

    db.session.commit()
    logging.debug('Task heartbeat: queue=%r, task_id=%r, message=%r, index=%d',
                  queue_name, task_id, message, index)
    return flask.jsonify(success=True)
735355
private final void startHeartbeat() {

        if (timerTask == null || HsqlTimer.isCancelled(timerTask)) {
            Runnable runner = new HeartbeatRunner();

            timerTask = timer.schedulePeriodicallyAfter(0, HEARTBEAT_INTERVAL,
                    runner, true);
        }
    }
524355
async def new_job(self, message: BackendNewJob):
        
        self._logger.info("Received request for jobid %s", message.job_id)
        future_results = asyncio.Future()
        out = await self._loop.run_in_executor(None, lambda: self.__new_job_sync(message, future_results))
        self._create_safe_task(self.handle_running_container(**out, future_results=future_results))
        await self._timeout_watcher.register_container(out["container_id"], out["orig_time_limit"], out["orig_hard_time_limit"])
257169
function (heartbeat, cb) {
        var self = this;
        this.swfClient.recordActivityTaskHeartbeat({
            taskToken: this.config.taskToken,
            details: stringify(heartbeat)
        }, function (err) {
            if (cb) {
                cb(err);
            }

        });
    }
914394
private synchronized void sendHeartBeat(
      final EvaluatorRuntimeProtocol.EvaluatorHeartbeatProto heartbeatProto) {
    if (LOG.isLoggable(Level.FINEST)) {
      LOG.log(Level.FINEST, "Heartbeat message:\n" + heartbeatProto, new Exception("Stack trace"));
    }
    this.evaluatorHeartbeatHandler.onNext(heartbeatProto);
  }
445725
public void prepareFailover() {
    if (!RemoteJTProxy.isJTRestartingEnabled(conf)) {
      return;
    }
    
    LOG.info("prepareFailover done");
    
    this.isPurgingJob = false;
    
    if (this.parentHeartbeat != null) {
      // Because our failover mechanism based on remotJTProxy can't 
      // reach remote job tracker, we stop the interTrackerServer to
      // trigger the failover
      this.interTrackerServer.stop();
    }
  }
793187
@Override
    public synchronized void start() {
        if (monitorActive) {
            return;
        }
        lastTime.set(milliSecondFromNano());
        long localCheckInterval = checkInterval.get();
        if (localCheckInterval > 0) {
            monitorActive = true;
            monitor = new MixedTrafficMonitoringTask((GlobalChannelTrafficShapingHandler) trafficShapingHandler, this);
            scheduledFuture =
                executor.schedule(monitor, localCheckInterval, TimeUnit.MILLISECONDS);
        }
    }
139275
@Override
  public void onNext(final JobSubmissionEvent jobSubmissionEvent) {
    LOG.log(Level.FINEST, "Submitting job: {0}", jobSubmissionEvent);

    try {
      this.applicationId = createApplicationId(jobSubmissionEvent);
      final String folderName = this.azureBatchFileNames.getStorageJobFolder(this.applicationId);

      LOG.log(Level.FINE, "Creating a job folder on Azure at: {0}.", folderName);
      final URI jobFolderURL = this.azureStorageClient.getJobSubmissionFolderUri(folderName);

      LOG.log(Level.FINE, "Getting a shared access signature for {0}.", folderName);
      final String storageContainerSAS = this.azureStorageClient.createContainerSharedAccessSignature();

      LOG.log(Level.FINE, "Assembling Configuration for the Driver.");
      final Configuration driverConfiguration = makeDriverConfiguration(jobSubmissionEvent, this.applicationId,
          jobFolderURL);

      LOG.log(Level.FINE, "Making Job JAR.");
      final File jobSubmissionJarFile =
          this.jobJarMaker.createJobSubmissionJAR(jobSubmissionEvent, driverConfiguration);

      LOG.log(Level.FINE, "Uploading Job JAR to Azure.");
      final URI jobJarSasUri = this.azureStorageClient.uploadFile(folderName, jobSubmissionJarFile);

      LOG.log(Level.FINE, "Assembling application submission.");
      final String command = this.launchCommandBuilder.buildDriverCommand(jobSubmissionEvent);

      this.azureBatchHelper.submitJob(getApplicationId(), storageContainerSAS, jobJarSasUri, command);

    } catch (final IOException e) {
      LOG.log(Level.SEVERE, "Error submitting Azure Batch request: {0}", e);
      throw new RuntimeException(e);
    }
  }
65626
@PostConstruct
    public final void init() {
        long timeToKeepAfterAccessInMillis = this.jobQueue.getTimeToKeepAfterAccessInMillis();
        if (timeToKeepAfterAccessInMillis >= 0) {
            if (TimeUnit.SECONDS.toMillis(this.abandonedTimeout) >=
                    this.jobQueue.getTimeToKeepAfterAccessInMillis()) {
                final String msg =
                        String.format("%s abandonTimeout must be smaller than %s timeToKeepAfterAccess",
                                      getClass().getName(), this.jobQueue.getClass().getName());
                throw new IllegalStateException(msg);
            }
            if (TimeUnit.SECONDS.toMillis(this.timeout) >= this.jobQueue.getTimeToKeepAfterAccessInMillis()) {
                final String msg = String.format("%s timeout must be smaller than %s timeToKeepAfterAccess",
                                                 getClass().getName(), this.jobQueue.getClass().getName());
                throw new IllegalStateException(msg);
            }
        }
        CustomizableThreadFactory threadFactory = new CustomizableThreadFactory();
        threadFactory.setDaemon(true);
        threadFactory.setThreadNamePrefix("PrintJobManager-");

        PriorityBlockingQueue<Runnable> queue =
                new PriorityBlockingQueue<>(
                        this.maxNumberOfWaitingJobs,
                        (o1, o2) -> {
                            if (o1 instanceof JobFutureTask<?> &&
                                    o2 instanceof JobFutureTask<?>) {
                                Callable<?> callable1 = ((JobFutureTask<?>) o1).getCallable();
                                Callable<?> callable2 = ((JobFutureTask<?>) o2).getCallable();

                                if (callable1 instanceof PrintJob) {
                                    if (callable2 instanceof PrintJob) {
                                        return ThreadPoolJobManager.this.jobPriorityComparator
                                                .compare((PrintJob) callable1,
                                                         (PrintJob) callable2);
                                    }
                                    return 1;
                                } else if (callable2 instanceof PrintJob) {
                                    return -1;
                                }
                            }
                            return 0;
                        });
        /* The ThreadPoolExecutor uses a unbounded queue (though we are enforcing a limit in `submit()`).
         * Because of that, the executor creates only `corePoolSize` threads. But to use all threads,
         * we set both `corePoolSize` and `maximumPoolSize` to `maxNumberOfRunningPrintJobs`. As a
         * consequence, the `maxIdleTime` will be ignored, idle threads will not be terminated.
         */
        this.executor =
                new ThreadPoolExecutor(this.maxNumberOfRunningPrintJobs, this.maxNumberOfRunningPrintJobs,
                                       this.maxIdleTime, TimeUnit.SECONDS, queue, threadFactory) {
                    @Override
                    protected <T> RunnableFuture<T> newTaskFor(final Callable<T> callable) {
                        return new JobFutureTask<>(callable);
                    }

                    @Override
                    protected void beforeExecute(final Thread t, final Runnable runnable) {
                        if (!ThreadPoolJobManager.this.clustered && runnable instanceof JobFutureTask<?>) {
                            JobFutureTask<?> task = (JobFutureTask<?>) runnable;
                            if (task.getCallable() instanceof PrintJob) {
                                PrintJob printJob = (PrintJob) task.getCallable();
                                try {
                                    ThreadPoolJobManager.this.jobQueue
                                            .start(printJob.getEntry().getReferenceId());
                                } catch (RuntimeException e) {
                                    LOGGER.error("failed to mark job as running", e);
                                } catch (NoSuchReferenceException e) {
                                    LOGGER.error("tried to mark non-existing job as 'running': {}",
                                                 printJob.getEntry().getReferenceId(), e);
                                }
                            }
                        }
                        super.beforeExecute(t, runnable);
                    }
                };

        this.timer = Executors.newScheduledThreadPool(1, timerTask -> {
            final Thread thread = new Thread(timerTask, "Post result to registry");
            thread.setDaemon(true);
            return thread;
        });
        this.timer.scheduleAtFixedRate(new RegistryTask(), RegistryTask.CHECK_INTERVAL,
                                       RegistryTask.CHECK_INTERVAL, TimeUnit.MILLISECONDS);

        if (this.oldFileCleanUp) {
            this.cleanUpTimer = Executors.newScheduledThreadPool(1, timerTask -> {
                final Thread thread = new Thread(timerTask, "Clean up old files");
                thread.setDaemon(true);
                return thread;
            });
            this.cleanUpTimer.scheduleAtFixedRate(
                    this.workingDirectories.getCleanUpTask(), 0,
                    this.oldFileCleanupInterval, TimeUnit.SECONDS);
        }
    }
403766
private static Runnable asRunnable(final ClientTransport.PingCallback callback,
                                     final Throwable failureCause) {
    return new Runnable() {
      @Override
      public void run() {
        callback.onFailure(failureCause);
      }
    };
  }