code: @Override
  	public JobExecutionResult execute(String jobName) throws Exception {
  		// transform the streaming program into a JobGraph
  		StreamGraph streamGraph = getStreamGraph();
  		streamGraph.setJobName(jobName);
  
  		JobGraph jobGraph = streamGraph.getJobGraph();
  		jobGraph.setAllowQueuedScheduling(true);
  
  		Configuration configuration = new Configuration();
  		configuration.addAll(jobGraph.getJobConfiguration());
  		configuration.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "0");
  
  		// add (and override) the settings with what the user defined
  		configuration.addAll(this.configuration);
  
  		if (!configuration.contains(RestOptions.BIND_PORT)) {
  			configuration.setString(RestOptions.BIND_PORT, "0");
  		}
  
  		int numSlotsPerTaskManager = configuration.getInteger(TaskManagerOptions.NUM_TASK_SLOTS, jobGraph.getMaximumParallelism());
  
  		MiniClusterConfiguration cfg = new MiniClusterConfiguration.Builder()
  			.setConfiguration(configuration)
  			.setNumSlotsPerTaskManager(numSlotsPerTaskManager)
  			.build();
  
  		if (LOG.isInfoEnabled()) {
  			LOG.info("Running job on local embedded Flink mini cluster");
  		}
  
  		MiniCluster miniCluster = new MiniCluster(cfg);
  
  		try {
  			miniCluster.start();
  			configuration.setInteger(RestOptions.PORT, miniCluster.getRestAddress().get().getPort());
  
  			return miniCluster.executeJobBlocking(jobGraph);
  		}
  		finally {
  			transformations.clear();
  			miniCluster.close();
  		}
  	}
docs: Executes the JobGraph of the on a mini cluster of CLusterUtil with a user
  specified name.
  
  @param jobName
  name of the job
  @return The result of the job execution, containing elapsed time and accumulators.