code: private <T> void deployJob(ExecutionContext<T> context, JobGraph jobGraph, Result<T> result) {
  		// create or retrieve cluster and deploy job
  		try (final ClusterDescriptor<T> clusterDescriptor = context.createClusterDescriptor()) {
  			try {
  				// new cluster
  				if (context.getClusterId() == null) {
  					deployJobOnNewCluster(clusterDescriptor, jobGraph, result, context.getClassLoader());
  				}
  				// reuse existing cluster
  				else {
  					deployJobOnExistingCluster(context.getClusterId(), clusterDescriptor, jobGraph, result);
  				}
  			} catch (Exception e) {
  				throw new SqlExecutionException("Could not retrieve or create a cluster.", e);
  			}
  		} catch (SqlExecutionException e) {
  			throw e;
  		} catch (Exception e) {
  			throw new SqlExecutionException("Could not locate a cluster.", e);
  		}
  	}
docs: Deploys a job. Depending on the deployment creates a new job cluster. It saves the cluster id in
  the result and blocks until job completion.