docs: public static JobGraph createJobGraph(
  			PackagedProgram packagedProgram,
  			Configuration configuration,
  			int defaultParallelism,
  			@Nullable JobID jobID) throws ProgramInvocationException {
  		Thread.currentThread().setContextClassLoader(packagedProgram.getUserCodeClassLoader());
  		final Optimizer optimizer = new Optimizer(new DataStatistics(), new DefaultCostEstimator(), configuration);
  		final FlinkPlan flinkPlan;
  
  		if (packagedProgram.isUsingProgramEntryPoint()) {
  
  			final JobWithJars jobWithJars = packagedProgram.getPlanWithJars();
  
  			final Plan plan = jobWithJars.getPlan();
  
  			if (plan.getDefaultParallelism() <= 0) {
  				plan.setDefaultParallelism(defaultParallelism);
  			}
  
  			flinkPlan = optimizer.compile(jobWithJars.getPlan());
  		} else if (packagedProgram.isUsingInteractiveMode()) {
  			final OptimizerPlanEnvironment optimizerPlanEnvironment = new OptimizerPlanEnvironment(optimizer);
  
  			optimizerPlanEnvironment.setParallelism(defaultParallelism);
  
  			flinkPlan = optimizerPlanEnvironment.getOptimizedPlan(packagedProgram);
  		} else {
  			throw new ProgramInvocationException("PackagedProgram does not have a valid invocation mode.");
  		}
  
  		final JobGraph jobGraph;
  
  		if (flinkPlan instanceof StreamingPlan) {
  			jobGraph = ((StreamingPlan) flinkPlan).getJobGraph(jobID);
  			jobGraph.setSavepointRestoreSettings(packagedProgram.getSavepointSettings());
  		} else {
  			final JobGraphGenerator jobGraphGenerator = new JobGraphGenerator(configuration);
  			jobGraph = jobGraphGenerator.compileJobGraph((OptimizedPlan) flinkPlan, jobID);
  		}
  
  		for (URL url : packagedProgram.getAllLibraries()) {
  			try {
  				jobGraph.addJar(new Path(url.toURI()));
  			} catch (URISyntaxException e) {
  				throw new ProgramInvocationException("Invalid URL for jar file: " + url + '.', jobGraph.getJobID(), e);
  			}
  		}
  
  		jobGraph.setClasspaths(packagedProgram.getClasspaths());
  
  		return jobGraph;
  	}