@Test public void sendsStopToCorrectChild() { String job1Node = startJob(createLaunchMessage(1)); String job2Node = startJob(createLaunchMessage(2)); String job3Node = startJob(createLaunchMessage(3)); verifyStopSentToNode(2, job2Node); verifyStopSentToNode(3, job3Node); verifyStopSentToNode(1, job1Node); }
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();  }