code: @Override
    public synchronized void scheduleJob(Properties jobProps, JobListener jobListener) throws JobException {
      Map<String, Object> additionalJobDataMap = Maps.newHashMap();
      additionalJobDataMap.put(ServiceConfigKeys.GOBBLIN_SERVICE_FLOWSPEC,
          this.scheduledFlowSpecs.get(jobProps.getProperty(ConfigurationKeys.JOB_NAME_KEY)));
  
      try {
        scheduleJob(jobProps, jobListener, additionalJobDataMap, GobblinServiceJob.class);
      } catch (Exception e) {
        throw new JobException("Failed to schedule job " + jobProps.getProperty(ConfigurationKeys.JOB_NAME_KEY), e);
      }
    }
docs: Synchronize the job scheduling because the same flowSpec can be scheduled by different threads.