public static SegmentGenerationJobSpec getSegmentGenerationJobSpec(String jobSpecFilePath, String propertyFilePath,
      Map<String, Object> context, Map<String, String> environmentValues) {
    Properties properties = new Properties();
    if (propertyFilePath != null) {
      try {
        properties.load(FileUtils.openInputStream(new File(propertyFilePath)));
      } catch (IOException e) {
        throw new RuntimeException(
            String.format("Unable to read property file [%s] into properties.", propertyFilePath), e);
      }
    }
    Map<String, Object> propertiesMap = (Map) properties;
    if (environmentValues != null) {
      for (String propertyName: propertiesMap.keySet()) {
        if (environmentValues.get(propertyName) != null) {
          propertiesMap.put(propertyName, environmentValues.get(propertyName));
        }
      }
    }
    if (context != null) {
      propertiesMap.putAll(context);
    }
    String jobSpecTemplate;
    try {
      jobSpecTemplate = IOUtils.toString(new BufferedReader(new FileReader(jobSpecFilePath)));
    } catch (IOException e) {
      throw new RuntimeException(String.format("Unable to read ingestion job spec file [%s].", jobSpecFilePath), e);
    }
    String jobSpecStr;
    try {
      jobSpecStr = GroovyTemplateUtils.renderTemplate(jobSpecTemplate, propertiesMap);
    } catch (Exception e) {
      throw new RuntimeException(String
          .format("Unable to render templates on ingestion job spec template file - [%s] with propertiesMap - [%s].",
              jobSpecFilePath, Arrays.toString(propertiesMap.entrySet().toArray())), e);
    }

    String jobSpecFormat = (String) propertiesMap.getOrDefault(JOB_SPEC_FORMAT, YAML);
    if (jobSpecFormat.equals(JSON)) {
      try {
        return JsonUtils.stringToObject(jobSpecStr, SegmentGenerationJobSpec.class);
      } catch (IOException e) {
        throw new RuntimeException(String
            .format("Unable to parse job spec - [%s] to JSON with propertiesMap - [%s]", jobSpecFilePath,
                Arrays.toString(propertiesMap.entrySet().toArray())), e);
      }
    }

    return new Yaml().loadAs(jobSpecStr, SegmentGenerationJobSpec.class);
  }
@Test
  public void testIngestionJobLauncherWithTemplateAndPropertyFileAndValueOverride() {
    Map<String, Object> context = GroovyTemplateUtils.getTemplateContext(Arrays.asList("year=2020"));
    SegmentGenerationJobSpec spec = IngestionJobLauncher.getSegmentGenerationJobSpec(
        GroovyTemplateUtils.class.getClassLoader().getResource("ingestion_job_spec_template.yaml").getFile(),
        GroovyTemplateUtils.class.getClassLoader().getResource("job.config").getFile(), context, null);
    Assert.assertEquals(spec.getInputDirURI(), "file:///path/to/input/2020/06/07");
    Assert.assertEquals(spec.getOutputDirURI(), "file:///path/to/output/2020/06/07");
    Assert.assertEquals(spec.getSegmentCreationJobParallelism(), 100);
  }