@Test public void testDeployerWithIsolatedConfiguration() throws Exception { YarnClusterConfiguration clusterConf = mock(YarnClusterConfiguration.class); doReturn(new YarnConfiguration()).when(clusterConf).conf(); ScheduledExecutorService executor = mock(ScheduledExecutorService.class); Configuration flinkConf = new Configuration(); YarnClient client = mock(YarnClient.class); JobDeployer deploy = new JobDeployer(clusterConf, client, executor, flinkConf); AthenaXYarnClusterDescriptor desc = mock(AthenaXYarnClusterDescriptor.class); YarnClusterClient clusterClient = mock(YarnClusterClient.class); doReturn(clusterClient).when(desc).deploy(); ActorGateway actorGateway = mock(ActorGateway.class); doReturn(actorGateway).when(clusterClient).getJobManagerGateway(); doReturn(Future$.MODULE$.successful(null)).when(actorGateway).ask(any(), any()); JobGraph jobGraph = mock(JobGraph.class); doReturn(JobID.generate()).when(jobGraph).getJobID(); deploy.start(desc, jobGraph); verify(clusterClient).runDetached(jobGraph, null); }
void start(JobGraph job, JobConf desc) throws Exception { AthenaXYarnClusterDescriptor descriptor = new AthenaXYarnClusterDescriptor(clusterConf, yarnClient, flinkConf, desc); start(descriptor, job); }
JobDeployer { void start(JobGraph job, JobConf desc) throws Exception { AthenaXYarnClusterDescriptor descriptor = new AthenaXYarnClusterDescriptor(clusterConf, yarnClient, flinkConf, desc); start(descriptor, job); } }
JobDeployer { void start(JobGraph job, JobConf desc) throws Exception { AthenaXYarnClusterDescriptor descriptor = new AthenaXYarnClusterDescriptor(clusterConf, yarnClient, flinkConf, desc); start(descriptor, job); } JobDeployer(YarnClusterConfiguration clusterConf, YarnClient yarnClient,
              ScheduledExecutorService executor, Configuration flinkConf); }
JobDeployer { void start(JobGraph job, JobConf desc) throws Exception { AthenaXYarnClusterDescriptor descriptor = new AthenaXYarnClusterDescriptor(clusterConf, yarnClient, flinkConf, desc); start(descriptor, job); } JobDeployer(YarnClusterConfiguration clusterConf, YarnClient yarnClient,
              ScheduledExecutorService executor, Configuration flinkConf);  }
JobDeployer { void start(JobGraph job, JobConf desc) throws Exception { AthenaXYarnClusterDescriptor descriptor = new AthenaXYarnClusterDescriptor(clusterConf, yarnClient, flinkConf, desc); start(descriptor, job); } JobDeployer(YarnClusterConfiguration clusterConf, YarnClient yarnClient,
              ScheduledExecutorService executor, Configuration flinkConf);   }