Client { public JobSubmissionResult runDetached(PackagedProgram prog, int parallelism) throws ProgramInvocationException { Thread.currentThread().setContextClassLoader(prog.getUserCodeClassLoader()); if (prog.isUsingProgramEntryPoint()) { return runDetached(prog.getPlanWithJars(), parallelism, prog.getSavepointPath()); } else if (prog.isUsingInteractiveMode()) { LOG.info("Starting program in interactive mode"); ContextEnvironmentFactory factory = new ContextEnvironmentFactory(this, prog.getAllLibraries(), prog.getClasspaths(), prog.getUserCodeClassLoader(), parallelism, false, prog.getSavepointPath()); ContextEnvironment.setAsContext(factory); try { prog.invokeInteractiveModeForExecution(); return ((DetachedEnvironment) factory.getLastEnvCreated()).finalizeExecute(); } finally { ContextEnvironment.unsetContext(); } } else { throw new RuntimeException("PackagedProgram does not have a valid invocation mode."); } } 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 testDetachedMode() throws Exception{ jobManagerSystem.actorOf(Props.create(SuccessReturningActor.class), JobManager.JOB_MANAGER_NAME()); Client out = new Client(config); try { PackagedProgram prg = new PackagedProgram(TestExecuteTwice.class); out.runDetached(prg, 1); fail(FAIL_MESSAGE); } catch (ProgramInvocationException e) { assertEquals( DetachedJobExecutionResult.DETACHED_MESSAGE + DetachedJobExecutionResult.EXECUTE_TWICE_MESSAGE, e.getCause().getMessage()); } try { PackagedProgram prg = new PackagedProgram(TestEager.class); out.runDetached(prg, 1); fail(FAIL_MESSAGE); } catch (ProgramInvocationException e) { assertEquals( DetachedJobExecutionResult.DETACHED_MESSAGE + DetachedJobExecutionResult.JOB_RESULT_MESSAGE + DetachedJobExecutionResult.EAGER_FUNCTION_MESSAGE, e.getCause().getMessage()); } try { PackagedProgram prg = new PackagedProgram(TestGetRuntime.class); out.runDetached(prg, 1); fail(FAIL_MESSAGE); } catch (ProgramInvocationException e) { assertEquals( DetachedJobExecutionResult.DETACHED_MESSAGE + DetachedJobExecutionResult.JOB_RESULT_MESSAGE, e.getCause().getMessage()); } try { PackagedProgram prg = new PackagedProgram(TestGetJobID.class); out.runDetached(prg, 1); fail(FAIL_MESSAGE); } catch (ProgramInvocationException e) { assertEquals( DetachedJobExecutionResult.DETACHED_MESSAGE + DetachedJobExecutionResult.JOB_RESULT_MESSAGE, e.getCause().getMessage()); } try { PackagedProgram prg = new PackagedProgram(TestGetAccumulator.class); out.runDetached(prg, 1); fail(FAIL_MESSAGE); } catch (ProgramInvocationException e) { assertEquals( DetachedJobExecutionResult.DETACHED_MESSAGE + DetachedJobExecutionResult.JOB_RESULT_MESSAGE + DetachedJobExecutionResult.EAGER_FUNCTION_MESSAGE, e.getCause().getMessage()); } try { PackagedProgram prg = new PackagedProgram(TestGetAllAccumulator.class); out.runDetached(prg, 1); fail(FAIL_MESSAGE); } catch (ProgramInvocationException e) { assertEquals( DetachedJobExecutionResult.DETACHED_MESSAGE + DetachedJobExecutionResult.JOB_RESULT_MESSAGE, e.getCause().getMessage()); } }
@Test public void shouldSubmitToJobClient() { try { jobManagerSystem.actorOf(Props.create(SuccessReturningActor.class), JobManager.JOB_MANAGER_NAME()); Client out = new Client(config); JobSubmissionResult result = out.runDetached(program.getPlanWithJars(), 1); assertNotNull(result); program.deleteExtractedLibraries(); } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }
@Test public void shouldSubmitToJobClientFails() { try { jobManagerSystem.actorOf(Props.create(FailureReturningActor.class), JobManager.JOB_MANAGER_NAME()); Client out = new Client(config); try { out.runDetached(program.getPlanWithJars(), 1); fail("This should fail with an exception"); } catch (ProgramInvocationException e) { } catch (Exception e) { fail("wrong exception " + e); } } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }