Client { public JobSubmissionResult runBlocking(PackagedProgram prog, int parallelism) throws ProgramInvocationException { Thread.currentThread().setContextClassLoader(prog.getUserCodeClassLoader()); if (prog.isUsingProgramEntryPoint()) { return runBlocking(prog.getPlanWithJars(), parallelism, prog.getSavepointPath()); } else if (prog.isUsingInteractiveMode()) { LOG.info("Starting program in interactive mode"); ContextEnvironment.setAsContext(new ContextEnvironmentFactory(this, prog.getAllLibraries(), prog.getClasspaths(), prog.getUserCodeClassLoader(), parallelism, true, prog.getSavepointPath())); try { prog.invokeInteractiveModeForExecution(); } finally { ContextEnvironment.unsetContext(); } return new JobSubmissionResult(lastJobID); } else { throw new RuntimeException(); } } Client(Configuration config);  Client(Configuration config, int maxSlots); void shutdown(); void setPrintStatusDuringExecution(boolean print); boolean getPrintStatusDuringExecution(); int getMaxSlots(); static String getOptimizedPlanAsJson(Optimizer compiler, PackagedProgram prog, int parallelism); static FlinkPlan getOptimizedPlan(Optimizer compiler, PackagedProgram prog, int parallelism); static OptimizedPlan getOptimizedPlan(Optimizer compiler, Plan p, int parallelism); JobSubmissionResult runBlocking(PackagedProgram prog, int parallelism); JobSubmissionResult runDetached(PackagedProgram prog, int parallelism); JobExecutionResult runBlocking(JobWithJars program, int parallelism); JobExecutionResult runBlocking(JobWithJars program, int parallelism, String savepointPath); JobSubmissionResult runDetached(JobWithJars program, int parallelism); JobSubmissionResult runDetached(JobWithJars program, int parallelism, String savepointPath); JobExecutionResult runBlocking(
			FlinkPlan compiledPlan, List<URL> libraries, List<URL> classpaths, ClassLoader classLoader); JobExecutionResult runBlocking(FlinkPlan compiledPlan, List<URL> libraries, List<URL> classpaths,
			ClassLoader classLoader, String savepointPath); JobSubmissionResult runDetached(FlinkPlan compiledPlan, List<URL> libraries, List<URL> classpaths, ClassLoader classLoader); JobSubmissionResult runDetached(FlinkPlan compiledPlan, List<URL> libraries, List<URL> classpaths,
			ClassLoader classLoader, String savepointPath); JobExecutionResult runBlocking(JobGraph jobGraph, ClassLoader classLoader); JobSubmissionResult runDetached(JobGraph jobGraph, ClassLoader classLoader); void cancel(JobID jobId); void stop(final JobID jobId); Map<String, Object> getAccumulators(JobID jobID); Map<String, Object> getAccumulators(JobID jobID, ClassLoader loader); void endSession(JobID jobId); void endSessions(List<JobID> jobIds); JobGraph getJobGraph(PackagedProgram prog, FlinkPlan optPlan); JobGraph getJobGraph(PackagedProgram prog, FlinkPlan optPlan, String savepointPath);  }
@Test public void tryLocalExecution() { try { jobManagerSystem.actorOf(Props.create(SuccessReturningActor.class), JobManager.JOB_MANAGER_NAME()); PackagedProgram packagedProgramMock = mock(PackagedProgram.class); when(packagedProgramMock.isUsingInteractiveMode()).thenReturn(true); doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { ExecutionEnvironment.createLocalEnvironment(); return null; } }).when(packagedProgramMock).invokeInteractiveModeForExecution(); try { new Client(config).runBlocking(packagedProgramMock, 1); fail("Creating the local execution environment should not be possible"); } catch (InvalidProgramException e) { } } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }