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 ) ; } }
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 .