Client { public static FlinkPlan getOptimizedPlan(Optimizer compiler, PackagedProgram prog, int parallelism) throws CompilerException, ProgramInvocationException { Thread.currentThread().setContextClassLoader(prog.getUserCodeClassLoader()); if (prog.isUsingProgramEntryPoint()) { return getOptimizedPlan(compiler, prog.getPlanWithJars(), parallelism); } else if (prog.isUsingInteractiveMode()) { OptimizerPlanEnvironment env = new OptimizerPlanEnvironment(compiler); if (parallelism > 0) { env.setParallelism(parallelism); } return env.getOptimizedPlan(prog); } else { throw new RuntimeException("Couldn't determine program 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 testGetExecutionPlan() { try { jobManagerSystem.actorOf(Props.create(FailureReturningActor.class), JobManager.JOB_MANAGER_NAME()); PackagedProgram prg = new PackagedProgram(TestOptimizerPlan.class, "/dev/random", "/tmp"); assertNotNull(prg.getPreviewPlan()); Optimizer optimizer = new Optimizer(new DataStatistics(), new DefaultCostEstimator(), config); OptimizedPlan op = (OptimizedPlan) Client.getOptimizedPlan(optimizer, prg, 1); assertNotNull(op); PlanJSONDumpGenerator dumper = new PlanJSONDumpGenerator(); assertNotNull(dumper.getOptimizerPlanAsJSON(op)); PlanJSONDumpGenerator dumper2 = new PlanJSONDumpGenerator(); dumper2.setEncodeForHTML(true); String htmlEscaped = dumper2.getOptimizerPlanAsJSON(op); assertEquals(-1, htmlEscaped.indexOf('\\')); } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }