code: @Override public JobSpecSchedule scheduleJob(JobSpec jobSpec, Runnable jobRunnable) {
      _log.info("Scheduling JobSpec " + jobSpec);
      final URI jobSpecURI = jobSpec.getUri();
  
      JobSpecSchedule newSchedule = null;
      Runnable runnableWithTriggerCallback = new TriggerRunnable(jobSpec, jobRunnable);
      synchronized (this) {
        JobSpecSchedule existingSchedule = _schedules.get(jobSpecURI);
        if (null != existingSchedule) {
          if (existingSchedule.getJobSpec().equals(jobSpec)) {
            _log.warn("Ignoring already scheduled job: " + jobSpec);
            return existingSchedule;
          }
  
          // a new job spec -- unschedule first so that we schedule the new version
          unscheduleJob(jobSpecURI);
        }
  
        newSchedule = doScheduleJob(jobSpec, runnableWithTriggerCallback);
        _schedules.put(jobSpecURI, newSchedule);
      }
      _callbacksDispatcher.onJobScheduled(newSchedule);
  
      return newSchedule;
    }
docs: {@inheritDoc}