code: public void rescheduleJob(final JobInstance toBeRescheduled) {
          final JobIdentifier jobIdentifier = toBeRescheduled.getIdentifier();
          synchronized (mutexForStageInstance(jobIdentifier)) {
              synchronized (mutexForJob(jobIdentifier)) {
                  transactionTemplate.execute(new TransactionCallbackWithoutResult() {
                      @Override
                      protected void doInTransactionWithoutResult(TransactionStatus status) {
                          LOGGER.warn("[Job Reschedule] Rescheduling and marking old job as ignored: {}", toBeRescheduled);
                          //Reloading it because we want to see the latest committed state after acquiring the mutex.
                          JobInstance oldJob = jobInstanceService.buildById(toBeRescheduled.getId());
                          if (oldJob.isCompleted() || oldJob.isRescheduled()) {
                              return;
                          }
                          JobInstance newJob = oldJob.clone();
                          oldJob.changeState(JobState.Rescheduled);
                          jobInstanceService.updateStateAndResult(oldJob);
  
                          //Make a new Job
                          newJob.reschedule();
                          jobInstanceService.save(oldJob.getIdentifier().getStageIdentifier(), oldJob.getStageId(), newJob);
  
                          //Copy the plan for the old job since we don't load job plan with jobInstance by default
                          JobPlan plan = jobInstanceDao.loadPlan(oldJob.getId());
                          jobInstanceDao.ignore(oldJob);
                          jobInstanceDao.save(newJob.getId(), plan);
                          LOGGER.info("[Job Reschedule] Scheduled new job: {}. Replacing old job: {}", newJob.getIdentifier(), oldJob.getIdentifier());
                      }
                  });
              }
          }
      }
docs: synchronized for updating job