class JobControllerService {

  List<JobTask> makeJobUpdateTasks(Iterable<Pair<Source, Set<Store>>> sourceToStores) {
    List<JobTask> jobTasks = new LinkedList<>();
    // Ensure a running job for each source to store mapping
    List<Job> activeJobs = new LinkedList<>();
    boolean isSafeToStopJobs = true;

    for (Pair<Source, Set<Store>> mapping : sourceToStores) {
      Source source = mapping.getKey();
      Set<Store> stores = mapping.getValue();

      Job job = groupingStrategy.getOrCreateJob(source, stores, this.jobLabels);

      if (job.isDeployed()) {
        if (!job.isRunning()) {
          jobTasks.add(new UpdateJobStatusTask(job, jobManager));

          // Mark that it is not safe to stop jobs without disrupting ingestion
          isSafeToStopJobs = false;
          continue;
        }

        if (jobRequiresUpgrade(job, stores) && job.isRunning()) {
          // Since we want to upgrade job without downtime
          // it would make sense to spawn clone of current job
          // and terminate old version on the next Poll.
          // Both jobs should be in the same consumer group and not conflict with each other
          job = job.cloneWithIdAndLabels(groupingStrategy.createJobId(job), this.jobLabels);
          job.addAllStores(stores);

          isSafeToStopJobs = false;

          jobTasks.add(new CreateJobTask(job, jobManager));
        } else {
          jobTasks.add(new UpdateJobStatusTask(job, jobManager));
        }
      } else {
        job.addAllFeatureSets(
            stores.stream()
                .flatMap(s -> getFeatureSetsForStore(s).stream())
                .filter(fs -> fs.getSpec().getSource().equals(source))
                .collect(Collectors.toSet()));

        jobTasks.add(new CreateJobTask(job, jobManager));
      }

      // Record the job as required to safeguard it from getting stopped
      activeJobs.add(job);
    }
    // Stop extra jobs that are not required to maintain ingestion when safe
    if (isSafeToStopJobs) {
      getExtraJobs(activeJobs)
          .forEach(
              extraJob -> {
                jobTasks.add(new TerminateJobTask(extraJob, jobManager));
              });
    }

    return jobTasks;
  }

  @Autowired public  JobControllerService(
      JobRepository jobRepository,
      CoreServiceGrpc.CoreServiceBlockingStub specService,
      JobManager jobManager,
      FeastProperties feastProperties,
      JobGroupingStrategy groupingStrategy,
      KafkaTemplate<String, FeatureSetSpec> specPublisher);

}