docs: @Override
  	public JobExecutionResult executePlan(Plan plan) throws Exception {
  		if (plan == null) {
  			throw new IllegalArgumentException("The plan may not be null.");
  		}
  
  		synchronized (this.lock) {
  
  			// check if we start a session dedicated for this execution
  			final boolean shutDownAtEnd;
  
  			if (jobExecutorService == null) {
  				shutDownAtEnd = true;
  
  				// configure the number of local slots equal to the parallelism of the local plan
  				if (this.taskManagerNumSlots == DEFAULT_TASK_MANAGER_NUM_SLOTS) {
  					int maxParallelism = plan.getMaximumParallelism();
  					if (maxParallelism > 0) {
  						this.taskManagerNumSlots = maxParallelism;
  					}
  				}
  
  				// start the cluster for us
  				start();
  			}
  			else {
  				// we use the existing session
  				shutDownAtEnd = false;
  			}
  
  			try {
  				// TODO: Set job's default parallelism to max number of slots
  				final int slotsPerTaskManager = jobExecutorServiceConfiguration.getInteger(TaskManagerOptions.NUM_TASK_SLOTS, taskManagerNumSlots);
  				final int numTaskManagers = jobExecutorServiceConfiguration.getInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1);
  				plan.setDefaultParallelism(slotsPerTaskManager * numTaskManagers);
  
  				Optimizer pc = new Optimizer(new DataStatistics(), jobExecutorServiceConfiguration);
  				OptimizedPlan op = pc.compile(plan);
  
  				JobGraphGenerator jgg = new JobGraphGenerator(jobExecutorServiceConfiguration);
  				JobGraph jobGraph = jgg.compileJobGraph(op, plan.getJobId());
  
  				return jobExecutorService.executeJobBlocking(jobGraph);
  			}
  			finally {
  				if (shutDownAtEnd) {
  					stop();
  				}
  			}
  		}
  	}