RemoteJobPlanner extends JobPlanner { @Override public List<JobConfig> prepareJobs() { String runId = String.valueOf(System.currentTimeMillis()) + "-" + UUID.randomUUID().toString().substring(0, 8); LOG.info("The run id for this run is {}", runId); ExecutionPlan plan = getExecutionPlan(runId); try { writePlanJsonFile(plan.getPlanAsJson()); } catch (Exception e) { throw new SamzaException("Failed to create plan JSON.", e); } List<JobConfig> jobConfigs = plan.getJobConfigs(); if (jobConfigs.isEmpty()) { throw new SamzaException("No jobs in the plan."); } JobConfig jobConfig = jobConfigs.get(0); StreamManager streamManager = null; try { streamManager = buildAndStartStreamManager(jobConfig); if (plan.getApplicationConfig().getAppMode() == ApplicationConfig.ApplicationMode.BATCH) { streamManager.clearStreamsFromPreviousRun(getConfigFromPrevRun()); } streamManager.createStreams(plan.getIntermediateStreams()); } finally { if (streamManager != null) { streamManager.stop(); } } return jobConfigs; } RemoteJobPlanner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> descriptor); @Override List<JobConfig> prepareJobs();  }
@Test public void testStreamCreation() throws Exception { remotePlanner = createRemoteJobPlanner(mock(StreamApplicationDescriptorImpl.class)); StreamManager streamManager = mock(StreamManager.class); doReturn(streamManager).when(remotePlanner).buildAndStartStreamManager(any(Config.class)); ExecutionPlan plan = mock(ExecutionPlan.class); when(plan.getIntermediateStreams()).thenReturn( Collections.singletonList(new StreamSpec("test-stream", "test-stream", "test-system"))); when(plan.getPlanAsJson()).thenReturn(""); when(plan.getJobConfigs()).thenReturn(Collections.singletonList(mock(JobConfig.class))); ApplicationConfig mockAppConfig = mock(ApplicationConfig.class); when(mockAppConfig.getAppMode()).thenReturn(ApplicationConfig.ApplicationMode.STREAM); when(plan.getApplicationConfig()).thenReturn(mockAppConfig); doReturn(plan).when(remotePlanner).getExecutionPlan(any()); remotePlanner.prepareJobs(); verify(streamManager, times(0)).clearStreamsFromPreviousRun(any()); ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class); verify(streamManager).createStreams(captor.capture()); List<StreamSpec> streamSpecs = captor.getValue(); assertEquals(streamSpecs.size(), 1); assertEquals(streamSpecs.get(0).getId(), "test-stream"); verify(streamManager).stop(); }