### Question:
ClassPathJobGraphRetriever implements JobGraphRetriever { @Override public JobGraph retrieveJobGraph(Configuration configuration) throws FlinkException { final PackagedProgram packagedProgram = createPackagedProgram(); final int defaultParallelism = configuration.getInteger(CoreOptions.DEFAULT_PARALLELISM); try { final JobGraph jobGraph = PackagedProgramUtils.createJobGraph( packagedProgram, configuration, defaultParallelism, jobId); jobGraph.setAllowQueuedScheduling(true); jobGraph.setSavepointRestoreSettings(savepointRestoreSettings); return jobGraph; } catch (Exception e) { throw new FlinkException("Could not create the JobGraph from the provided user code jar.", e); } } ClassPathJobGraphRetriever(
			@Nonnull JobID jobId,
			@Nonnull SavepointRestoreSettings savepointRestoreSettings,
			@Nonnull String[] programArguments,
			@Nullable String jobClassName); @VisibleForTesting  ClassPathJobGraphRetriever(
			@Nonnull JobID jobId,
			@Nonnull SavepointRestoreSettings savepointRestoreSettings,
			@Nonnull String[] programArguments,
			@Nullable String jobClassName,
			@Nonnull Supplier<Iterable<File>> jarsOnClassPath); @Override JobGraph retrieveJobGraph(Configuration configuration);  }### Answer:
@Test public void testJobGraphRetrieval() throws FlinkException { final int parallelism = 42; final Configuration configuration = new Configuration(); configuration.setInteger(CoreOptions.DEFAULT_PARALLELISM, parallelism); final JobID jobId = new JobID(); final ClassPathJobGraphRetriever classPathJobGraphRetriever = new ClassPathJobGraphRetriever( jobId, SavepointRestoreSettings.none(), PROGRAM_ARGUMENTS, TestJob.class.getCanonicalName()); final JobGraph jobGraph = classPathJobGraphRetriever.retrieveJobGraph(configuration); assertThat(jobGraph.getName(), is(equalTo(TestJob.class.getCanonicalName() + "-suffix"))); assertThat(jobGraph.getMaximumParallelism(), is(parallelism)); assertEquals(jobGraph.getJobID(), jobId); }
@Test public void testJobGraphRetrievalFromJar() throws FlinkException, FileNotFoundException { final File testJar = TestJob.getTestJobJar(); final ClassPathJobGraphRetriever classPathJobGraphRetriever = new ClassPathJobGraphRetriever( new JobID(), SavepointRestoreSettings.none(), PROGRAM_ARGUMENTS, null, () -> Collections.singleton(testJar)); final JobGraph jobGraph = classPathJobGraphRetriever.retrieveJobGraph(new Configuration()); assertThat(jobGraph.getName(), is(equalTo(TestJob.class.getCanonicalName() + "-suffix"))); }
@Test public void testJobGraphRetrievalJobClassNameHasPrecedenceOverClassPath() throws FlinkException, FileNotFoundException { final File testJar = new File("non-existing"); final ClassPathJobGraphRetriever classPathJobGraphRetriever = new ClassPathJobGraphRetriever( new JobID(), SavepointRestoreSettings.none(), PROGRAM_ARGUMENTS, TestJob.class.getCanonicalName(), () -> Collections.singleton(testJar)); final JobGraph jobGraph = classPathJobGraphRetriever.retrieveJobGraph(new Configuration()); assertThat(jobGraph.getName(), is(equalTo(TestJob.class.getCanonicalName() + "-suffix"))); }
@Test public void testSavepointRestoreSettings() throws FlinkException { final Configuration configuration = new Configuration(); final SavepointRestoreSettings savepointRestoreSettings = SavepointRestoreSettings.forPath("foobar", true); final JobID jobId = new JobID(); final ClassPathJobGraphRetriever classPathJobGraphRetriever = new ClassPathJobGraphRetriever( jobId, savepointRestoreSettings, PROGRAM_ARGUMENTS, TestJob.class.getCanonicalName()); final JobGraph jobGraph = classPathJobGraphRetriever.retrieveJobGraph(configuration); assertThat(jobGraph.getSavepointRestoreSettings(), is(equalTo(savepointRestoreSettings))); assertEquals(jobGraph.getJobID(), jobId); }