@Test public void doesNotStartSameJobMultipleTimes() { LaunchStreamingJobMessage launchMessage = createLaunchMessage(1); LaunchStreamingJobMessage dupLaunchMessage = createLaunchMessage(1); startJob(launchMessage); try { startJob(dupLaunchMessage); fail("Expected IllegalStateException"); } catch (IllegalStateException ignored) { } verify(_mockChannel, times(1)) .sendToChild(any(), any()); }
public void startJob(LaunchStreamingJobMessage launchMessage, Collection<String> runningNodes) { synchronized (_jobToNodeMapper) { String nodeWithMinJobs = _jobToNodeMapper.getNodeWithMinJobs(runningNodes); _jobToNodeMapper.addJob(launchMessage.jobId, nodeWithMinJobs); LOG.info("Sending LaunchStreamingJobMessage for job {} to node {}", launchMessage.jobId, nodeWithMinJobs); _channelNode.sendToChild(nodeWithMinJobs, launchMessage); } }
MasterStreamingJobManager { public void startJob(LaunchStreamingJobMessage launchMessage, Collection<String> runningNodes) { synchronized (_jobToNodeMapper) { String nodeWithMinJobs = _jobToNodeMapper.getNodeWithMinJobs(runningNodes); _jobToNodeMapper.addJob(launchMessage.jobId, nodeWithMinJobs); LOG.info("Sending LaunchStreamingJobMessage for job {} to node {}", launchMessage.jobId, nodeWithMinJobs); _channelNode.sendToChild(nodeWithMinJobs, launchMessage); } } }
MasterStreamingJobManager { public void startJob(LaunchStreamingJobMessage launchMessage, Collection<String> runningNodes) { synchronized (_jobToNodeMapper) { String nodeWithMinJobs = _jobToNodeMapper.getNodeWithMinJobs(runningNodes); _jobToNodeMapper.addJob(launchMessage.jobId, nodeWithMinJobs); LOG.info("Sending LaunchStreamingJobMessage for job {} to node {}", launchMessage.jobId, nodeWithMinJobs); _channelNode.sendToChild(nodeWithMinJobs, launchMessage); } } @Autowired  MasterStreamingJobManager(ChannelNode channelNode); }
MasterStreamingJobManager { public void startJob(LaunchStreamingJobMessage launchMessage, Collection<String> runningNodes) { synchronized (_jobToNodeMapper) { String nodeWithMinJobs = _jobToNodeMapper.getNodeWithMinJobs(runningNodes); _jobToNodeMapper.addJob(launchMessage.jobId, nodeWithMinJobs); LOG.info("Sending LaunchStreamingJobMessage for job {} to node {}", launchMessage.jobId, nodeWithMinJobs); _channelNode.sendToChild(nodeWithMinJobs, launchMessage); } } @Autowired  MasterStreamingJobManager(ChannelNode channelNode); void startJob(LaunchStreamingJobMessage launchMessage, Collection<String> runningNodes); void stopJob(StopStreamingJobMessage stopMessage); void streamingJobExited(StreamingJobExitedMessage message); void stopAllJobs(); }
MasterStreamingJobManager { public void startJob(LaunchStreamingJobMessage launchMessage, Collection<String> runningNodes) { synchronized (_jobToNodeMapper) { String nodeWithMinJobs = _jobToNodeMapper.getNodeWithMinJobs(runningNodes); _jobToNodeMapper.addJob(launchMessage.jobId, nodeWithMinJobs); LOG.info("Sending LaunchStreamingJobMessage for job {} to node {}", launchMessage.jobId, nodeWithMinJobs); _channelNode.sendToChild(nodeWithMinJobs, launchMessage); } } @Autowired  MasterStreamingJobManager(ChannelNode channelNode); void startJob(LaunchStreamingJobMessage launchMessage, Collection<String> runningNodes); void stopJob(StopStreamingJobMessage stopMessage); void streamingJobExited(StreamingJobExitedMessage message); void stopAllJobs();  }