@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); }
@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 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 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); }
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); }
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);  }